"""tests/taskctl/test_lifecycle.py PR Lifecycle 정상 흐름 검증 (task-2467) 벨레스(개발6팀 테스터) 작성. 스바로그의 구현 완료 전 선작성(TDD). 케이스: 1. test_init_dispatch_ack_run_committed_chain 2. test_pr_open_records_pr_number_and_evidence 3. test_verify_evidence_file_created 4. test_done_requires_merged_state 5. test_done_creates_done_file """ from __future__ import annotations import json import os import subprocess from pathlib import Path import pytest WORKSPACE = Path("/home/jay/workspace/.worktrees/task-2467-dev6") TASKCTL = WORKSPACE / "scripts" / "taskctl.py" # --------------------------------------------------------------------------- # helpers # --------------------------------------------------------------------------- def _isolated_workspace(tmp_path: Path) -> dict: """test마다 WORKSPACE_ROOT를 격리된 tmp_path로 설정.""" env = {**os.environ} env["WORKSPACE_ROOT"] = str(tmp_path) (tmp_path / ".tasks" / "state").mkdir(parents=True, exist_ok=True) (tmp_path / ".tasks" / "evidence").mkdir(parents=True, exist_ok=True) (tmp_path / "memory" / "events").mkdir(parents=True, exist_ok=True) (tmp_path / "memory" / "orchestration-audit").mkdir(parents=True, exist_ok=True) return env def _run(args: list[str], env: dict) -> subprocess.CompletedProcess: return subprocess.run( ["python3", str(TASKCTL)] + args, capture_output=True, text=True, env=env, timeout=30, ) def _state(tmp_path: Path, task_id: str) -> dict: p = tmp_path / ".tasks" / "state" / f"{task_id}.json" assert p.exists(), f"state 파일 없음: {p}" return json.loads(p.read_text()) def _force_state_with_checksum(tmp_path: Path, task_id: str, target_state: str) -> None: """테스트 전용: state를 강제로 특정 상태로 설정하고 체크섬 재계산.""" import hashlib p = tmp_path / ".tasks" / "state" / f"{task_id}.json" state = json.loads(p.read_text()) state["current_state"] = target_state state.pop("_checksum", None) payload = {k: v for k, v in state.items() if k != "_checksum"} canon = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")) checksum = hashlib.sha256(canon.encode("utf-8")).hexdigest() state["_checksum"] = checksum p.write_text(json.dumps(state, ensure_ascii=False, indent=2)) # --------------------------------------------------------------------------- # 케이스 1: init → dispatch → ack → run → committed 정상 전이 체인 # --------------------------------------------------------------------------- def test_init_dispatch_ack_run_committed_chain(tmp_path): """정상 lifecycle: init → dispatch → ack → run → commit 체인. commit 명령이 구현되면 COMMITTED 진입 검증. 현재 MVP에서는 RUNNING까지만 확인. """ env = _isolated_workspace(tmp_path) task_id = "task-lifecycle-01" proc = _run(["init", task_id], env) assert proc.returncode == 0, f"init 실패: {proc.stderr}" assert _state(tmp_path, task_id)["current_state"] == "CREATED" proc = _run(["dispatch", task_id], env) assert proc.returncode == 0, f"dispatch 실패: {proc.stderr}" assert _state(tmp_path, task_id)["current_state"] in {"DISPATCHED", "WORKTREE_READY"} proc = _run(["ack", task_id], env) assert proc.returncode == 0, f"ack 실패: {proc.stderr}" assert _state(tmp_path, task_id)["current_state"] in {"ACKED", "WORKTREE_READY"} proc = _run(["run", task_id], env) assert proc.returncode == 0, f"run 실패: {proc.stderr}" assert _state(tmp_path, task_id)["current_state"] == "RUNNING" # commit 명령 시도 (신규 구현에서 COMMITTED 진입) proc_commit = _run(["commit", task_id], env) state_after = _state(tmp_path, task_id) if proc_commit.returncode == 0: # 신규 구현: COMMITTED 전이 확인 assert state_after["current_state"] in {"COMMITTED", "RUNNING"}, ( f"commit 후 예상 외 상태: {state_after['current_state']}" ) else: # MVP: commit 명령 미구현, RUNNING 유지 허용 assert state_after["current_state"] == "RUNNING", ( f"commit 실패 후 상태가 RUNNING이 아님: {state_after['current_state']}" ) # 전이 이력 검증 — init 이후 최소 3건 이상 기록 transitions = state_after["transitions"] assert len(transitions) >= 3, f"전이 이력 부족: {len(transitions)}건" # 각 전이에 ts/actor 필드 포함 여부 for t in transitions: if t.get("from") is not None: # init의 from=None 제외 assert "ts" in t, f"전이에 ts 누락: {t}" assert "actor" in t, f"전이에 actor 누락: {t}" # --------------------------------------------------------------------------- # 케이스 2: pr-open --pr N → pr_number 기록 + evidence/pr-open.json 생성 # --------------------------------------------------------------------------- def test_pr_open_records_pr_number_and_evidence(tmp_path): """pr-open --pr 100 → state.evidence.pr_number == 100, evidence/pr-open.json 생성. evidence 디렉토리 방식은 신규 구현에서 추가. MVP는 state.evidence dict 방식. """ env = _isolated_workspace(tmp_path) task_id = "task-lifecycle-02" for cmd in [["init", task_id], ["dispatch", task_id], ["ack", task_id], ["run", task_id]]: assert _run(cmd, env).returncode == 0 proc = _run(["pr-open", task_id, "--pr", "100"], env) assert proc.returncode == 0, f"pr-open 실패: {proc.stderr}" state = _state(tmp_path, task_id) assert state["current_state"] in {"PR_OPEN"}, ( f"pr-open 후 상태가 PR_OPEN이 아님: {state['current_state']}" ) # MVP: state.evidence.pr_number 검증 ev = state.get("evidence", {}) assert ev.get("pr_number") == 100, ( f"pr_number 미기록: {ev.get('pr_number')}" ) # 신규 구현: evidence/pr-open.json 파일 생성 여부 (soft check) pr_open_ev = tmp_path / ".tasks" / "evidence" / task_id / "pr-open.json" if pr_open_ev.exists(): ev_data = json.loads(pr_open_ev.read_text()) assert ev_data.get("pr_number") == 100, ( f"pr-open.json에 pr_number 미기록: {ev_data}" ) # evidence 필수 필드 검증 (명세 §5) for field in ("command", "exit_code", "timestamp"): assert field in ev_data or "ts" in ev_data, ( f"pr-open.json에 필수 필드 '{field}' 누락: {ev_data.keys()}" ) else: # MVP: evidence 파일 방식 미구현 — xfail pytest.xfail("evidence/pr-open.json 파일 방식 미구현 (스바로그 대기)") # --------------------------------------------------------------------------- # 케이스 3: verify 실행 후 evidence/verify.json 생성 # --------------------------------------------------------------------------- def test_verify_evidence_file_created(tmp_path): """verify 실행 후 evidence/verify.json 생성 (guard.sh 통과 여부와 무관). guard.sh가 없거나 실패해도 evidence 파일 자체는 기록되어야 함. """ env = _isolated_workspace(tmp_path) task_id = "task-lifecycle-03" for cmd in [["init", task_id], ["dispatch", task_id], ["ack", task_id], ["run", task_id], ["pr-open", task_id, "--pr", "103"]]: assert _run(cmd, env).returncode == 0 # verify 실행 (guard.sh 없는 환경에서도 실행 가능) _run(["verify", task_id], env) state = _state(tmp_path, task_id) # MVP: state.evidence에 guard_sh_result 기록 확인 ev = state.get("evidence", {}) assert "guard_sh_result" in ev, ( f"verify 후 guard_sh_result 미기록: {ev.keys()}" ) assert ev["guard_sh_result"] in {"PASS", "FAIL", "MISSING"}, ( f"guard_sh_result 값 이상: {ev['guard_sh_result']}" ) assert "exit_codes" in ev and "verify" in ev.get("exit_codes", {}), ( f"exit_codes.verify 미기록: {ev}" ) # 신규 구현: evidence/verify.json 파일 방식 (soft check) verify_ev = tmp_path / ".tasks" / "evidence" / task_id / "verify.json" if verify_ev.exists(): ev_data = json.loads(verify_ev.read_text()) assert "guard_sh_result" in ev_data or "guard_result" in ev_data, ( f"verify.json에 guard_sh_result 누락: {ev_data.keys()}" ) # verify.json 없으면 MVP 방식으로 통과 (xfail 아님 — MVP evidence dict로 충분) # --------------------------------------------------------------------------- # 케이스 4: DONE 시도 시 state != MERGED → 차단 # --------------------------------------------------------------------------- def test_done_requires_merged_state(tmp_path): """§3.9 (4) 보완: PR_OPEN/RUNNING 등 비-MERGED 상태에서 done → 차단.""" env = _isolated_workspace(tmp_path) task_id = "task-lifecycle-04" for cmd in [["init", task_id], ["dispatch", task_id], ["ack", task_id], ["run", task_id], ["pr-open", task_id, "--pr", "104"]]: assert _run(cmd, env).returncode == 0 # PR_OPEN 상태에서 done 시도 proc = _run(["done", task_id], env) assert proc.returncode != 0, "PR_OPEN 상태에서 done이 차단되어야 함" state = _state(tmp_path, task_id) assert state["current_state"] not in {"DONE"}, ( f"비-MERGED 상태에서 DONE 진입 발생: {state['current_state']}" ) # --------------------------------------------------------------------------- # 케이스 5: MERGED 상태에서 done → memory/events/.done 생성 # --------------------------------------------------------------------------- def test_done_creates_done_file(tmp_path): """MERGED 상태에서 done → memory/events/.done 생성. MERGED 상태를 강제로 만든 후 done 명령 실행. 신규 구현에서 done 명령이 MERGED를 검증하고 .done 파일을 생성해야 함. """ env = _isolated_workspace(tmp_path) task_id = "task-lifecycle-05" for cmd in [["init", task_id], ["dispatch", task_id], ["ack", task_id], ["run", task_id], ["pr-open", task_id, "--pr", "105"]]: assert _run(cmd, env).returncode == 0 # MERGED 상태로 강제 설정 (bypass 경로) _force_state_with_checksum(tmp_path, task_id, "MERGED") state = _state(tmp_path, task_id) assert state["current_state"] == "MERGED", "강제 MERGED 설정 실패" # done 명령 실행 proc = _run(["done", task_id], env) done_file = tmp_path / "memory" / "events" / f"{task_id}.done" if proc.returncode == 0: # 신규 구현: .done 파일 생성 확인 assert done_file.exists(), ( f"done 명령 성공했지만 .done 파일 없음: {done_file}" ) state = _state(tmp_path, task_id) assert state["current_state"] == "DONE", ( f"done 후 상태가 DONE이 아님: {state['current_state']}" ) else: # MVP: done 명령 미구현 또는 gh api 호출 실패 — xfail pytest.xfail( f"done 명령 미구현 또는 외부 의존성 실패 (스바로그 대기): {proc.stderr[:200]}" )