"""task-2503 회귀 테스트 — Phase 1 amendment audit jsonl format 검증. 회장 amendment 2026-05-08T11:32 Phase 1 추가 요건: - audit jsonl schema (회장 §5 9 필드) 준수 - dry_run=true 필드 옵션 동작 - dispatch.py 미경유 (classifier 직접 호출) - production 차단 동작 X (단순 기록만) """ import json import sys from pathlib import Path import pytest WORKSPACE = Path(__file__).resolve().parent.parent.parent if str(WORKSPACE) not in sys.path: sys.path.insert(0, str(WORKSPACE)) from utils.merge_topology_gate import ( # type: ignore[import] ALLOW, BLOCK, TopologyDecision, audit_log, _dry_run_from_task_file, ) REQUIRED_AUDIT_FIELDS = { "task_id", "decision", "reason_codes", "overlap_score", "conflicting_tasks", "active_tasks_snapshot", "open_prs_snapshot", "override_used", "timestamp", } VALID_DECISIONS = {"ALLOW", "LIMITED_PARALLEL", "BLOCK", "REQUIRE_CHAIR_OVERRIDE"} def _redirect_audit(monkeypatch, tmp_path): """AUDIT_LOG_PATH를 tmp 경로로 우회 (실제 파일 오염 방지).""" target = tmp_path / "merge-topology-gate.jsonl" monkeypatch.setattr( "utils.merge_topology_gate.AUDIT_LOG_PATH", target ) return target # ─── TC-1: 회장 §5 9 필드 모두 기록 ──────────────────────────────────────── def test_audit_jsonl_records_all_9_required_fields(monkeypatch, tmp_path): target = _redirect_audit(monkeypatch, tmp_path) decision = TopologyDecision(decision=ALLOW, reason_codes=[]) audit_log(task_id="task-test-9fields", decision=decision) assert target.exists() line = target.read_text(encoding="utf-8").strip() record = json.loads(line) missing = REQUIRED_AUDIT_FIELDS - set(record.keys()) assert not missing, f"audit jsonl missing required fields: {missing}" # ─── TC-2: decision 값이 4 enum 중 하나 ──────────────────────────────────── def test_audit_decision_value_is_in_enum(monkeypatch, tmp_path): target = _redirect_audit(monkeypatch, tmp_path) for d in ("ALLOW", "BLOCK", "LIMITED_PARALLEL", "REQUIRE_CHAIR_OVERRIDE"): audit_log(task_id=f"task-enum-{d}", decision=TopologyDecision(decision=d)) lines = target.read_text(encoding="utf-8").strip().splitlines() assert len(lines) == 4 for line in lines: record = json.loads(line) assert record["decision"] in VALID_DECISIONS # ─── TC-3: timestamp KST ISO 8601 (+09:00) ──────────────────────────────── def test_audit_timestamp_is_kst_iso_8601(monkeypatch, tmp_path): target = _redirect_audit(monkeypatch, tmp_path) audit_log(task_id="task-tz", decision=TopologyDecision(decision=ALLOW)) record = json.loads(target.read_text(encoding="utf-8").strip()) ts = record["timestamp"] assert "+09:00" in ts, f"timestamp must be KST (+09:00): {ts}" # ISO 8601 'T' 구분 assert "T" in ts # ─── TC-4: append-only 보장 (기존 라인 보존 + 새 라인 append) ───────────── def test_audit_jsonl_is_append_only(monkeypatch, tmp_path): target = _redirect_audit(monkeypatch, tmp_path) target.parent.mkdir(parents=True, exist_ok=True) target.write_text("{\"task_id\": \"existing\", \"decision\": \"ALLOW\"}\n", encoding="utf-8") audit_log(task_id="task-new", decision=TopologyDecision(decision=BLOCK)) lines = target.read_text(encoding="utf-8").strip().splitlines() assert len(lines) == 2 assert json.loads(lines[0])["task_id"] == "existing" assert json.loads(lines[1])["task_id"] == "task-new" # ─── TC-5: dry_run=True 시 dry_run 필드 추가 (Phase 1 amendment) ────────── def test_audit_dry_run_field_recorded_when_set(monkeypatch, tmp_path): target = _redirect_audit(monkeypatch, tmp_path) audit_log( task_id="task-dryrun", decision=TopologyDecision(decision=ALLOW), dry_run=True, ) record = json.loads(target.read_text(encoding="utf-8").strip()) assert record.get("dry_run") is True # ─── TC-6: dry_run=False (기본) 시 dry_run 필드 미기록 ──────────────────── def test_audit_dry_run_absent_by_default(monkeypatch, tmp_path): target = _redirect_audit(monkeypatch, tmp_path) audit_log(task_id="task-prod", decision=TopologyDecision(decision=ALLOW)) record = json.loads(target.read_text(encoding="utf-8").strip()) assert "dry_run" not in record # ─── TC-7: override_used 필드 정확 기록 ─────────────────────────────────── def test_audit_override_used_field(monkeypatch, tmp_path): target = _redirect_audit(monkeypatch, tmp_path) audit_log( task_id="task-override", decision=TopologyDecision(decision="REQUIRE_CHAIR_OVERRIDE"), override_used=True, ) record = json.loads(target.read_text(encoding="utf-8").strip()) assert record["override_used"] is True # ─── TC-8: dry-run CLI helper가 dispatch.py를 import 하지 않음 ─────────── def test_dry_run_helper_does_not_import_dispatch(): """Phase 1 핵심 보장: classifier/CLI는 dispatch 모듈을 import 하지 않는다.""" src_path = WORKSPACE / "utils" / "merge_topology_gate.py" src = src_path.read_text(encoding="utf-8") # 코드 라인만 검사 (docstring/주석은 제외) code_lines = [ ln.strip() for ln in src.splitlines() if ln.strip() and not ln.strip().startswith("#") ] bad = [ ln for ln in code_lines if ln.startswith("import dispatch") or ln.startswith("from dispatch") ] assert not bad, f"Phase 1 위반: classifier가 dispatch import — {bad}" # ─── TC-9: _dry_run_from_task_file이 dry_run=true 필드를 audit에 기록 ──── def test_dry_run_from_task_file_records_dry_run_field(monkeypatch, tmp_path): target = _redirect_audit(monkeypatch, tmp_path) # 최소 task spec 파일 작성 spec = tmp_path / "task-test-dryrun-2503.md" spec.write_text( "# task-test-dryrun-2503\n\n" "```yaml\n" "expected_files:\n" " - utils/example.py\n" "risk_area: \"governance\"\n" "dependency: [\"none\"]\n" "parallel_policy: \"serial_only\"\n" "merge_queue_position: 1\n" "stale_recheck_required: false\n" "cherry_pick_allowed: false\n" "```\n", encoding="utf-8", ) _decision, summary = _dry_run_from_task_file( task_file=spec, timer_path=tmp_path / "no-such-timers.json", write_audit=True, ) assert summary["dry_run"] is True record = json.loads(target.read_text(encoding="utf-8").strip().splitlines()[-1]) assert record.get("dry_run") is True # ─── TC-10: jsonl 라인이 valid JSON임을 보장 (parse 성공) ───────────────── def test_audit_jsonl_lines_are_valid_json(monkeypatch, tmp_path): target = _redirect_audit(monkeypatch, tmp_path) decisions = [ TopologyDecision(decision=ALLOW), TopologyDecision(decision=BLOCK, reason_codes=["DUPLICATE_FILE"]), TopologyDecision(decision="LIMITED_PARALLEL", reason_codes=["DUPLICATE_LIFECYCLE"]), ] for i, dec in enumerate(decisions): audit_log(task_id=f"task-json-{i}", decision=dec) for line in target.read_text(encoding="utf-8").strip().splitlines(): json.loads(line) # parse 실패 시 자동 fail if __name__ == "__main__": pytest.main([__file__, "-v"])