"""tests/regression/test_state_done_consistency.py — task-2471+1 자동화 회귀. 회장 명령 (D): state/DONE 정합성 machine-readable JSON 박제. 5 check: 1. state.current_state == "DONE" 2. .done 마커 존재 3. .g3-fail 마커 부재 4. .done.escalated 부재 또는 archived 5. .done.acked 정합 (acked 있으면 .done도 있어야 함) ``cmd_verify_consistency`` 가 위 5 check 결과를 JSON으로 출력하고 exit code (0=PASS / 1=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_consistency", 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" archive_dir = events_dir / "archive" for d in (state_dir, evidence_dir, events_dir, audit_dir, archive_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, "archive_dir": archive_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": "MERGED", "merge_commit_sha": "abcd1234", "ci_checks": {}, "guard_sh_result": None, "qc_report_guard_result": None, "merge_timestamp": None, "exit_codes": {}, }, "human_approved": True, "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) -> argparse.Namespace: return argparse.Namespace(task_id=task_id) def test_consistency_pass_clean_done(isolated, capsys) -> None: task_id = "task-cons-001" _write_state(isolated["state_dir"], task_id, "DONE") (isolated["events_dir"] / f"{task_id}.done").write_text("{}") rc = isolated["mod"].cmd_verify_consistency(_ns(task_id)) out = capsys.readouterr().out rec = json.loads(out) assert rec["ok"] is True assert rec["current_state"] == "DONE" assert rec["checks"]["state_done"]["ok"] is True assert rec["checks"]["done_present"]["ok"] is True assert rec["checks"]["g3_fail_absent"]["ok"] is True assert rec["checks"]["escalated_resolved"]["ok"] is True assert rc == 0 def test_consistency_fail_state_not_done(isolated, capsys) -> None: task_id = "task-cons-002" _write_state(isolated["state_dir"], task_id, "MERGED") (isolated["events_dir"] / f"{task_id}.done").write_text("{}") rc = isolated["mod"].cmd_verify_consistency(_ns(task_id)) rec = json.loads(capsys.readouterr().out) assert rec["ok"] is False assert rec["checks"]["state_done"]["ok"] is False assert rc == 1 def test_consistency_fail_done_missing(isolated, capsys) -> None: task_id = "task-cons-003" _write_state(isolated["state_dir"], task_id, "DONE") rc = isolated["mod"].cmd_verify_consistency(_ns(task_id)) rec = json.loads(capsys.readouterr().out) assert rec["ok"] is False assert rec["checks"]["done_present"]["ok"] is False assert rc == 1 def test_consistency_fail_g3_fail_present(isolated, capsys) -> None: task_id = "task-cons-004" _write_state(isolated["state_dir"], task_id, "DONE") (isolated["events_dir"] / f"{task_id}.done").write_text("{}") (isolated["events_dir"] / f"{task_id}.g3-fail").write_text("{}") rc = isolated["mod"].cmd_verify_consistency(_ns(task_id)) rec = json.loads(capsys.readouterr().out) assert rec["ok"] is False assert rec["checks"]["g3_fail_absent"]["ok"] is False assert rc == 1 def test_consistency_fail_live_escalated_blocks(isolated, capsys) -> None: """live .done.escalated 가 archive 없이 잔존하면 fail.""" task_id = "task-cons-005" _write_state(isolated["state_dir"], task_id, "DONE") (isolated["events_dir"] / f"{task_id}.done").write_text("{}") (isolated["events_dir"] / f"{task_id}.done.escalated").write_text( json.dumps({"trigger": "x", "reason": "y"}) ) rc = isolated["mod"].cmd_verify_consistency(_ns(task_id)) rec = json.loads(capsys.readouterr().out) assert rec["ok"] is False assert rec["checks"]["escalated_resolved"]["ok"] is False assert rc == 1 def test_consistency_pass_with_archived_escalated(isolated, capsys) -> None: """live escalated 부재 + archive 존재 → escalated_resolved.ok=True.""" task_id = "task-cons-006" _write_state(isolated["state_dir"], task_id, "DONE") (isolated["events_dir"] / f"{task_id}.done").write_text("{}") # archive 존재 (live는 부재) archived = ( isolated["archive_dir"] / f"{task_id}.done.escalated.20260507T120000Z.test" ) archived.write_text( json.dumps({"trigger": "x", "reason": "test"}) ) rc = isolated["mod"].cmd_verify_consistency(_ns(task_id)) rec = json.loads(capsys.readouterr().out) assert rec["checks"]["escalated_resolved"]["ok"] is True assert rec["checks"]["escalated_resolved"]["archived_count"] == 1 assert rec["ok"] is True assert rc == 0 def test_consistency_acked_alignment_fail(isolated, capsys) -> None: """.done.acked 있는데 .done 부재 → acked_alignment fail.""" task_id = "task-cons-007" _write_state(isolated["state_dir"], task_id, "DONE") (isolated["events_dir"] / f"{task_id}.done.acked").write_text("{}") rc = isolated["mod"].cmd_verify_consistency(_ns(task_id)) rec = json.loads(capsys.readouterr().out) # done_present도 fail이지만 acked_alignment도 fail assert rec["checks"]["acked_alignment"]["ok"] is False assert rec["ok"] is False assert rc == 1 def test_consistency_emits_machine_readable_json_with_required_keys( isolated, capsys ) -> None: """JSON output은 11 완료 조건 검증에 필요한 핵심 키를 모두 포함.""" task_id = "task-cons-008" _write_state(isolated["state_dir"], task_id, "DONE") (isolated["events_dir"] / f"{task_id}.done").write_text("{}") isolated["mod"].cmd_verify_consistency(_ns(task_id)) rec = json.loads(capsys.readouterr().out) for k in ("ts", "task_id", "actor", "command", "ok", "checks", "current_state", "pr_number", "pr_state", "merge_commit_sha"): assert k in rec, f"missing required key: {k}" for sub in ("state_done", "done_present", "g3_fail_absent", "escalated_resolved", "acked_alignment"): assert sub in rec["checks"], f"missing check: {sub}"