"""anu_v2.tests.test_gemini_second_review_bottleneck_2565 — task-2565 Phase 1+2 테스트. Phase 1: Schema + Policy 상수 박제 테스트 (6 종). Phase 2: Stale detection + owner trigger 자동 호출 테스트 (5 종). 회장 §명시 task-2565 §5 Phase 1 Test 요구사항 1:1 박제: - test_second_review_decision_schema_validation - test_pre_merge_commit_decision_schema_validation - test_state_enum_serialization - test_long_polling_forbidden_constant - test_grace_default_180s - test_max_recheck_default_1 Phase 2 테스트 (task-2565 §5 Phase 2): - test_follow_up_commit_marks_gemini_stale_on_head - test_second_review_owner_trigger_called_after_grace - test_same_head_duplicate_owner_trigger_deduped - test_grace_not_elapsed_returns_pending - test_phase2_markers_emitted_correctly """ from __future__ import annotations import json from pathlib import Path from unittest.mock import MagicMock import pytest from anu_v2.polling_policy import ( LONG_POLLING_FORBIDDEN, MAX_SECOND_REVIEW_RECHECKS, SECOND_REVIEW_GRACE_SECONDS, ) from anu_v2.second_review_recovery import ( MAX_CI_RERUN_PER_HEAD, CIRerunInput, MergeReadyInput, PreMergeCommitInput, SchemaViolation, SecondReviewInput, SecondReviewState, auto_rerun_failed_ci_jobs, auto_trigger_owner_review, build_pre_merge_block_decision, build_pre_merge_commit_decision, build_second_review_decision, classify_pre_merge_commit, determine_state, emit_phase2_markers, is_gemini_stale_on_head_after_followup, is_merge_ready, is_owner_trigger_deduped, post_merge_evidence_redirect_payload, should_rerun_failed_ci_jobs, validate_pre_merge_commit_decision, validate_second_review_decision, write_marker, ) # fixture 경로 _FIXTURE_DIR = ( Path(__file__).parent.parent / "fixtures" / "gemini_second_review_bottleneck" ) # ─── Phase 1: Schema + Policy 상수 ─────────────────────────────────────────── class TestPhase1SchemaAndPolicy: """Phase 1 — Schema 검증 + 정책 상수 박제 테스트.""" # ── 헬퍼 ──────────────────────────────────────────────────────────────── def _valid_second_review_decision(self) -> dict: """정상 second_review_decision.v1 dict 반환.""" return build_second_review_decision( task_id="task-2565", pr_number=107, old_head_sha="d251399c", current_head_sha="cee55afe", latest_gemini_commit_id="d251399c", gemini_fresh_on_current_head=False, stale_reason="SHA_MISMATCH_AFTER_FOLLOW_UP_COMMIT", ci_gate_reason="GEMINI_EVIDENCE_STALE", owner_trigger_required=True, owner_trigger_result="POSTED", fresh_detected_at=None, ci_rerun_required=True, final_decision="WAITING", ) def _valid_pre_merge_commit_decision(self) -> dict: """정상 pre_merge_commit_decision.v1 dict 반환.""" return build_pre_merge_commit_decision( merge_ready=True, changed_files=["anu_v2/second_review_recovery.py"], change_type="report_only", pre_merge_commit_allowed=False, redirect_to_post_merge_evidence=True, ) # ── test 1: second_review_decision schema 검증 ────────────────────────── def test_second_review_decision_schema_validation(self) -> None: """second_review_decision.v1 schema 검증 5 종 케이스 (task-2565 §5 Phase1 Test). 케이스: 1. 정상 decision → validate 통과 (SchemaViolation raise 없음) 2. manual_owner_input_requested=True → SchemaViolation 3. long_polling_used=True → SchemaViolation 4. 필수 필드 누락 → SchemaViolation 5. 잘못된 final_decision enum → SchemaViolation """ # 케이스 1: 정상 PASS decision = self._valid_second_review_decision() validate_second_review_decision(decision) # SchemaViolation 없어야 함 # 케이스 2: manual_owner_input_requested=True → raise bad_manual = dict(decision) bad_manual["manual_owner_input_requested"] = True with pytest.raises(SchemaViolation, match="manual_owner_input_requested"): validate_second_review_decision(bad_manual) # 케이스 3: long_polling_used=True → raise bad_polling = dict(decision) bad_polling["long_polling_used"] = True with pytest.raises(SchemaViolation, match="long_polling_used"): validate_second_review_decision(bad_polling) # 케이스 4: 필수 필드 누락 → raise missing_field = dict(decision) del missing_field["task_id"] with pytest.raises(SchemaViolation, match="누락 필드"): validate_second_review_decision(missing_field) # 케이스 5: 잘못된 final_decision enum → raise bad_enum = dict(decision) bad_enum["final_decision"] = "INVALID_DECISION_VALUE" with pytest.raises(SchemaViolation, match="final_decision"): validate_second_review_decision(bad_enum) # ── test 2: pre_merge_commit_decision schema 검증 ─────────────────────── def test_pre_merge_commit_decision_schema_validation(self) -> None: """pre_merge_commit_decision.v1 schema 검증 3 종 케이스 (task-2565 §5 Phase1 Test). 케이스: 1. 정상 decision → validate 통과 (SchemaViolation raise 없음) 2. 잘못된 change_type → SchemaViolation 3. 필수 필드 누락 → SchemaViolation """ # 케이스 1: 정상 PASS decision = self._valid_pre_merge_commit_decision() validate_pre_merge_commit_decision(decision) # SchemaViolation 없어야 함 # 케이스 2: 잘못된 change_type → raise bad_type = dict(decision) bad_type["change_type"] = "INVALID_CHANGE_TYPE" with pytest.raises(SchemaViolation, match="change_type"): validate_pre_merge_commit_decision(bad_type) # 케이스 3: 필수 필드 누락 → raise missing_field = dict(decision) del missing_field["merge_ready"] with pytest.raises(SchemaViolation, match="누락 필드"): validate_pre_merge_commit_decision(missing_field) # ── test 3: state enum 직렬화 ──────────────────────────────────────────── def test_state_enum_serialization(self) -> None: """9 상태 모두 string 변환 + 역변환 가능 (task-2565 §2). SecondReviewState 9 개 전수: - .value 가 non-empty string - SecondReviewState(value) 역변환 == 원본 enum """ expected_states = { "FOLLOW_UP_COMMIT_CREATED", "GEMINI_EVIDENCE_STALE_ON_HEAD", "SECOND_REVIEW_PENDING", "SECOND_REVIEW_TRIGGER_REQUIRED", "SECOND_REVIEW_OWNER_TRIGGER_POSTED", "SECOND_REVIEW_FRESH_DETECTED", "SECOND_REVIEW_TIMEOUT", "CI_RERUN_REQUIRED_AFTER_FRESH", "MERGE_READY_AFTER_SECOND_REVIEW", } actual_values = {s.value for s in SecondReviewState} assert actual_values == expected_states, ( f"상태 enum 불일치: 누락={expected_states - actual_values}, " f"추가={actual_values - expected_states}" ) # 역변환 확인 for state in SecondReviewState: assert isinstance(state.value, str), ( f"{state.name}.value 가 str 이 아님: {type(state.value)}" ) assert state.value, f"{state.name}.value 가 빈 string" roundtrip = SecondReviewState(state.value) assert roundtrip == state, ( f"역변환 실패: SecondReviewState({state.value!r}) != {state}" ) # ── test 4: LONG_POLLING_FORBIDDEN 상수 박제 ────────────────────────────── def test_long_polling_forbidden_constant(self) -> None: """LONG_POLLING_FORBIDDEN is True (task-2565 §3, §8 #4). long polling 금지 상수가 polling_policy.py 에 정확히 True 로 박제되어 있음을 검증. """ assert LONG_POLLING_FORBIDDEN is True, ( "LONG_POLLING_FORBIDDEN must be True — long polling 금지 박제 (task-2565 §3)" ) # ── test 5: SECOND_REVIEW_GRACE_SECONDS == 180 ────────────────────────── def test_grace_default_180s(self) -> None: """SECOND_REVIEW_GRACE_SECONDS == 180 (task-2565 §3, 회장 결정 180초). follow-up commit 후 fresh evidence 대기 grace 구간이 180초로 박제됨. """ assert SECOND_REVIEW_GRACE_SECONDS == 180, ( f"SECOND_REVIEW_GRACE_SECONDS must be 180, got {SECOND_REVIEW_GRACE_SECONDS}" ) # ── test 6: MAX_SECOND_REVIEW_RECHECKS == 1 ────────────────────────────── def test_max_recheck_default_1(self) -> None: """MAX_SECOND_REVIEW_RECHECKS == 1 (task-2565 §3). owner_trigger 호출 후 재확인 횟수가 max 1 회로 박제됨. """ assert MAX_SECOND_REVIEW_RECHECKS == 1, ( f"MAX_SECOND_REVIEW_RECHECKS must be 1, got {MAX_SECOND_REVIEW_RECHECKS}" ) # ─── Phase 2: Stale detection + owner trigger 자동 호출 ─────────────────────── def _load_fixture(name: str) -> dict: """fixture JSON 로드 헬퍼.""" path = _FIXTURE_DIR / name return json.loads(path.read_text(encoding="utf-8")) def _inp_from_fixture(data: dict) -> SecondReviewInput: """fixture dict → SecondReviewInput 변환 헬퍼.""" return SecondReviewInput( task_id=data["task_id"], pr_number=data["pr_number"], old_head_sha=data["old_head_sha"], current_head_sha=data["current_head_sha"], latest_gemini_commit_id=data.get("latest_gemini_commit_id"), ci_gate_failure_reason=data.get("ci_gate_failure_reason"), unresolved_thread_outdated=data["unresolved_thread_outdated"], follow_up_commit_detected=data["follow_up_commit_detected"], elapsed_since_follow_up_seconds=data["elapsed_since_follow_up_seconds"], owner_trigger_audit_entries=tuple(data.get("owner_trigger_audit_entries", [])), ) class TestPhase2StaleDetectionAndOwnerTrigger: """Phase 2 — stale detection + owner trigger 자동 호출 테스트 (task-2565 §5 Phase 2).""" def test_follow_up_commit_marks_gemini_stale_on_head(self) -> None: """fixture dev6 load → 4조건 충족 → stale 판정 + SECOND_REVIEW_TRIGGER_REQUIRED. fixture elapsed_since_follow_up_seconds=200 (>= 180) → grace 경과 → TRIGGER_REQUIRED. """ data = _load_fixture("dev6_old_gemini_sha.json") inp = _inp_from_fixture(data) # 4조건 True assert is_gemini_stale_on_head_after_followup(inp) is True, ( "dev6 fixture는 4조건 모두 충족해야 함" ) # grace 경과 (200 >= 180) → SECOND_REVIEW_TRIGGER_REQUIRED state = determine_state(inp) assert state == SecondReviewState.SECOND_REVIEW_TRIGGER_REQUIRED, ( f"expected SECOND_REVIEW_TRIGGER_REQUIRED, got {state}" ) assert state.value == data["expected_state"], ( f"fixture expected_state 불일치: {state.value!r} != {data['expected_state']!r}" ) def test_second_review_owner_trigger_called_after_grace(self) -> None: """fixture dev6 + grace 경과 + trigger_callable mock → triggered=True, result='POSTED'. mock callable이 (pr_number, current_head_sha)로 호출되었는지 확인. """ data = _load_fixture("dev6_old_gemini_sha.json") inp = _inp_from_fixture(data) mock_callable = MagicMock(return_value={"status": "POSTED"}) result = auto_trigger_owner_review(inp, trigger_callable=mock_callable) assert result["triggered"] is True, f"triggered 기대값 True, 실제값: {result['triggered']}" assert result["result"] == "POSTED", f"result 기대값 POSTED, 실제값: {result['result']!r}" # mock이 올바른 인자로 호출됐는지 확인 mock_callable.assert_called_once_with(inp.pr_number, inp.current_head_sha) assert result["dedupe_key"] == f"{inp.pr_number}+{inp.current_head_sha}" def test_same_head_duplicate_owner_trigger_deduped(self) -> None: """fixture dev6 + audit에 동일 (pr, head) POSTED 존재 → triggered=False, result='DEDUPED'.""" data = _load_fixture("dev6_old_gemini_sha.json") # audit entries에 동일 (pr_number, head_sha) POSTED 항목 추가 audit_entries: tuple[dict, ...] = ( { "pr_number": data["pr_number"], "head_sha": data["current_head_sha"], "result": "POSTED", "source": "test_dedupe", }, ) inp = SecondReviewInput( task_id=data["task_id"], pr_number=data["pr_number"], old_head_sha=data["old_head_sha"], current_head_sha=data["current_head_sha"], latest_gemini_commit_id=data.get("latest_gemini_commit_id"), ci_gate_failure_reason=data.get("ci_gate_failure_reason"), unresolved_thread_outdated=data["unresolved_thread_outdated"], follow_up_commit_detected=data["follow_up_commit_detected"], elapsed_since_follow_up_seconds=data["elapsed_since_follow_up_seconds"], owner_trigger_audit_entries=audit_entries, ) mock_callable = MagicMock(return_value={"status": "POSTED"}) result = auto_trigger_owner_review(inp, trigger_callable=mock_callable) assert result["triggered"] is False, f"triggered 기대값 False, 실제값: {result['triggered']}" assert result["result"] == "DEDUPED", f"result 기대값 DEDUPED, 실제값: {result['result']!r}" # dedupe 발생 시 callable 미호출 mock_callable.assert_not_called() def test_grace_not_elapsed_returns_pending(self) -> None: """elapsed_since_follow_up_seconds=60 (< 180) → SECOND_REVIEW_PENDING + GRACE_PENDING. 4조건은 충족하지만 grace 미경과 → pending 상태. """ data = _load_fixture("dev6_old_gemini_sha.json") inp = SecondReviewInput( task_id=data["task_id"], pr_number=data["pr_number"], old_head_sha=data["old_head_sha"], current_head_sha=data["current_head_sha"], latest_gemini_commit_id=data.get("latest_gemini_commit_id"), ci_gate_failure_reason=data.get("ci_gate_failure_reason"), unresolved_thread_outdated=data["unresolved_thread_outdated"], follow_up_commit_detected=data["follow_up_commit_detected"], elapsed_since_follow_up_seconds=60, # grace 미경과 owner_trigger_audit_entries=(), ) # 상태 → SECOND_REVIEW_PENDING state = determine_state(inp) assert state == SecondReviewState.SECOND_REVIEW_PENDING, ( f"grace 미경과 시 SECOND_REVIEW_PENDING 기대, got {state}" ) # auto_trigger → GRACE_PENDING result = auto_trigger_owner_review(inp, trigger_callable=MagicMock()) assert result["triggered"] is False, f"triggered 기대값 False, 실제값: {result['triggered']}" assert result["result"] == "GRACE_PENDING", ( f"result 기대값 GRACE_PENDING, 실제값: {result['result']!r}" ) def test_phase2_markers_emitted_correctly(self, tmp_path: Path) -> None: """emit_phase2_markers → TRIGGER_REQUIRED marker + second-review-decision.json 생성 확인. Finding 1 fix: emit_phase2_markers는 이제 schema 필수 필드 검증을 수행하므로 build_second_review_decision()의 완전한 결과를 전달해야 한다. """ data = _load_fixture("dev6_old_gemini_sha.json") inp = _inp_from_fixture(data) decision = build_second_review_decision( task_id=inp.task_id, pr_number=inp.pr_number, old_head_sha=inp.old_head_sha, current_head_sha=inp.current_head_sha, latest_gemini_commit_id=inp.latest_gemini_commit_id or "", gemini_fresh_on_current_head=False, stale_reason="SHA_MISMATCH_AFTER_FOLLOW_UP_COMMIT", ci_gate_reason="GEMINI_EVIDENCE_STALE", owner_trigger_required=True, owner_trigger_result="NONE", fresh_detected_at=None, ci_rerun_required=False, final_decision="WAITING", ) state = SecondReviewState.SECOND_REVIEW_TRIGGER_REQUIRED paths = emit_phase2_markers(inp.task_id, state, decision, marker_dir=tmp_path) # 최소 2개 경로 반환 (state marker + decision json) assert len(paths) >= 2, f"marker 경로 최소 2개 기대, got {len(paths)}" # state marker 존재 확인 trigger_marker = tmp_path / "task-2565.second-review-owner-trigger-requested" assert trigger_marker.exists(), ( f"TRIGGER_REQUIRED marker 미생성: {trigger_marker}" ) # decision JSON 존재 + 내용 확인 decision_json = tmp_path / "task-2565.second-review-decision.json" assert decision_json.exists(), f"second-review-decision.json 미생성: {decision_json}" content = json.loads(decision_json.read_text(encoding="utf-8")) assert content["task_id"] == inp.task_id, ( f"decision JSON task_id 불일치: {content['task_id']!r}" ) assert content["state"] == SecondReviewState.SECOND_REVIEW_TRIGGER_REQUIRED.value # ─── Phase 3: CI rerun + pre-merge guard ───────────────────────────────────── class TestPhase3CIRerunAndPreMergeGuard: """Phase 3 — CI rerun 자동화 + MERGE_READY pre-merge guard 테스트 (task-2565 §5 Phase 3).""" def test_fresh_evidence_after_ci_failure_reruns_failed_jobs(self) -> None: """race fixture load → should_rerun_failed_ci_jobs → True. auto_rerun_failed_ci_jobs with mock rerun_callable → result="POSTED". mock callable 호출 인자 (pr, failed_jobs) 검증. """ data = _load_fixture("race_fresh_after_ci_fail.json") ci_inp_data = data["ci_rerun_input"] inp = CIRerunInput( pr_number=ci_inp_data["pr_number"], current_head_sha=ci_inp_data["current_head_sha"], gemini_fresh_on_current_head=ci_inp_data["gemini_fresh_on_current_head"], ci_gate_failed_with_stale_reason=ci_inp_data["ci_gate_failed_with_stale_reason"], current_head_unchanged_since_failure=ci_inp_data["current_head_unchanged_since_failure"], unresolved_thread_count=ci_inp_data["unresolved_thread_count"], failed_job_identifiers=tuple(ci_inp_data["failed_job_identifiers"]), previous_rerun_attempts=ci_inp_data["previous_rerun_attempts"], ) # should_rerun → True (4조건 충족) assert should_rerun_failed_ci_jobs(inp) is True, ( "race fixture는 4조건 모두 충족해야 함" ) assert data["expected_should_rerun"] is True # mock rerun_callable → POSTED mock_callable = MagicMock(return_value={"status": "rerun triggered"}) result = auto_rerun_failed_ci_jobs(inp, rerun_callable=mock_callable) assert result["result"] == "POSTED", ( f"result 기대값 POSTED, 실제값: {result['result']!r}" ) # mock callable이 올바른 인자로 호출됐는지 검증 mock_callable.assert_called_once_with( inp.pr_number, inp.failed_job_identifiers, ) assert result["decision"]["long_polling_used"] is False, ( "long_polling_used 항상 False 박제 위반" ) def test_merge_ready_report_only_commit_blocked(self) -> None: """dev5 fixture load → is_merge_ready → True. classify_pre_merge_commit → ("report_only", False). build_pre_merge_block_decision → allowed=False, redirect=True. """ data = _load_fixture("dev5_merge_ready_report_only.json") # MergeReadyInput 구성 mr_data = data["merge_ready_input"] mr_inp = MergeReadyInput( ci_all_green=mr_data["ci_all_green"], gemini_fresh_on_current_head=mr_data["gemini_fresh_on_current_head"], unresolved_thread_count=mr_data["unresolved_thread_count"], merge_state_status=mr_data["merge_state_status"], effective_diff_matches_expected_files=mr_data["effective_diff_matches_expected_files"], forbidden_path_count=mr_data["forbidden_path_count"], ) # is_merge_ready → True assert is_merge_ready(mr_inp) is True, ( "dev5 fixture는 6조건 모두 충족해야 함" ) # PreMergeCommitInput 구성 pm_data = data["pre_merge_commit_input"] pm_inp = PreMergeCommitInput( merge_ready=pm_data["merge_ready"], changed_files=tuple(pm_data["changed_files"]), commit_message_hint=pm_data["commit_message_hint"], only_touches_report_files=pm_data["only_touches_report_files"], ) # classify_pre_merge_commit → ("report_only", False) change_type, allowed = classify_pre_merge_commit(pm_inp) assert change_type == data["expected_change_type"], ( f"change_type 기대값 {data['expected_change_type']!r}, 실제값: {change_type!r}" ) assert allowed is data["expected_pre_merge_commit_allowed"], ( f"allowed 기대값 {data['expected_pre_merge_commit_allowed']}, 실제값: {allowed}" ) # build_pre_merge_block_decision → allowed=False, redirect=True decision = build_pre_merge_block_decision(pm_inp) assert decision["pre_merge_commit_allowed"] is False, ( "merge_ready + report_only → pre_merge_commit_allowed=False 기대" ) assert decision["redirect_to_post_merge_evidence"] is True, ( "merge_ready + report_only → redirect_to_post_merge_evidence=True 기대" ) assert decision["change_type"] == "report_only" def test_post_merge_evidence_redirect_allowed(self) -> None: """blocked decision → post_merge_evidence_redirect_payload. redirect_to 필드 존재 + original_change_type=report_only. """ data = _load_fixture("dev5_merge_ready_report_only.json") pm_data = data["pre_merge_commit_input"] pm_inp = PreMergeCommitInput( merge_ready=pm_data["merge_ready"], changed_files=tuple(pm_data["changed_files"]), commit_message_hint=pm_data["commit_message_hint"], only_touches_report_files=pm_data["only_touches_report_files"], ) blocked_decision = build_pre_merge_block_decision(pm_inp) redirect_payload = post_merge_evidence_redirect_payload(blocked_decision) assert "redirect_to" in redirect_payload, ( "redirect_payload에 redirect_to 필드 필수" ) assert redirect_payload["original_change_type"] == "report_only", ( f"original_change_type 기대값 'report_only', 실제값: {redirect_payload['original_change_type']!r}" ) assert redirect_payload["redirect_to"] in ( "post-merge smoke", "reconcile evidence", "lifecycle marker" ), ( f"redirect_to 허용 값 위반: {redirect_payload['redirect_to']!r}" ) assert isinstance(redirect_payload["original_changed_files"], list) def test_stale_ci_merge_forbidden(self) -> None: """MergeReadyInput with ci_all_green=False → is_merge_ready → False. 또는 gemini_fresh_on_current_head=False → False. stale CI 또는 stale Gemini 상태에서는 merge 불허. """ # ci_all_green=False → False inp_stale_ci = MergeReadyInput( ci_all_green=False, gemini_fresh_on_current_head=True, unresolved_thread_count=0, merge_state_status="CLEAN", effective_diff_matches_expected_files=True, forbidden_path_count=0, ) assert is_merge_ready(inp_stale_ci) is False, ( "ci_all_green=False → is_merge_ready=False 기대" ) # gemini_fresh_on_current_head=False → False inp_stale_gemini = MergeReadyInput( ci_all_green=True, gemini_fresh_on_current_head=False, unresolved_thread_count=0, merge_state_status="CLEAN", effective_diff_matches_expected_files=True, forbidden_path_count=0, ) assert is_merge_ready(inp_stale_gemini) is False, ( "gemini_fresh_on_current_head=False → is_merge_ready=False 기대" ) # merge_state_status="BEHIND" → False inp_behind = MergeReadyInput( ci_all_green=True, gemini_fresh_on_current_head=True, unresolved_thread_count=0, merge_state_status="BEHIND", effective_diff_matches_expected_files=True, forbidden_path_count=0, ) assert is_merge_ready(inp_behind) is False, ( "merge_state_status=BEHIND → is_merge_ready=False 기대" ) def test_ci_rerun_cap_prevents_infinite_loop(self) -> None: """CIRerunInput with previous_rerun_attempts=3 (cap). should_rerun_failed_ci_jobs → False. auto_rerun_failed_ci_jobs → result="CAP_EXCEEDED". """ assert MAX_CI_RERUN_PER_HEAD == 3, ( f"MAX_CI_RERUN_PER_HEAD must be 3, got {MAX_CI_RERUN_PER_HEAD}" ) inp = CIRerunInput( pr_number=99002, current_head_sha="abc12345", gemini_fresh_on_current_head=True, ci_gate_failed_with_stale_reason=True, current_head_unchanged_since_failure=True, unresolved_thread_count=0, failed_job_identifiers=("gemini-review-gate",), previous_rerun_attempts=3, # cap 도달 ) # should_rerun → False (cap 초과) assert should_rerun_failed_ci_jobs(inp) is False, ( "previous_rerun_attempts=3 (cap) → should_rerun=False 기대" ) # auto_rerun → CAP_EXCEEDED mock_callable = MagicMock(return_value={"status": "triggered"}) result = auto_rerun_failed_ci_jobs(inp, rerun_callable=mock_callable) assert result["result"] == "CAP_EXCEEDED", ( f"result 기대값 CAP_EXCEEDED, 실제값: {result['result']!r}" ) # cap 초과 시 callable 미호출 확인 mock_callable.assert_not_called() assert result["decision"]["long_polling_used"] is False def test_ci_rerun_skipped_without_fresh_evidence(self) -> None: """gemini_fresh_on_current_head=False. should_rerun_failed_ci_jobs → False. auto_rerun_failed_ci_jobs → result="SKIPPED". """ inp = CIRerunInput( pr_number=99003, current_head_sha="def67890", gemini_fresh_on_current_head=False, # fresh evidence 없음 ci_gate_failed_with_stale_reason=True, current_head_unchanged_since_failure=True, unresolved_thread_count=0, failed_job_identifiers=("gemini-review-gate",), previous_rerun_attempts=0, ) # should_rerun → False (gemini_fresh 없음) assert should_rerun_failed_ci_jobs(inp) is False, ( "gemini_fresh=False → should_rerun=False 기대" ) # auto_rerun → SKIPPED mock_callable = MagicMock(return_value={"status": "triggered"}) result = auto_rerun_failed_ci_jobs(inp, rerun_callable=mock_callable) assert result["result"] == "SKIPPED", ( f"result 기대값 SKIPPED, 실제값: {result['result']!r}" ) # SKIPPED 시 callable 미호출 mock_callable.assert_not_called() def test_merge_ready_code_fix_commit_allowed(self) -> None: """commit message hint = "fix:", changed_files includes anu_v2/*.py. classify_pre_merge_commit → ("code_fix" or "test_fix", True) — 허용. merge_ready=True이더라도 코드 수정 commit은 허용되어야 함. """ inp = PreMergeCommitInput( merge_ready=True, changed_files=( "anu_v2/second_review_recovery.py", "anu_v2/merge_queue_executor.py", ), commit_message_hint="fix: CI rerun 조건 버그 수정", only_touches_report_files=False, ) change_type, allowed = classify_pre_merge_commit(inp) assert allowed is True, ( f"코드 변경 commit은 허용되어야 함 — allowed 기대값 True, 실제값: {allowed}" ) assert change_type in ("code_fix", "test_fix", "evidence_required"), ( f"change_type 기대값 code_fix/test_fix/evidence_required, 실제값: {change_type!r}" ) # ─── Gemini medium Fix 3건 (PR #119) ──────────────────────────────────────── class TestGeminiMediumFixes: """PR #119 Gemini medium 3건 Fix 검증 (task-2568+1).""" def test_dedupe_entry_key_handles_none_pr_number(self) -> None: """entry의 pr_number=None 일 때 'pr' fallback이 사용되어 dedupe 매칭 성공. Fix 1: entry_key = f"{entry.get('pr_number') or entry.get('pr') or ''}+..." pr_number=None → falsy → pr=119 fallback → key "119+abc" 생성 → 매칭 True. """ data = _load_fixture("dev6_old_gemini_sha.json") audit_entries: tuple[dict, ...] = ( { "pr_number": None, # None — 이전 코드라면 "None+abc" 생성 → 매칭 실패 "pr": 119, "head_sha": "abc", "result": "POSTED", }, ) inp = SecondReviewInput( task_id=data["task_id"], pr_number=119, old_head_sha=data["old_head_sha"], current_head_sha="abc", latest_gemini_commit_id=data.get("latest_gemini_commit_id"), ci_gate_failure_reason=data.get("ci_gate_failure_reason"), unresolved_thread_outdated=data["unresolved_thread_outdated"], follow_up_commit_detected=data["follow_up_commit_detected"], elapsed_since_follow_up_seconds=data["elapsed_since_follow_up_seconds"], owner_trigger_audit_entries=audit_entries, ) # Fix 1 적용 후: None → falsy → "pr"=119 fallback → entry_key="119+abc" → 매칭 True assert is_owner_trigger_deduped(inp) is True, ( "pr_number=None 이면 'pr' fallback 사용해 키 '119+abc' 생성 → dedupe=True 기대" ) def test_write_marker_cleanup_on_os_replace_failure( self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """os.replace 실패 시 임시 파일(.tmp) 잔류 0건 보장. Fix 2: try/finally + tmp_path.unlink(missing_ok=True) 로 cleanup 보장. """ import os as _os def _raise_on_replace(*args: object, **kwargs: object) -> None: del args, kwargs raise OSError("injected os.replace failure") monkeypatch.setattr(_os, "replace", _raise_on_replace) marker_name = "test-cleanup.json" with pytest.raises(OSError, match="injected os.replace failure"): write_marker(marker_name, payload={"x": 1}, marker_dir=tmp_path) # tmp 파일 잔류 0건 검증 tmp_files = list(tmp_path.glob(f".{marker_name}.*.tmp")) assert len(tmp_files) == 0, ( f"OSError 발생 후 .tmp 파일이 잔류해서는 안 됨 — 잔류 파일: {tmp_files}" ) def test_detect_change_type_does_not_match_non_memory_lifecycle_paths(self) -> None: """anu_v2/lifecycle/utils.py 는 memory/ 외부 경로 → lifecycle_files 미매칭 → report_only 아님. Fix 3: (f.startswith("memory/") and "/lifecycle/" in f) 조건으로 비-memory 경로 제외. """ inp = PreMergeCommitInput( merge_ready=True, changed_files=("anu_v2/lifecycle/utils.py",), commit_message_hint="fix: anu_v2 lifecycle util", only_touches_report_files=False, ) change_type, allowed = classify_pre_merge_commit(inp) assert change_type != "report_only", ( f"anu_v2/lifecycle/utils.py 는 memory/ 외부 경로 → report_only 가 아니어야 함, " f"실제값: {change_type!r}" ) assert allowed is True, ( "code_fix 분류면 allowed=True 기대 (merge_ready commit 허용)" ) # ─── Gemini Fresh Review 3건 회귀 테스트 (task-2568+1, PR #119 attempt-3) ───── class TestGeminiFreshReview3Regressions: """PR #119 attempt-3 Gemini fresh review 3건 회귀 테스트 (task-2568+1). Finding 1 (HIGH): decision_payload schema 누락 시 ValueError Finding 2 (MEDIUM): lifecycle 경로 다양성 — 기능 회귀 없음 Finding 3 (MEDIUM): fcntl ImportError → NotImplementedError (Linux-only guard) """ # ── Finding 1: emit_phase2_markers schema 누락 필드 ValueError ───────────── def test_emit_phase2_markers_raises_valueerror_on_missing_schema_fields( self, tmp_path: Path ) -> None: """decision에 필수 schema 필드가 누락되면 ValueError raise. Finding 1 fix: emit_phase2_markers 내부에서 _SCHEMA_V1_REQUIRED 대비 누락 필드를 검사하고 ValueError를 raise한다. """ # 의도적으로 old_head_sha, current_head_sha, latest_gemini_commit_id 누락 partial_decision: dict = { "triggered": True, "result": "POSTED", "dedupe_key": "107+cee55afe", } state = SecondReviewState.SECOND_REVIEW_TRIGGER_REQUIRED with pytest.raises(ValueError, match="누락 필드"): emit_phase2_markers("task-2565", state, partial_decision, marker_dir=tmp_path) def test_emit_phase2_markers_passes_with_complete_decision( self, tmp_path: Path ) -> None: """build_second_review_decision()의 완전한 결과를 넘기면 ValueError 없이 정상 동작. Finding 1 fix의 happy-path 확인: 완전한 schema dict이면 검증 통과. """ complete_decision = build_second_review_decision( task_id="task-2565", pr_number=119, old_head_sha="d251399c", current_head_sha="cee55afe", latest_gemini_commit_id="d251399c", gemini_fresh_on_current_head=False, stale_reason="SHA_MISMATCH_AFTER_FOLLOW_UP_COMMIT", ci_gate_reason="GEMINI_EVIDENCE_STALE", owner_trigger_required=True, owner_trigger_result="POSTED", fresh_detected_at=None, ci_rerun_required=False, final_decision="WAITING", ) state = SecondReviewState.SECOND_REVIEW_TRIGGER_REQUIRED # ValueError 없이 통과해야 함 paths = emit_phase2_markers("task-2565", state, complete_decision, marker_dir=tmp_path) assert len(paths) >= 1, "marker 경로가 최소 1개 이상 반환되어야 함" # ── Finding 2: lifecycle 경로 다양성 — 기능 회귀 검증 ─────────────────────── def test_classify_pre_merge_commit_deep_lifecycle_path_is_report_only(self) -> None: """memory/somedir/lifecycle/event.json → lifecycle_files 매칭 → report_only. Finding 2 fix 후에도 memory/ 하위 /lifecycle/ 포함 경로는 정상 매칭됨을 확인. 중복 조건 제거가 기능 회귀를 유발하지 않음을 검증. """ inp = PreMergeCommitInput( merge_ready=True, changed_files=("memory/somedir/lifecycle/event.json",), commit_message_hint="lifecycle: add event", only_touches_report_files=False, ) change_type, _allowed = classify_pre_merge_commit(inp) assert change_type == "report_only", ( f"memory/somedir/lifecycle/event.json 는 lifecycle 경로 → report_only 기대, " f"실제값: {change_type!r}" ) assert _allowed is False, ( "report_only 분류면 allowed=False 기대 (lifecycle 경로 커밋 차단)" ) def test_classify_pre_merge_commit_direct_lifecycle_path_is_report_only(self) -> None: """memory/lifecycle/event.json → lifecycle_files 매칭 → report_only. Finding 2: f.startswith("memory/lifecycle/") 조건 제거 후에도 (f.startswith("memory/") and "/lifecycle/" in f) 조건으로 올바르게 매칭됨. """ inp = PreMergeCommitInput( merge_ready=True, changed_files=("memory/lifecycle/event.json",), commit_message_hint="lifecycle: direct path", only_touches_report_files=False, ) change_type, _allowed = classify_pre_merge_commit(inp) assert change_type == "report_only", ( f"memory/lifecycle/event.json 는 lifecycle 경로 → report_only 기대, " f"실제값: {change_type!r}" ) assert _allowed is False, ( "report_only 분류면 allowed=False 기대 (lifecycle 경로 커밋 차단)" ) # ── Finding 3: fcntl ImportError guard (Linux-only) ───────────────────────── def test_append_owner_trigger_audit_linux_import_succeeds( self, tmp_path: Path ) -> None: """Linux 환경에서 append_owner_trigger_audit 정상 동작 확인. Finding 3 fix: try/except ImportError → NotImplementedError 추가. Linux self-hosted ANU CI에서는 fcntl import가 성공하므로 NotImplementedError 없이 audit 항목이 기록되어야 함. """ from anu_v2.second_review_recovery import append_owner_trigger_audit audit_file = tmp_path / "owner_trigger_audit.jsonl" entry = { "timestamp": "2026-05-13T00:00:00Z", "pr_number": 119, "head_sha": "cee55afe", "result": "POSTED", "source": "test_finding3", } result_path = append_owner_trigger_audit(entry, audit_path=audit_file) assert result_path.exists(), f"audit 파일이 생성되어야 함: {result_path}" lines = result_path.read_text(encoding="utf-8").strip().splitlines() assert len(lines) == 1, f"audit 항목 1건 기대, 실제: {len(lines)}건" recorded = json.loads(lines[0]) assert recorded["pr_number"] == 119 assert recorded["result"] == "POSTED"