"""tests/regression/test_finish_task_routing.py — task-2471+1 자동화 회귀. 회장 명령 (A): ``finish-task`` 자동화 경로 — state machine 라우팅 검증. dry-run 모드에서 현재 state별 다음 액션 라우팅을 검증한다 (실제 gh / git 호출 불필요). 잘못된 routing / no_route 상태에서 fail-closed (exit 1) 검증. 검증 routing: - COMMITTED → pr-open - PR_OPEN → ci-check - CI_PENDING → gemini-evidence - GEMINI_PENDING → review-ready - REVIEW_READY → verify - VERIFIED → approve - HUMAN_APPROVED → merge - MERGED → done - DONE → halt (steps_taken=0, no action) - terminal non-DONE (FAILED/CANCELLED) → halt fail """ from __future__ import annotations import argparse import hashlib import importlib.util import json import sys from pathlib import Path from typing import Any import pytest WORKSPACE = Path(__file__).resolve().parents[2] def _load_taskctl(monkeypatch: pytest.MonkeyPatch, root: Path): monkeypatch.setenv("WORKSPACE_ROOT", str(root)) spec = importlib.util.spec_from_file_location( "taskctl_isolated_finish", str(WORKSPACE / "scripts" / "taskctl.py"), ) assert spec and spec.loader mod = importlib.util.module_from_spec(spec) sys.modules[spec.name] = mod spec.loader.exec_module(mod) return mod @pytest.fixture() def isolated(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): state_dir = tmp_path / ".tasks" / "state" evidence_dir = tmp_path / ".tasks" / "evidence" events_dir = tmp_path / "memory" / "events" audit_dir = tmp_path / "memory" / "orchestration-audit" for d in (state_dir, evidence_dir, events_dir, audit_dir): d.mkdir(parents=True, exist_ok=True) mod = _load_taskctl(monkeypatch, tmp_path) return { "tmp_path": tmp_path, "mod": mod, "state_dir": state_dir, "events_dir": events_dir, } def _write_state(state_dir: Path, task_id: str, current_state: str) -> None: payload: dict[str, Any] = { "task_id": task_id, "current_state": current_state, "transitions": [], "evidence": { "git_diff_sha": None, "changed_paths": [], "branch": "task/x-dev0", "pr_number": 1, "pr_state": None, "ci_checks": {}, "guard_sh_result": None, "qc_report_guard_result": None, "merge_timestamp": None, "exit_codes": {}, }, "human_approved": False, "bypass": {"used": False, "ts": None, "actor": None}, "admin_override": { "used": False, "ts": None, "actor": None, "reason": None, "audit_log_offset": None, }, } canonical = json.dumps( payload, ensure_ascii=False, sort_keys=True, separators=(",", ":") ) payload["_checksum"] = hashlib.sha256(canonical.encode("utf-8")).hexdigest() (state_dir / f"{task_id}.json").write_text( json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8" ) def _ns(task_id: str, **kw) -> argparse.Namespace: return argparse.Namespace( task_id=task_id, dry_run=kw.get("dry_run", True), max_steps=kw.get("max_steps", 12), ) @pytest.mark.parametrize( "src_state,expected_action", [ ("COMMITTED", "pr-open"), ("PR_OPEN", "ci-check"), ("CI_PENDING", "gemini-evidence"), ("GEMINI_PENDING", "review-ready"), ("REVIEW_READY", "verify"), ("VERIFIED", "approve"), ("HUMAN_APPROVED", "merge"), ("MERGED", "done"), ], ) def test_finish_task_dryrun_routes_to_correct_next_action( isolated, capsys, src_state: str, expected_action: str ) -> None: task_id = f"task-finish-{src_state.lower()}" _write_state(isolated["state_dir"], task_id, src_state) isolated["mod"].cmd_finish_task(_ns(task_id, dry_run=True)) out = capsys.readouterr().out # 출력은 ① verify-consistency JSON ② finish-task summary JSON (항상 후행). # 마지막 raw_decode 객체를 finish-task 결과로 채택. decoder = json.JSONDecoder() idx = 0 objs = [] while idx < len(out): s_strip = out[idx:].lstrip() if not s_strip: break idx += len(out[idx:]) - len(s_strip) obj, end = decoder.raw_decode(out[idx:]) objs.append(obj) idx += end assert objs, "no JSON output from cmd_finish_task" rec = objs[-1] assert rec.get("log"), f"empty log for state={src_state}: {rec}" first = rec["log"][0] assert first.get("dry_run") is True assert first["state"] == src_state assert first["action"] == expected_action def test_finish_task_done_already_short_circuits(isolated, capsys) -> None: """이미 DONE이면 추가 액션 없이 verify-consistency 후 종료.""" task_id = "task-finish-already-done" _write_state(isolated["state_dir"], task_id, "DONE") (isolated["events_dir"] / f"{task_id}.done").write_text("{}") isolated["mod"].cmd_finish_task(_ns(task_id, dry_run=False)) out = capsys.readouterr().out decoder = json.JSONDecoder() idx = 0 objs = [] while idx < len(out): if out[idx] != "{": idx += 1 continue try: obj, end = decoder.raw_decode(out[idx:]) except json.JSONDecodeError: idx += 1 continue objs.append(obj) idx += end assert len(objs) >= 1 final = objs[-1] assert final["task_id"] == task_id assert final["final_state"] == "DONE" def test_finish_task_terminal_failed_blocks(isolated, capsys) -> None: """FAILED 상태 → halt + exit 1.""" task_id = "task-finish-failed" _write_state(isolated["state_dir"], task_id, "FAILED") rc = isolated["mod"].cmd_finish_task(_ns(task_id, dry_run=False)) assert rc == 1 out = capsys.readouterr().out rec = json.loads(out) assert rec["ok"] is False assert rec["final_state"] == "FAILED" def test_finish_task_no_route_blocks(isolated, capsys) -> None: """라우팅 불가능한 state (예: BLOCKED) → halt + exit 1.""" task_id = "task-finish-blocked" _write_state(isolated["state_dir"], task_id, "BLOCKED") rc = isolated["mod"].cmd_finish_task(_ns(task_id, dry_run=False)) assert rc == 1 out = capsys.readouterr().out rec = json.loads(out) assert rec["ok"] is False assert "no automation route" in rec["reason"] def test_finish_task_max_steps_guard(isolated) -> None: """max_steps=0 가드는 정수로 강제되며 1 이상 (안전 가드).""" task_id = "task-finish-guard" _write_state(isolated["state_dir"], task_id, "COMMITTED") # max_steps=0 입력해도 내부에서 1로 강제 (max(1, ...)) rc = isolated["mod"].cmd_finish_task( _ns(task_id, dry_run=True, max_steps=0) ) # dry-run에서 1 step만 진행하고 break — exit 0 또는 1 가능 (state가 DONE 아님 → 0) # 단, 본 테스트의 핵심: 무한 루프 발생하지 않음 assert rc in (0, 1) def test_finish_task_state_missing_blocks(isolated) -> None: """state 파일 없으면 즉시 fail.""" task_id = "task-finish-no-state" with pytest.raises(SystemExit) as excinfo: isolated["mod"].cmd_finish_task(_ns(task_id, dry_run=False)) assert excinfo.value.code == 1