"""tests/regression/test_automation_contracts_2509_plus_2.py — task-2509+2 회귀 테스트. 루(Lugh, 개발3팀 백엔드) 작성. 대상: utils/automation_contracts.py 14건: 1 CriticalEscalationType 멤버 정확히 7개 + 이름 검증 2 CriticalEscalationType 알 수 없는 값 → ValueError 3 dataclass 8종 모두 JSON 직렬화 가능 4 AutomationDecision: critical 있으면 requires_chair=True 강제 5 AutomationDecision: critical=None + requires_chair=False 정상 6 ReplacementResult: success=False 시 failure_reason 필수 7 ReviewGateStatus: quota + fallback_passed 조합 보존 8 GeminiStatus.GEMINI_SCOPE_EXPANSION → EscalationPacket 연결 가능 9 SmokeResult 실패 → EscalationPacket.POST_MERGE_SMOKE_FAILED evidence 정상 10 QueueAuditRecord 필수 필드 asdict 후 보존 11 AutoMergeResult: merged=True 시 merge_commit 필수 12 EscalationPacket: string 입력 자동 변환 / 알 수 없는 string → ValueError 13 merge_queue_executor import 호환성 (subprocess) 14 import smoke (subprocess) """ from __future__ import annotations import dataclasses import json import subprocess import sys from pathlib import Path import pytest # workspace root → sys.path 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.automation_contracts import ( # noqa: E402 # pyright: ignore[reportMissingImports] AutoMergeResult, AutomationDecision, CriticalEscalationType, EscalationPacket, FallbackReviewResult, GeminiStatus, GeminiTriageResult, QueueAuditRecord, ReplacementResult, ReviewGateStatus, RiskLevel, SmokeResult, to_json, ) # --------------------------------------------------------------------------- # 공용 샘플 빌더 # --------------------------------------------------------------------------- def _make_automation_decision(**kw) -> AutomationDecision: defaults = dict( decision="AUTO_MERGE_ALLOWED", reason_codes=[], critical_escalation_type=None, auto_handled=True, requires_chair=False, audit={}, ) defaults.update(kw) return AutomationDecision(**defaults) def _make_review_gate_status(**kw) -> ReviewGateStatus: defaults = dict( gemini_status=GeminiStatus.GEMINI_COMPLETED, unresolved_threads=0, fallback_review_used=False, fallback_review_passed=False, review_gate_passed=True, reason="ok", ) defaults.update(kw) return ReviewGateStatus(**defaults) def _make_smoke_result(**kw) -> SmokeResult: defaults = dict( command="pytest", passed=True, exit_code=0, stdout_tail="", stderr_tail="", failure_reason=None, ) defaults.update(kw) return SmokeResult(**defaults) # --------------------------------------------------------------------------- # 1. CriticalEscalationType 멤버 정확히 7개 # --------------------------------------------------------------------------- EXPECTED_CRITICAL_MEMBERS = { "FORBIDDEN_PATH_INTRUSION", "REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF", "GEMINI_REAL_BUG_REQUIRES_SCOPE_EXPANSION", "BLOCK_OVERRIDE_REQUIRED_OR_REASON_INSUFFICIENT", "DEPENDENCY_CYCLE_OR_SERIAL_ONLY_COLLISION", "REPLACEMENT_PR_FAILED", "POST_MERGE_SMOKE_FAILED", } def test_critical_escalation_type_exact_seven_members(): members = list(CriticalEscalationType) assert len(members) == 7, f"Expected 7 members, got {len(members)}" actual_names = {m.name for m in members} assert actual_names == EXPECTED_CRITICAL_MEMBERS, ( f"Member name mismatch: {actual_names ^ EXPECTED_CRITICAL_MEMBERS}" ) # --------------------------------------------------------------------------- # 2. CriticalEscalationType 알 수 없는 값 → ValueError # --------------------------------------------------------------------------- def test_critical_escalation_type_no_extra_members(): with pytest.raises(ValueError): CriticalEscalationType("UNKNOWN_CRITICAL") # --------------------------------------------------------------------------- # 3. dataclass 8종 모두 JSON 직렬화 가능 # --------------------------------------------------------------------------- def test_all_dataclasses_json_serializable(): decision = _make_automation_decision() gate = _make_review_gate_status() smoke = _make_smoke_result() instances = [ decision, gate, FallbackReviewResult( used=False, passed=False, risk_level=RiskLevel.LOW, checks={}, reason="ok", ), ReplacementResult( source_pr=1, replacement_pr=None, original_pr_preserved=True, expected_files=[], effective_diff_files=[], forbidden_paths=[], success=True, failure_reason=None, ), GeminiTriageResult( status=GeminiStatus.GEMINI_COMPLETED, false_positive_count=0, style_only_count=0, real_bug_small_count=0, scope_expansion_count=0, unresolved_count=0, actions_taken=[], ), smoke, QueueAuditRecord( task_id="T-001", pr_number=42, queue_position=0, head_sha="abc123", base_sha="def456", decision=decision, checks={}, review_gate=gate, smoke=None, critical_escalation=None, timestamp="2026-05-08T00:00:00Z", ), AutoMergeResult( merged=False, merge_commit=None, smoke_result=None, following_prs_rechecked=[], critical_escalation=None, ), ] for obj in instances: d = dataclasses.asdict(obj) serialized = json.dumps(d) assert isinstance(serialized, str), ( f"{type(obj).__name__} JSON 직렬화 실패" ) # Enum 필드가 string으로 직렬화되는지 확인 reparsed = json.loads(serialized) assert isinstance(reparsed, dict) # to_json helper 경로 동등성 검증 helper_serialized = to_json(obj) assert json.loads(helper_serialized) == reparsed, ( f"{type(obj).__name__} to_json mismatch with dataclasses.asdict+json.dumps" ) # --------------------------------------------------------------------------- # 4. AutomationDecision: critical 있으면 requires_chair=True 강제 # --------------------------------------------------------------------------- def test_automation_decision_critical_requires_chair_true(): # requires_chair=False → ValueError with pytest.raises(ValueError): _make_automation_decision( critical_escalation_type=CriticalEscalationType.FORBIDDEN_PATH_INTRUSION, requires_chair=False, ) # requires_chair=True → 정상 obj = _make_automation_decision( critical_escalation_type=CriticalEscalationType.FORBIDDEN_PATH_INTRUSION, requires_chair=True, ) assert obj.requires_chair is True assert obj.critical_escalation_type == CriticalEscalationType.FORBIDDEN_PATH_INTRUSION # --------------------------------------------------------------------------- # 5. AutomationDecision: critical=None + requires_chair=False 정상 # --------------------------------------------------------------------------- def test_automation_decision_no_critical_allows_chair_false(): obj = _make_automation_decision( critical_escalation_type=None, requires_chair=False, ) assert obj.critical_escalation_type is None assert obj.requires_chair is False # --------------------------------------------------------------------------- # 6. ReplacementResult: success=False 시 failure_reason 필수 # --------------------------------------------------------------------------- def test_replacement_result_failure_requires_reason(): # failure_reason=None → ValueError with pytest.raises(ValueError): ReplacementResult( source_pr=1, replacement_pr=None, original_pr_preserved=True, expected_files=[], effective_diff_files=[], forbidden_paths=[], success=False, failure_reason=None, ) # failure_reason="" → ValueError with pytest.raises(ValueError): ReplacementResult( source_pr=1, replacement_pr=None, original_pr_preserved=True, expected_files=[], effective_diff_files=[], forbidden_paths=[], success=False, failure_reason="", ) # failure_reason 있음 → 정상 obj = ReplacementResult( source_pr=1, replacement_pr=None, original_pr_preserved=True, expected_files=[], effective_diff_files=[], forbidden_paths=[], success=False, failure_reason="diff contamination", ) assert obj.failure_reason == "diff contamination" # success=True, failure_reason=None → 정상 obj2 = ReplacementResult( source_pr=1, replacement_pr=2, original_pr_preserved=True, expected_files=[], effective_diff_files=[], forbidden_paths=[], success=True, failure_reason=None, ) assert obj2.success is True # --------------------------------------------------------------------------- # 7. ReviewGateStatus: quota + fallback_passed 조합 보존 # --------------------------------------------------------------------------- def test_review_gate_quota_with_fallback_passed(): gate = ReviewGateStatus( gemini_status=GeminiStatus.GEMINI_UNAVAILABLE_QUOTA, unresolved_threads=0, fallback_review_used=True, fallback_review_passed=True, review_gate_passed=True, reason="fallback used due to quota", ) assert gate.gemini_status == GeminiStatus.GEMINI_UNAVAILABLE_QUOTA assert gate.fallback_review_passed is True assert gate.review_gate_passed is True # --------------------------------------------------------------------------- # 8. GeminiStatus.GEMINI_SCOPE_EXPANSION → EscalationPacket 연결 가능 # --------------------------------------------------------------------------- def test_gemini_scope_expansion_to_critical_3(): # GEMINI_SCOPE_EXPANSION → CriticalEscalationType.GEMINI_REAL_BUG_REQUIRES_SCOPE_EXPANSION 매핑 escalation_type = CriticalEscalationType.GEMINI_REAL_BUG_REQUIRES_SCOPE_EXPANSION assert escalation_type.value == "GEMINI_REAL_BUG_REQUIRES_SCOPE_EXPANSION" packet = EscalationPacket( task_id="T-002", pr_number=100, escalation_type=CriticalEscalationType.GEMINI_REAL_BUG_REQUIRES_SCOPE_EXPANSION, reason="Gemini detected scope expansion", why_auto_cannot_continue="scope beyond original PR", safe_options=["reject", "expand scope manually"], recommended_option="expand scope manually", evidence={"gemini_status": GeminiStatus.GEMINI_SCOPE_EXPANSION}, ) assert packet.escalation_type == CriticalEscalationType.GEMINI_REAL_BUG_REQUIRES_SCOPE_EXPANSION assert packet.escalation_type.value == "GEMINI_REAL_BUG_REQUIRES_SCOPE_EXPANSION" # --------------------------------------------------------------------------- # 9. SmokeResult 실패 → EscalationPacket.POST_MERGE_SMOKE_FAILED evidence 정상 # --------------------------------------------------------------------------- def test_smoke_failure_to_post_merge_critical(): smoke = SmokeResult( command="pytest tests/", passed=False, exit_code=1, stdout_tail="FAILED tests/test_foo.py", stderr_tail="", failure_reason="smoke test exit code 1", ) assert smoke.passed is False packet = EscalationPacket( task_id="T-003", pr_number=200, escalation_type=CriticalEscalationType.POST_MERGE_SMOKE_FAILED, reason="Post-merge smoke failed", why_auto_cannot_continue="smoke failure after merge", safe_options=["revert merge"], recommended_option="revert merge", evidence={"smoke_result": dataclasses.asdict(smoke)}, ) assert packet.escalation_type == CriticalEscalationType.POST_MERGE_SMOKE_FAILED assert "smoke_result" in packet.evidence assert packet.evidence["smoke_result"]["passed"] is False # --------------------------------------------------------------------------- # 10. QueueAuditRecord 필수 필드 asdict 후 보존 # --------------------------------------------------------------------------- def test_queue_audit_record_required_fields_preserved(): decision = _make_automation_decision() gate = _make_review_gate_status() record = QueueAuditRecord( task_id="T-007", pr_number=999, queue_position=3, head_sha="deadbeef", base_sha="cafebabe", decision=decision, checks={"ci": "pass"}, review_gate=gate, smoke=None, critical_escalation=None, timestamp="2026-05-08T12:00:00Z", ) d = dataclasses.asdict(record) assert d["task_id"] == "T-007" assert d["pr_number"] == 999 assert d["head_sha"] == "deadbeef" assert d["timestamp"] == "2026-05-08T12:00:00Z" # --------------------------------------------------------------------------- # 11. AutoMergeResult: merged=True 시 merge_commit 필수 # --------------------------------------------------------------------------- def test_auto_merge_result_merged_requires_merge_commit(): # merge_commit=None → ValueError with pytest.raises(ValueError): AutoMergeResult( merged=True, merge_commit=None, smoke_result=None, following_prs_rechecked=[], critical_escalation=None, ) # merge_commit="" → ValueError with pytest.raises(ValueError): AutoMergeResult( merged=True, merge_commit="", smoke_result=None, following_prs_rechecked=[], critical_escalation=None, ) # merge_commit 있음 → 정상 obj = AutoMergeResult( merged=True, merge_commit="abc123", smoke_result=None, following_prs_rechecked=[], critical_escalation=None, ) assert obj.merge_commit == "abc123" # merged=False, merge_commit=None → 정상 obj2 = AutoMergeResult( merged=False, merge_commit=None, smoke_result=None, following_prs_rechecked=[], critical_escalation=None, ) assert obj2.merged is False # --------------------------------------------------------------------------- # 12. EscalationPacket: string 입력 자동 변환 / 알 수 없는 string → ValueError # --------------------------------------------------------------------------- def test_escalation_packet_only_critical_enum(): # string으로 입력 → 자동 enum 변환 (7종 멤버명과 일치) packet = EscalationPacket( task_id="T-010", pr_number=50, escalation_type="FORBIDDEN_PATH_INTRUSION", # type: ignore[arg-type] reason="string input", why_auto_cannot_continue="test", safe_options=[], recommended_option="none", evidence={}, ) assert isinstance(packet.escalation_type, CriticalEscalationType) assert packet.escalation_type == CriticalEscalationType.FORBIDDEN_PATH_INTRUSION # 알 수 없는 string → ValueError with pytest.raises(ValueError): EscalationPacket( task_id="T-011", pr_number=51, escalation_type="UNKNOWN_STRING", # type: ignore[arg-type] reason="bad input", why_auto_cannot_continue="test", safe_options=[], recommended_option="none", evidence={}, ) # enum 인스턴스 직접 → 정상 packet2 = EscalationPacket( task_id="T-012", pr_number=52, escalation_type=CriticalEscalationType.POST_MERGE_SMOKE_FAILED, reason="enum input", why_auto_cannot_continue="test", safe_options=[], recommended_option="none", evidence={}, ) assert packet2.escalation_type == CriticalEscalationType.POST_MERGE_SMOKE_FAILED # --------------------------------------------------------------------------- # 13. merge_queue_executor import 호환성 # --------------------------------------------------------------------------- def test_merge_queue_executor_can_import_contracts(): repo_root = str(WORKSPACE) cmd = [ sys.executable, "-c", ( f"import sys; sys.path.insert(0, '{repo_root}'); " "from utils.automation_contracts import CriticalEscalationType; " "from utils.merge_queue_executor import *; " "print('ok')" ), ] result = subprocess.run(cmd, capture_output=True, text=True) assert result.returncode == 0, ( f"subprocess failed:\nstdout={result.stdout}\nstderr={result.stderr}" ) assert "ok" in result.stdout # --------------------------------------------------------------------------- # 14. import smoke # --------------------------------------------------------------------------- def test_pyright_or_import_smoke(): repo_root = str(WORKSPACE) cmd = [ sys.executable, "-c", ( f"import sys; sys.path.insert(0, '{repo_root}'); " "from utils.automation_contracts import (" "CriticalEscalationType, RiskLevel, GeminiStatus, " "AutomationDecision, ReviewGateStatus, FallbackReviewResult, " "ReplacementResult, GeminiTriageResult, SmokeResult, " "QueueAuditRecord, AutoMergeResult, EscalationPacket, to_json" "); print('ok')" ), ] result = subprocess.run(cmd, capture_output=True, text=True) assert result.returncode == 0, ( f"import smoke failed:\nstdout={result.stdout}\nstderr={result.stderr}" ) assert "ok" in result.stdout