"""anu_v2.tests.test_owner_trigger_pat_phase1_2553 — Phase 1 decision schema 단위 테스트. 회장 §명시 Phase 1 (task-2553): - `OwnerTriggerDecision` frozen dataclass schema 필드 검증 - `serialize_decision` JSON 직렬화 가능 + token 필드 0 - `write_decision_json` atomic write (`.tmp` → rename) - dedupe key = `{pr_number}#{head_sha}` — head 변경 시 자동 stale - `is_duplicate_trigger` jsonl audit lookup 본 회귀는 anu_v2/* 모듈만 import 한다 (one-way isolation). """ from __future__ import annotations import dataclasses import json import sys from pathlib import Path import pytest # workspace root → sys.path WORKSPACE_ROOT = Path(__file__).resolve().parents[2] if str(WORKSPACE_ROOT) not in sys.path: sys.path.insert(0, str(WORKSPACE_ROOT)) from anu_v2.owner_trigger_pat import ( # noqa: E402 DECISION_PASS, DECISION_REJECT, EVIDENCE_MISSING_FOR_CURRENT_HEAD, OUTCOME_OK, OwnerTriggerDecision, is_duplicate_trigger, make_dedupe_key, serialize_decision, write_decision_json, ) # ─── A. decision schema 필드 / 상수 검증 ───────────────────────────────────── def test_phase1_decision_constants() -> None: assert DECISION_PASS == "PASS" assert DECISION_REJECT == "REJECT" def test_phase1_decision_is_frozen() -> None: """frozen dataclass — 인스턴스 mutation 차단 (audit trail 박제 무결성).""" decision = OwnerTriggerDecision( pr_number=81, head_sha="deadbeef", decision=DECISION_PASS, reason="ok", gemini_evidence_state=EVIDENCE_MISSING_FOR_CURRENT_HEAD, queue_position=0, dedupe_key="81#deadbeef", ts="2026-05-11T00:00:00+00:00", ) with pytest.raises(dataclasses.FrozenInstanceError): decision.decision = DECISION_REJECT # type: ignore[misc] def test_phase1_decision_has_required_fields() -> None: fields = {f.name for f in dataclasses.fields(OwnerTriggerDecision)} assert fields == { "pr_number", "head_sha", "decision", "reason", "gemini_evidence_state", "queue_position", "dedupe_key", "ts", } # ─── B. serialize_decision ─────────────────────────────────────────────────── def test_phase1_serialize_decision_returns_dict_with_all_fields() -> None: decision = OwnerTriggerDecision( pr_number=81, head_sha="abc123", decision=DECISION_PASS, reason="all_gates_pass", gemini_evidence_state=EVIDENCE_MISSING_FOR_CURRENT_HEAD, queue_position=0, dedupe_key="81#abc123", ts="2026-05-11T01:23:45+00:00", ) out = serialize_decision(decision) assert isinstance(out, dict) assert out["pr_number"] == 81 assert out["head_sha"] == "abc123" assert out["decision"] == DECISION_PASS assert out["dedupe_key"] == "81#abc123" def test_phase1_serialize_decision_is_json_serializable() -> None: decision = OwnerTriggerDecision( pr_number=99, head_sha="cafefeed", decision=DECISION_REJECT, reason="duplicate_trigger", gemini_evidence_state=EVIDENCE_MISSING_FOR_CURRENT_HEAD, queue_position=0, dedupe_key="99#cafefeed", ts="2026-05-11T01:00:00+00:00", ) payload = serialize_decision(decision) # json.dumps 가 raise 없이 성공해야 함 text = json.dumps(payload, sort_keys=True) parsed = json.loads(text) assert parsed["pr_number"] == 99 def test_phase1_serialize_decision_no_token_fields() -> None: """decision schema 에는 토큰 필드가 구조적으로 없음 — 박제 시 token raw 노출 0.""" decision = OwnerTriggerDecision( pr_number=1, head_sha="x", decision=DECISION_PASS, reason="r", gemini_evidence_state=EVIDENCE_MISSING_FOR_CURRENT_HEAD, queue_position=0, dedupe_key="1#x", ts="t", ) payload = serialize_decision(decision) keys_lower = {str(k).lower() for k in payload.keys()} forbidden_hints = {"token", "pat", "secret", "authorization", "ghp_", "ghs_"} for hint in forbidden_hints: assert not any(hint in k for k in keys_lower) # ─── C. write_decision_json atomic ─────────────────────────────────────────── def test_phase1_write_decision_json_atomic_write(tmp_path: Path) -> None: target = tmp_path / "subdir" / "owner_trigger_decision.json" payload = { "pr_number": 81, "head_sha": "deadbeef", "decision": DECISION_PASS, "reason": "ok", "gemini_evidence_state": EVIDENCE_MISSING_FOR_CURRENT_HEAD, "queue_position": 0, "dedupe_key": "81#deadbeef", "ts": "2026-05-11T00:00:00+00:00", } write_decision_json(payload, target) assert target.exists() # tmp 파일은 rename 으로 사라져야 함 assert not target.with_suffix(target.suffix + ".tmp").exists() # 내용 검증 parsed = json.loads(target.read_text(encoding="utf-8")) assert parsed["pr_number"] == 81 assert parsed["decision"] == DECISION_PASS def test_phase1_write_decision_json_overwrite_existing(tmp_path: Path) -> None: """이미 파일이 있어도 새 PASS/REJECT 결정으로 덮어쓰기 가능.""" target = tmp_path / "decision.json" target.write_text("OLD CONTENT", encoding="utf-8") payload = {"decision": DECISION_REJECT, "reason": "evidence_not_missing"} write_decision_json(payload, target) parsed = json.loads(target.read_text(encoding="utf-8")) assert parsed["decision"] == DECISION_REJECT # ─── D. dedupe key + lookup ────────────────────────────────────────────────── def test_phase1_make_dedupe_key_format() -> None: assert make_dedupe_key(81, "deadbeef") == "81#deadbeef" # 새 head_sha 면 다른 dedupe_key (update-branch 후 stale 자동 처리). assert make_dedupe_key(81, "newshax") != make_dedupe_key(81, "deadbeef") def test_phase1_is_duplicate_trigger_returns_false_when_no_audit_file(tmp_path: Path) -> None: audit_path = tmp_path / "owner_trigger_audit.jsonl" assert is_duplicate_trigger(audit_path, "81#deadbeef") is False def test_phase1_is_duplicate_trigger_returns_true_for_matching_ok_entry(tmp_path: Path) -> None: audit_path = tmp_path / "owner_trigger_audit.jsonl" audit_path.write_text( json.dumps({ "dedupe_key": "81#deadbeef", "outcome": OUTCOME_OK, "ts": "2026-05-11T00:00:00+00:00", }) + "\n", encoding="utf-8", ) assert is_duplicate_trigger(audit_path, "81#deadbeef") is True def test_phase1_is_duplicate_trigger_returns_false_for_different_dedupe_key(tmp_path: Path) -> None: audit_path = tmp_path / "owner_trigger_audit.jsonl" audit_path.write_text( json.dumps({"dedupe_key": "81#oldsha", "outcome": OUTCOME_OK}) + "\n", encoding="utf-8", ) # 새 head_sha 는 stale 처리되어 새 trigger 허용 assert is_duplicate_trigger(audit_path, "81#newsha") is False def test_phase1_is_duplicate_trigger_ignores_rejected_entries(tmp_path: Path) -> None: """REJECT/failed outcome 은 dedupe 로 카운트되지 않음 (재시도 가능).""" audit_path = tmp_path / "owner_trigger_audit.jsonl" audit_path.write_text( json.dumps({"dedupe_key": "81#deadbeef", "outcome": "rejected"}) + "\n" + json.dumps({"dedupe_key": "81#deadbeef", "outcome": "failed"}) + "\n", encoding="utf-8", ) assert is_duplicate_trigger(audit_path, "81#deadbeef") is False def test_phase1_is_duplicate_trigger_skips_malformed_lines(tmp_path: Path) -> None: """깨진 jsonl 라인은 skip — 안전 fallback.""" audit_path = tmp_path / "owner_trigger_audit.jsonl" audit_path.write_text( "this is not json\n" + json.dumps({"dedupe_key": "81#deadbeef", "outcome": OUTCOME_OK}) + "\n" + "{broken\n", encoding="utf-8", ) assert is_duplicate_trigger(audit_path, "81#deadbeef") is True def test_phase1_dedupe_key_in_decision(tmp_path: Path) -> None: """decision dataclass 의 dedupe_key 가 make_dedupe_key 와 동일 포맷.""" decision = OwnerTriggerDecision( pr_number=42, head_sha="cafebabe", decision=DECISION_PASS, reason="ok", gemini_evidence_state=EVIDENCE_MISSING_FOR_CURRENT_HEAD, queue_position=0, dedupe_key=make_dedupe_key(42, "cafebabe"), ts="2026-05-11T02:00:00+00:00", ) assert decision.dedupe_key == "42#cafebabe"