"""task-2509+1 회귀 테스트 — review_gate_passed + fallback_review_passed 12 케이스. QA 담당: 모리건(Morrigan) 대상: utils/merge_queue_executor.py (task-2509+1 보강) 케이스 목록 (회장 §1~12): TC-01 Gemini COMPLETED + unresolved 0 → review_gate_passed=true TC-02 Gemini UNAVAILABLE_QUOTA + fallback PASS → review_gate_passed=true (★ PR #58 fixture 재현) TC-03 Gemini UNAVAILABLE_QUOTA + fallback FAIL → BLOCK (FALLBACK_REVIEW_FAILED) TC-04 Gemini TIMEOUT + fallback PASS → AUTO_MERGE_ALLOWED TC-05 Gemini UNRESOLVED + real_bug → BLOCK (GEMINI_UNRESOLVED_BLOCK) TC-06 Gemini SCOPE_EXPANSION → CRITICAL_GEMINI_SCOPE_EXPANSION TC-07 smoke_command=None + dry_run=True → 정상 진행 (WARN/SKIP 허용) TC-08 smoke_command=None + dry_run=False → BLOCK (NON_DRY_RUN_REQUIRES_SMOKE_COMMAND) TC-09 HIGH_CORE risk file 변경 + Gemini quota → static_risky_pattern_scan 호출 TC-10 후행 PR stale 재검증에서 BEHIND 감지 TC-11 후행 PR effective diff 오염 감지 TC-12 audit JSON에 신규 7 필드 기록 검증 """ from __future__ import annotations import json import subprocess import sys from pathlib import Path import pytest # workspace root → sys.path (force position 0 to shadow /home/jay/workspace/utils) WORKSPACE = Path(__file__).resolve().parent.parent.parent if str(WORKSPACE) in sys.path: sys.path.remove(str(WORKSPACE)) sys.path.insert(0, str(WORKSPACE)) from utils.merge_queue_executor import ( # noqa: E402 # pyright: ignore[reportMissingImports] AUTO_MERGE_ALLOWED, BLOCKED_WITH_REASON, CRITICAL_GEMINI_SCOPE_EXPANSION, FALLBACK_REVIEW_FAILED, GEMINI_COMPLETED, GEMINI_REAL_BUG, GEMINI_SCOPE_EXPANSION, GEMINI_TIMEOUT, GEMINI_UNAVAILABLE_QUOTA, GEMINI_UNRESOLVED, GEMINI_UNRESOLVED_BLOCK, NON_DRY_RUN_REQUIRES_SMOKE_COMMAND, RISK_LEVEL_HIGH_CORE, RISK_LEVEL_LOW, ExecutorContext, QueueDecision, TaskSpec, assess_risk_level, evaluate_pr, recheck_following_prs, static_risky_pattern_scan, verify_head_lock_then_merge, ) # ─── helpers ────────────────────────────────────────────────────────────────── def cp(returncode: int = 0, stdout: str = "", stderr: str = "") -> subprocess.CompletedProcess: """CompletedProcess 생성 헬퍼.""" return subprocess.CompletedProcess(args=[], returncode=returncode, stdout=stdout, stderr=stderr) def make_runner(returns_by_args: dict | None = None): """args 패턴별 CompletedProcess 반환 fake runner. key = tuple of arg tokens to match (subset of args), value = CompletedProcess to return. Unmatched → returncode=0, empty stdout/stderr. """ calls: list[dict] = [] def runner(args, cwd=None, timeout=60): calls.append({"args": list(args), "cwd": cwd, "timeout": timeout}) for pattern, completed in (returns_by_args or {}).items(): if all(p in args for p in pattern): return completed return cp() runner.calls = calls # type: ignore[attr-defined] # pyright: ignore[reportFunctionMemberAccess] return runner # ─── fixtures ───────────────────────────────────────────────────────────────── EXPECTED_FILES = [ "utils/merge_queue_executor.py", "tests/regression/test_merge_queue_executor_review_gate_2509p1.py", ] CLEAN_MERGE_STATE = { "mergeStateStatus": "CLEAN", "headRefOid": "sha-clean-001", "baseRefName": "main", } CI_OK = {"status": "SUCCESS", "details": ["SUCCESS"], "raw": []} GEMINI_OK = {"status": "ok", "unresolved": [], "hook": None} @pytest.fixture def serial_task_spec() -> TaskSpec: return TaskSpec( task_id="task-2509+1", expected_files=list(EXPECTED_FILES), risk_area="merge_queue/review_gate/fallback_review", dependency=[], parallel_policy="serial_only", merge_queue_position=5, stale_recheck_required=True, cherry_pick_allowed=False, smoke_command=None, ) def _make_ctx( runner=None, pr_workdir: str | None = None, smoke_command=None, fixture_main_sha: str = "main-sha-fixture-001", main_log_grep=None, ) -> ExecutorContext: """기본 ExecutorContext 생성 헬퍼.""" if runner is None: runner = make_runner({}) return ExecutorContext( runner=runner, pr_workdir=pr_workdir, smoke_command=smoke_command, no_audit=True, fixture_main_sha=fixture_main_sha, main_log_grep=main_log_grep, ) # ═══════════════════════════════════════════════════════════════════════════════ # TC-01: Gemini COMPLETED + unresolved 0 → review_gate_passed=true # ═══════════════════════════════════════════════════════════════════════════════ def test_tc01_gemini_completed_review_gate_passes(serial_task_spec: TaskSpec) -> None: """Gemini status=ok + unresolved 0 → gemini_status=GEMINI_COMPLETED, review_gate_passed=true, fallback_review_used=false, AUTO_MERGE_ALLOWED.""" ctx = _make_ctx() # 의도적으로 LOW risk 파일 사용 (HIGH_CORE 트리거 회피 — TC-01은 정상 경로 검증) low_risk_files = [ "tests/regression/test_merge_queue_executor_review_gate_2509p1.py", ] serial_task_spec.expected_files = list(low_risk_files) decision = evaluate_pr( pr_number=58, task_spec=serial_task_spec, pr_head_sha="sha-pr-tc01", effective_files=list(low_risk_files), merge_state=CLEAN_MERGE_STATE, ci_state=CI_OK, gemini_state=GEMINI_OK, ctx=ctx, ) assert decision.gemini_status == GEMINI_COMPLETED, ( f"expected GEMINI_COMPLETED, got {decision.gemini_status}" ) assert decision.review_gate_passed is True, ( f"review_gate_passed must be True for COMPLETED+unresolved=0: {decision.reason}" ) assert decision.fallback_review_used is False, ( f"fallback must NOT be used when Gemini COMPLETED: {decision.fallback_check_details}" ) assert decision.decision == AUTO_MERGE_ALLOWED, ( f"expected AUTO_MERGE_ALLOWED, got {decision.decision} / reason={decision.reason}" ) # ═══════════════════════════════════════════════════════════════════════════════ # TC-02: Gemini UNAVAILABLE_QUOTA + fallback PASS → review_gate_passed=true # (★ PR #58 fixture 재현) # ═══════════════════════════════════════════════════════════════════════════════ def test_tc02_gemini_unavailable_quota_fallback_pass(serial_task_spec: TaskSpec) -> None: """★ PR #58 fixture 재현 — Gemini quota daily limit 상황에서 fallback_review 8조건 모두 PASS이면 review_gate_passed=true이고 AUTO_MERGE_ALLOWED 결정을 받아야 한다.""" # smoke_command 정의 (8조건 중 smoke_command_defined 만족) smoke_cmd = ["python3", "-m", "pytest", "tests/smoke"] ctx = _make_ctx(smoke_command=smoke_cmd) # Gemini quota 미가용 fixture gemini_quota = { "status": "unavailable_quota", "unresolved": [], "hook": None, "errors": [{"message": "Quota exceeded for daily limit"}], } # LOW risk 파일만 사용 (static scan 회피, fallback 8조건 PASS 가능) low_risk_files = [ "tests/regression/test_merge_queue_executor_review_gate_2509p1.py", ] serial_task_spec.expected_files = list(low_risk_files) decision = evaluate_pr( pr_number=58, task_spec=serial_task_spec, pr_head_sha="38334b09", # PR #58 fixture main HEAD effective_files=list(low_risk_files), merge_state=CLEAN_MERGE_STATE, ci_state=CI_OK, gemini_state=gemini_quota, ctx=ctx, ) # Gemini 분류 assert decision.gemini_status == GEMINI_UNAVAILABLE_QUOTA, ( f"expected GEMINI_UNAVAILABLE_QUOTA, got {decision.gemini_status}" ) # fallback 사용 assert decision.fallback_review_used is True, ( "fallback_review_used must be True for Gemini UNAVAILABLE_QUOTA" ) # fallback 8조건 PASS assert decision.fallback_review_passed is True, ( f"fallback failed: {decision.fallback_check_details}" ) # review gate 통과 assert decision.review_gate_passed is True, ( f"review_gate_passed must be True: {decision.reason}" ) # 최종 결정 assert decision.decision == AUTO_MERGE_ALLOWED, ( f"expected AUTO_MERGE_ALLOWED, got {decision.decision} / reason={decision.reason}" ) # ═══════════════════════════════════════════════════════════════════════════════ # TC-03: Gemini UNAVAILABLE_QUOTA + fallback FAIL → BLOCK # ═══════════════════════════════════════════════════════════════════════════════ def test_tc03_gemini_unavailable_quota_fallback_fail_blocks(serial_task_spec: TaskSpec) -> None: """Gemini quota 미가용이지만 fallback 8조건 중 1건 이상 FAIL 시 BLOCKED_WITH_REASON + reason에 FALLBACK_REVIEW_FAILED.""" # smoke_command=None → smoke_command_defined=False 조건 FAIL 유발 ctx = _make_ctx(smoke_command=None) gemini_quota = { "status": "unavailable_quota", "unresolved": [], "hook": None, "errors": [{"message": "Quota exceeded for daily limit"}], } low_risk_files = [ "tests/regression/test_merge_queue_executor_review_gate_2509p1.py", ] serial_task_spec.expected_files = list(low_risk_files) decision = evaluate_pr( pr_number=58, task_spec=serial_task_spec, pr_head_sha="sha-pr-tc03", effective_files=list(low_risk_files), merge_state=CLEAN_MERGE_STATE, ci_state=CI_OK, gemini_state=gemini_quota, ctx=ctx, ) assert decision.gemini_status == GEMINI_UNAVAILABLE_QUOTA, ( f"expected GEMINI_UNAVAILABLE_QUOTA, got {decision.gemini_status}" ) assert decision.fallback_review_used is True assert decision.fallback_review_passed is False, ( f"fallback must FAIL when smoke_command=None: {decision.fallback_check_details}" ) assert decision.review_gate_passed is False assert decision.decision == BLOCKED_WITH_REASON, ( f"expected BLOCKED_WITH_REASON, got {decision.decision}" ) assert FALLBACK_REVIEW_FAILED in decision.reason, ( f"reason must mention FALLBACK_REVIEW_FAILED: {decision.reason}" ) # ═══════════════════════════════════════════════════════════════════════════════ # TC-04: Gemini TIMEOUT + fallback PASS → AUTO_MERGE_ALLOWED # ═══════════════════════════════════════════════════════════════════════════════ def test_tc04_gemini_timeout_fallback_pass(serial_task_spec: TaskSpec) -> None: """Gemini polling timeout이지만 fallback 8조건 모두 PASS → 통과.""" smoke_cmd = ["python3", "-m", "pytest", "tests/smoke"] ctx = _make_ctx(smoke_command=smoke_cmd) gemini_timeout = { "status": "timeout", "unresolved": [], "hook": None, "errors": [{"message": "Polling deadline exceeded"}], } low_risk_files = [ "tests/regression/test_merge_queue_executor_review_gate_2509p1.py", ] serial_task_spec.expected_files = list(low_risk_files) decision = evaluate_pr( pr_number=58, task_spec=serial_task_spec, pr_head_sha="sha-pr-tc04", effective_files=list(low_risk_files), merge_state=CLEAN_MERGE_STATE, ci_state=CI_OK, gemini_state=gemini_timeout, ctx=ctx, ) assert decision.gemini_status == GEMINI_TIMEOUT, ( f"expected GEMINI_TIMEOUT, got {decision.gemini_status}" ) assert decision.fallback_review_used is True assert decision.fallback_review_passed is True, ( f"fallback must PASS: {decision.fallback_check_details}" ) assert decision.review_gate_passed is True assert decision.decision == AUTO_MERGE_ALLOWED, ( f"expected AUTO_MERGE_ALLOWED, got {decision.decision} / reason={decision.reason}" ) # ═══════════════════════════════════════════════════════════════════════════════ # TC-05: Gemini UNRESOLVED + real_bug → BLOCK # ═══════════════════════════════════════════════════════════════════════════════ def test_tc05_gemini_unresolved_real_bug_blocks(serial_task_spec: TaskSpec) -> None: """Gemini auto_triage_candidate (inside expected) → GEMINI_UNRESOLVED_BLOCK. real_bug=True 분류 시에도 머지 차단되어야 한다.""" ctx = _make_ctx() gemini_unresolved = { "status": "auto_triage_candidate", "unresolved": [ {"path": "utils/merge_queue_executor.py", "body": "real bug: NPE on line 42"}, ], "inside": [ {"path": "utils/merge_queue_executor.py", "body": "real bug: NPE on line 42"}, ], "hook": "auto_gemini_triage", "real_bug": True, } decision = evaluate_pr( pr_number=58, task_spec=serial_task_spec, pr_head_sha="sha-pr-tc05", effective_files=list(EXPECTED_FILES), merge_state=CLEAN_MERGE_STATE, ci_state=CI_OK, gemini_state=gemini_unresolved, ctx=ctx, ) # gemini_status: real_bug=True 시 GEMINI_REAL_BUG으로 분류 assert decision.gemini_status in (GEMINI_REAL_BUG, GEMINI_UNRESOLVED), ( f"expected GEMINI_REAL_BUG or GEMINI_UNRESOLVED, got {decision.gemini_status}" ) # 머지 차단 assert decision.decision == GEMINI_UNRESOLVED_BLOCK, ( f"expected GEMINI_UNRESOLVED_BLOCK, got {decision.decision} / reason={decision.reason}" ) assert decision.review_gate_passed is False # ═══════════════════════════════════════════════════════════════════════════════ # TC-06: Gemini SCOPE_EXPANSION → Critical #3 (GEMINI_REAL_BUG_SCOPE_EXPANSION) # ═══════════════════════════════════════════════════════════════════════════════ def test_tc06_gemini_scope_expansion_critical(serial_task_spec: TaskSpec) -> None: """Gemini critical_scope_expansion → BLOCKED_WITH_REASON + critical_code=CRITICAL_GEMINI_SCOPE_EXPANSION + classify→GEMINI_SCOPE_EXPANSION.""" ctx = _make_ctx() gemini_scope = { "status": "critical_scope_expansion", "unresolved": [ {"path": "utils/unrelated_file.py", "body": "fix scope expansion outside expected"}, ], "outside": [ {"path": "utils/unrelated_file.py", "body": "fix scope expansion outside expected"}, ], "hook": "critical_escalation_reporter", "critical_code": CRITICAL_GEMINI_SCOPE_EXPANSION, } decision = evaluate_pr( pr_number=58, task_spec=serial_task_spec, pr_head_sha="sha-pr-tc06", effective_files=list(EXPECTED_FILES), merge_state=CLEAN_MERGE_STATE, ci_state=CI_OK, gemini_state=gemini_scope, ctx=ctx, ) assert decision.gemini_status == GEMINI_SCOPE_EXPANSION, ( f"expected GEMINI_SCOPE_EXPANSION, got {decision.gemini_status}" ) assert decision.decision == BLOCKED_WITH_REASON, ( f"expected BLOCKED_WITH_REASON, got {decision.decision}" ) assert decision.critical_code == CRITICAL_GEMINI_SCOPE_EXPANSION, ( f"expected critical_code=CRITICAL_GEMINI_SCOPE_EXPANSION, got {decision.critical_code}" ) assert decision.review_gate_passed is False # ═══════════════════════════════════════════════════════════════════════════════ # TC-07: smoke_command=None + dry_run=True → 정상 진행 (WARN/SKIP) # ═══════════════════════════════════════════════════════════════════════════════ def test_tc07_dry_run_without_smoke_allowed() -> None: """smoke_command=None + dry_run=True 시: verify_head_lock_then_merge가 BLOCK하지 않고 dry-run 결과 그대로 통과. 회장 §5: dry_run=True + None → WARN/SKIP 허용.""" ctx = _make_ctx(smoke_command=None) decision = QueueDecision( decision=AUTO_MERGE_ALLOWED, pr_number=58, task_id="task-2509+1", pr_head_sha_start="sha-locked-tc07", ) result = verify_head_lock_then_merge( decision=decision, pr_number=58, ctx=ctx, fetch_pr_head_at_merge=lambda _n: "sha-locked-tc07", dry_run=True, ) # dry_run=True → smoke_command=None이라도 BLOCK 금지 assert result.decision == AUTO_MERGE_ALLOWED, ( f"dry_run=True + smoke_command=None must NOT BLOCK, got {result.decision} / reason={result.reason}" ) # NON_DRY_RUN_REQUIRES_SMOKE_COMMAND가 reason에 등장하면 안 됨 assert NON_DRY_RUN_REQUIRES_SMOKE_COMMAND not in (result.reason or ""), ( f"dry_run=True path must not raise NON_DRY_RUN_REQUIRES_SMOKE_COMMAND: {result.reason}" ) # ═══════════════════════════════════════════════════════════════════════════════ # TC-08: smoke_command=None + dry_run=False → BLOCK (NON_DRY_RUN_REQUIRES_SMOKE_COMMAND) # ═══════════════════════════════════════════════════════════════════════════════ def test_tc08_non_dry_run_without_smoke_blocks() -> None: """smoke_command=None + dry_run=False → BLOCK + NON_DRY_RUN_REQUIRES_SMOKE_COMMAND. 회장 §5: 자동 머지를 실제 수행하려면 smoke_command 정의 필수.""" ctx = _make_ctx(smoke_command=None) decision = QueueDecision( decision=AUTO_MERGE_ALLOWED, pr_number=58, task_id="task-2509+1", pr_head_sha_start="sha-locked-tc08", ) result = verify_head_lock_then_merge( decision=decision, pr_number=58, ctx=ctx, fetch_pr_head_at_merge=lambda _n: "sha-locked-tc08", dry_run=False, ) assert result.decision == BLOCKED_WITH_REASON, ( f"expected BLOCKED_WITH_REASON, got {result.decision}" ) assert NON_DRY_RUN_REQUIRES_SMOKE_COMMAND in (result.reason or ""), ( f"reason must contain NON_DRY_RUN_REQUIRES_SMOKE_COMMAND: {result.reason}" ) # ═══════════════════════════════════════════════════════════════════════════════ # TC-09: HIGH_CORE risk file 변경 + Gemini quota → static_risky_pattern_scan + Codex 요구 # ═══════════════════════════════════════════════════════════════════════════════ def test_tc09_high_core_static_risky_pattern_scan(serial_task_spec: TaskSpec) -> None: """utils/merge_queue_executor.py 변경 시 risk_level=HIGH_CORE. Gemini quota 미가용 fallback 시 static_risky_pattern_scan이 실행되고, fallback_check_details.checks.static_risky_scan_pass_if_high_core 항목 존재.""" smoke_cmd = ["python3", "-m", "pytest", "tests/smoke"] ctx = _make_ctx(smoke_command=smoke_cmd) # HIGH_CORE 트리거 (utils/merge_queue_executor.py) high_core_files = ["utils/merge_queue_executor.py"] serial_task_spec.expected_files = list(high_core_files) gemini_quota = { "status": "unavailable_quota", "unresolved": [], "hook": None, "errors": [{"message": "Quota exceeded for daily limit"}], } decision = evaluate_pr( pr_number=58, task_spec=serial_task_spec, pr_head_sha="sha-pr-tc09", effective_files=list(high_core_files), merge_state=CLEAN_MERGE_STATE, ci_state=CI_OK, gemini_state=gemini_quota, ctx=ctx, ) # risk_level 분류 assert decision.risk_level == RISK_LEVEL_HIGH_CORE, ( f"expected RISK_LEVEL_HIGH_CORE, got {decision.risk_level}" ) # fallback 사용 assert decision.fallback_review_used is True # fallback 8조건 중 static_risky_scan_pass_if_high_core 항목이 반영돼야 함 checks = (decision.fallback_check_details or {}).get("checks", {}) assert "static_risky_scan_pass_if_high_core" in checks, ( f"fallback_check_details.checks must include static_risky_scan_pass_if_high_core: {checks}" ) # static_risky_pattern_scan 단독 호출 — 정적 분석 직접 검증 scan = static_risky_pattern_scan(high_core_files) assert "passed" in scan, f"scan result must have 'passed' key: {scan}" assert isinstance(scan.get("violations"), list), ( f"scan.violations must be list: {scan}" ) # assess_risk_level 단독 검증 assert assess_risk_level(high_core_files) == RISK_LEVEL_HIGH_CORE assert assess_risk_level(["docs/readme.md"]) == RISK_LEVEL_LOW # ═══════════════════════════════════════════════════════════════════════════════ # TC-10: 후행 PR stale 재검증에서 BEHIND 감지 # ═══════════════════════════════════════════════════════════════════════════════ def test_tc10_recheck_following_pr_behind() -> None: """후행 PR이 BEHIND 상태일 때 needs_recheck=True + behind=True.""" behind_payload = json.dumps({ "mergeStateStatus": "BEHIND", "headRefOid": "sha-behind-100", "baseRefName": "main", "files": [{"path": "utils/x.py"}], }) runner = make_runner({ ("100",): cp(stdout=behind_payload), }) queue = [{"pr_number": 100, "expected_files": ["utils/x.py"]}] states = recheck_following_prs(queue, runner) assert len(states) == 1 state = states[0] assert state["pr_number"] == 100 assert state["behind"] is True, f"behind must be True: {state}" assert state["needs_recheck"] is True, f"needs_recheck must be True: {state}" assert state["conflict"] is False assert state["blocked"] is False # ═══════════════════════════════════════════════════════════════════════════════ # TC-11: 후행 PR effective diff 오염 감지 # ═══════════════════════════════════════════════════════════════════════════════ def test_tc11_recheck_following_pr_diff_drift() -> None: """후행 PR의 effective_files가 expected에서 벗어났을 때(또는 prior와 달라졌을 때) needs_recheck=True 또는 contamination 신호 감지. state machine이 신규 필드(effective_diff_drift / expected_files_maintained / forbidden_path_present)를 모두 포함하지 않아도 최소한 needs_recheck=True 또는 DIRTY/conflict 신호로 재검증 필요를 표현해야 한다.""" # 후행 PR이 expected_files 외 추가 파일을 변경한 fixture contaminated_payload = json.dumps({ "mergeStateStatus": "DIRTY", # 가장 흔한 contamination 신호 "headRefOid": "sha-101", "baseRefName": "main", "files": [ {"path": "utils/x.py"}, {"path": "utils/y.py"}, # 추가 파일 (drift) ], }) runner = make_runner({ ("101",): cp(stdout=contaminated_payload), }) queue = [ { "pr_number": 101, "expected_files": ["utils/x.py"], "prior_effective_files": ["utils/x.py"], } ] states = recheck_following_prs(queue, runner) assert len(states) == 1 state = states[0] assert state["pr_number"] == 101 # 재검증 필요 신호 (DIRTY/effective_diff_drift 중 1) assert state["needs_recheck"] is True, ( f"contamination 후행 PR은 needs_recheck=True: {state}" ) # DIRTY → conflict=True assert state["conflict"] is True, f"DIRTY → conflict=True: {state}" # ═══════════════════════════════════════════════════════════════════════════════ # TC-12: audit JSON에 7 신규 필드 기록 검증 # ═══════════════════════════════════════════════════════════════════════════════ def test_tc12_audit_includes_7_new_fields(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: """write_audit 출력 JSON에 신규 7 필드(gemini_status, fallback_review_used, fallback_review_passed, risk_level, review_gate_passed, final_decision, critical_escalation)가 포함돼야 한다. 글로벌 audit 로그가 더럽혀지지 않도록 EVENTS_DIR/AUDIT_DIR/GLOBAL_AUDIT_LOG를 monkeypatch로 tmp_path 하위 디렉토리로 격리한다.""" import utils.merge_queue_executor as mq # pyright: ignore[reportMissingImports] events_dir = tmp_path / "events" audit_dir = tmp_path / "audit" events_dir.mkdir(parents=True, exist_ok=True) audit_dir.mkdir(parents=True, exist_ok=True) global_log = audit_dir / "merge-queue.jsonl" monkeypatch.setattr(mq, "EVENTS_DIR", events_dir) monkeypatch.setattr(mq, "AUDIT_DIR", audit_dir) monkeypatch.setattr(mq, "GLOBAL_AUDIT_LOG", global_log) decision = QueueDecision( decision=AUTO_MERGE_ALLOWED, pr_number=58, task_id="task-2509+1", gemini_status=GEMINI_UNAVAILABLE_QUOTA, fallback_review_used=True, fallback_review_passed=True, risk_level=RISK_LEVEL_HIGH_CORE, review_gate_passed=True, final_decision=AUTO_MERGE_ALLOWED, critical_escalation=None, fallback_check_details={"passed": True, "checks": {"effective_diff_equals_expected": True}}, ) audit_path = mq.write_audit(decision, task_id="task-2509+1", no_audit=False) assert audit_path is not None, "write_audit must return a path when no_audit=False" # 1. per-task audit JSON 파일 검증 per_task_path = Path(audit_path) assert per_task_path.exists(), f"per-task audit must be written: {per_task_path}" data = json.loads(per_task_path.read_text(encoding="utf-8")) required_fields = [ "gemini_status", "fallback_review_used", "fallback_review_passed", "risk_level", "review_gate_passed", "final_decision", "critical_escalation", # Gemini PR 리뷰 medium 코멘트 수용: fallback 추적 필드도 audit 검증 대상에 포함. "fallback_check_details", "static_scan_violations", ] for field in required_fields: assert field in data, ( f"audit JSON missing required field '{field}': keys={list(data.keys())}" ) # 값 정합성 assert data["gemini_status"] == GEMINI_UNAVAILABLE_QUOTA assert data["fallback_review_used"] is True assert data["fallback_review_passed"] is True assert data["risk_level"] == RISK_LEVEL_HIGH_CORE assert data["review_gate_passed"] is True assert data["final_decision"] == AUTO_MERGE_ALLOWED assert data["critical_escalation"] is None # 2. 글로벌 audit JSONL에도 동일 필드 포함 assert global_log.exists(), f"global audit log must be appended: {global_log}" last_line = global_log.read_text(encoding="utf-8").splitlines()[-1] global_data = json.loads(last_line) for field in required_fields: assert field in global_data, ( f"global audit jsonl missing field '{field}': keys={list(global_data.keys())}" )