"""tests/regression/test_chairman_audit.py — task-2471 회귀 테스트. 헤임달이 task-2471 1차 hardening 의 일환으로 신설한 ``utils/audit_chairman_recovery.py`` 의 schema + atomic append 동작 검증. - ``AUDIT_JSONL_PATH`` 경로 상수가 ``memory/orchestration-audit/chairman-manual-recovery.jsonl`` 인지 확인. - ``RECOVERY_RECORD_KEYS`` schema 6 키 모두 정의. - ``append_recovery`` atomic append 동작 (디렉토리 자동 생성, JSONL 직렬화). - ``read_recoveries`` 파일 부재 시 graceful (빈 리스트). 헤임달(개발2팀 테스터) 작성. """ from __future__ import annotations import importlib.util import json import sys from pathlib import Path import pytest WORKSPACE = Path(__file__).resolve().parents[2] @pytest.fixture(scope="module") def acr(): """``utils/audit_chairman_recovery.py`` 절대 경로 로드.""" file_path = WORKSPACE / "utils" / "audit_chairman_recovery.py" if not file_path.exists(): pytest.skip( "utils/audit_chairman_recovery.py 미작성 — 헤임달 신규 모듈 대기" ) spec = importlib.util.spec_from_file_location( "audit_chairman_recovery_test_alias", str(file_path) ) assert spec is not None and spec.loader is not None mod = importlib.util.module_from_spec(spec) sys.modules[spec.name] = mod spec.loader.exec_module(mod) return mod # --------------------------------------------------------------------------- # 경로 상수 + schema # --------------------------------------------------------------------------- def test_audit_jsonl_path_constant(acr): """AUDIT_JSONL_PATH 가 정확한 상대 경로인지 검증.""" assert hasattr(acr, "AUDIT_JSONL_PATH") p = acr.AUDIT_JSONL_PATH assert isinstance(p, Path) # 경로 마지막 두 segment 가 (orchestration-audit, chairman-manual-recovery.jsonl) parts = p.parts assert "orchestration-audit" in parts assert parts[-1] == "chairman-manual-recovery.jsonl" assert "memory" in parts def test_recovery_record_keys_schema(acr): """6 schema 키 모두 정의.""" assert hasattr(acr, "RECOVERY_RECORD_KEYS") keys = set(acr.RECOVERY_RECORD_KEYS) expected = {"task_id", "ts", "from_state", "to_state", "reason", "evidence_paths"} missing = expected - keys assert not missing, f"누락된 schema 키: {missing}" # --------------------------------------------------------------------------- # append_recovery 기본 동작 # --------------------------------------------------------------------------- def test_append_recovery_creates_directory_and_file(acr, tmp_path): """디렉토리/파일 부재 시 자동 생성.""" out_path = acr.append_recovery( task_id="task-2471", from_state="RECOVERABLE_BLOCKED", to_state="MERGING", reason="branch protection 해제 확인", evidence_paths=[".tasks/evidence/task-2471/recovery.json"], workspace=tmp_path, ) assert out_path.exists() assert out_path.is_file() # 디렉토리 자동 생성 확인 assert out_path.parent.is_dir() def test_append_recovery_record_schema(acr, tmp_path): """append 된 JSONL 라인이 6 schema 키를 모두 포함.""" acr.append_recovery( task_id="task-2471", from_state="RECOVERABLE_BLOCKED", to_state="MERGING", reason="ok", evidence_paths=["a.json", "b.json"], workspace=tmp_path, ) target = tmp_path / acr.AUDIT_JSONL_PATH content = target.read_text(encoding="utf-8").strip().splitlines() assert len(content) == 1 record = json.loads(content[0]) expected_keys = { "task_id", "ts", "from_state", "to_state", "reason", "evidence_paths", } assert expected_keys <= set(record.keys()) assert record["task_id"] == "task-2471" assert record["from_state"] == "RECOVERABLE_BLOCKED" assert record["to_state"] == "MERGING" assert record["evidence_paths"] == ["a.json", "b.json"] def test_append_recovery_appends_not_overwrites(acr, tmp_path): """동일 파일에 2회 append 시 라인 2개 (덮어쓰기 X).""" for i in range(3): acr.append_recovery( task_id=f"task-{i}", from_state="RECOVERABLE_BLOCKED", to_state="MERGING", reason=f"reason-{i}", evidence_paths=[], workspace=tmp_path, ) target = tmp_path / acr.AUDIT_JSONL_PATH lines = target.read_text(encoding="utf-8").strip().splitlines() assert len(lines) == 3 task_ids = [json.loads(l)["task_id"] for l in lines] assert task_ids == ["task-0", "task-1", "task-2"] def test_append_recovery_explicit_ts(acr, tmp_path): """``ts`` 인자가 그대로 기록되는지.""" custom_ts = "2026-05-07T01:23:45Z" acr.append_recovery( task_id="task-X", from_state="A", to_state="B", reason="t", evidence_paths=[], ts=custom_ts, workspace=tmp_path, ) target = tmp_path / acr.AUDIT_JSONL_PATH record = json.loads(target.read_text().strip().splitlines()[0]) assert record["ts"] == custom_ts def test_append_recovery_default_ts_iso8601(acr, tmp_path): """``ts`` 미지정 시 ISO 8601 UTC 형식.""" acr.append_recovery( task_id="task-Y", from_state="A", to_state="B", reason="t", evidence_paths=[], workspace=tmp_path, ) target = tmp_path / acr.AUDIT_JSONL_PATH record = json.loads(target.read_text().strip().splitlines()[0]) ts = record["ts"] # ISO 형식 검증 (Z 접미) assert ts.endswith("Z") assert "T" in ts # --------------------------------------------------------------------------- # 입력 검증 # --------------------------------------------------------------------------- def test_append_recovery_rejects_empty_task_id(acr, tmp_path): with pytest.raises(ValueError): acr.append_recovery( task_id="", from_state="A", to_state="B", reason="x", evidence_paths=[], workspace=tmp_path, ) def test_append_recovery_rejects_empty_from_state(acr, tmp_path): with pytest.raises(ValueError): acr.append_recovery( task_id="task-1", from_state="", to_state="B", reason="x", evidence_paths=[], workspace=tmp_path, ) def test_append_recovery_rejects_empty_to_state(acr, tmp_path): with pytest.raises(ValueError): acr.append_recovery( task_id="task-1", from_state="A", to_state="", reason="x", evidence_paths=[], workspace=tmp_path, ) # --------------------------------------------------------------------------- # read_recoveries graceful 동작 # --------------------------------------------------------------------------- def test_read_recoveries_missing_file_returns_empty_list(acr, tmp_path): """파일 부재 시 빈 리스트 반환 (graceful).""" out = acr.read_recoveries(workspace=tmp_path) assert out == [] def test_read_recoveries_round_trip(acr, tmp_path): """append 한 다음 read 로 동일 record 회수.""" acr.append_recovery( task_id="task-A", from_state="X", to_state="Y", reason="r1", evidence_paths=["e1"], workspace=tmp_path, ) acr.append_recovery( task_id="task-B", from_state="X", to_state="Y", reason="r2", evidence_paths=["e2"], workspace=tmp_path, ) out = acr.read_recoveries(workspace=tmp_path) assert len(out) == 2 assert out[0]["task_id"] == "task-A" assert out[1]["task_id"] == "task-B" def test_read_recoveries_skips_malformed_lines(acr, tmp_path): """JSON parse 실패 라인은 skip.""" target = tmp_path / acr.AUDIT_JSONL_PATH target.parent.mkdir(parents=True, exist_ok=True) target.write_text( '{"task_id": "task-1", "ts": "T", "from_state": "A", ' '"to_state": "B", "reason": "x", "evidence_paths": []}\n' "garbage line not json\n" '{"task_id": "task-2", "ts": "T", "from_state": "A", ' '"to_state": "B", "reason": "y", "evidence_paths": []}\n', encoding="utf-8", ) out = acr.read_recoveries(workspace=tmp_path) assert len(out) == 2 assert [r["task_id"] for r in out] == ["task-1", "task-2"] # --------------------------------------------------------------------------- # evidence_paths 정규화 # --------------------------------------------------------------------------- def test_evidence_paths_none_becomes_empty_list(acr, tmp_path): """evidence_paths=None 또는 빈 sequence 시 빈 리스트 직렬화.""" acr.append_recovery( task_id="task-Z", from_state="A", to_state="B", reason="x", evidence_paths=[], workspace=tmp_path, ) target = tmp_path / acr.AUDIT_JSONL_PATH record = json.loads(target.read_text().strip().splitlines()[0]) assert record["evidence_paths"] == []