# -*- coding: utf-8 -*- """tests.regression.test_ci_watch_handoff_audit — task-2642. 회장 verbatim §9 (정책 spec) ANU 8 완료 보고 항목 박제 회귀: watcher 주체 / schedule_id / terminal state / 자동수렴 내역 / CI 상태 / merge-ready / callback 수신. Layer A / NO-CRON: subprocess / cokacdir / merge / cron 호출 0. audit 파일은 pytest tmp_path 로 격리. """ from __future__ import annotations import json import pytest from utils.ci_watch_handoff_audit import ( ALL_EVENTS, ALLOWED_AUDIT_KEYS, AUDIT_REL_PATH, AUDIT_SCHEMA, AuditError, CiWatchHandoffAudit, EVENT_AUTO_REMEDIATE, EVENT_CALLBACK_FIRED, EVENT_HANDOFF_RECEIVED, EVENT_OWNER_NUDGE, EVENT_POLL_TICK, EVENT_TERMINAL_REACHED, ) from utils.ci_watch_handoff_schema import ( TERMINAL_CHAIR_REQUIRED, TERMINAL_MERGE_READY, ) HEAD_A = "a" * 40 HEAD_B = "b" * 40 # ── ANCHOR-1: lifecycle event 6종 ──────────────────────────────────────────── def test_audit_schema_and_rel_path(): assert AUDIT_SCHEMA == "utils.ci_watch_handoff_audit.v1" assert AUDIT_REL_PATH == "memory/events/ci-watch-handoff-runner-audit.jsonl" def test_event_enum_exact_6(): assert ALL_EVENTS == frozenset( { EVENT_HANDOFF_RECEIVED, EVENT_POLL_TICK, EVENT_AUTO_REMEDIATE, EVENT_OWNER_NUDGE, EVENT_TERMINAL_REACHED, EVENT_CALLBACK_FIRED, } ) assert len(ALL_EVENTS) == 6 def test_append_handoff_received_writes_record(tmp_path): audit = CiWatchHandoffAudit(tmp_path) audit.append( { "task_id": "task-2642", "pr_number": 146, "head_sha": HEAD_A, "watcher_owner": "dev6-cron-watcher", "watcher_schedule_id": "sched-abc", "event": EVENT_HANDOFF_RECEIVED, "reason": "PR #146 handoff received", } ) lines = audit.path.read_text(encoding="utf-8").strip().splitlines() assert len(lines) == 1 rec = json.loads(lines[0]) assert rec["schema"] == AUDIT_SCHEMA assert rec["event"] == EVENT_HANDOFF_RECEIVED assert rec["watcher_owner"] == "dev6-cron-watcher" assert rec["watcher_schedule_id"] == "sched-abc" assert "ts_utc" in rec def test_append_normalizes_head_sha_to_lower(tmp_path): audit = CiWatchHandoffAudit(tmp_path) audit.append( { "pr_number": 146, "head_sha": "ABCDEF" + "0" * 34, "watcher_owner": "dev6", "event": EVENT_POLL_TICK, "ci_status": "PASS", "loop_iterations": 1, } ) rec = json.loads(audit.path.read_text(encoding="utf-8").strip()) assert rec["head_sha"] == "abcdef" + "0" * 34 # ── ANCHOR-2: terminal_state 필수 (TERMINAL_REACHED / CALLBACK_FIRED) ───────── def test_append_terminal_reached_requires_terminal_state(tmp_path): audit = CiWatchHandoffAudit(tmp_path) with pytest.raises(AuditError, match="terminal_state must be one of"): audit.append( { "pr_number": 146, "head_sha": HEAD_A, "watcher_owner": "dev6", "event": EVENT_TERMINAL_REACHED, # terminal_state 누락 "reason": "should fail", } ) def test_append_callback_fired_requires_terminal_state(tmp_path): audit = CiWatchHandoffAudit(tmp_path) with pytest.raises(AuditError, match="terminal_state must be one of"): audit.append( { "pr_number": 146, "head_sha": HEAD_A, "watcher_owner": "dev6", "event": EVENT_CALLBACK_FIRED, "callback_prompt_bytes": 512, "reason": "should fail", } ) def test_append_terminal_reached_with_valid_state_ok(tmp_path): audit = CiWatchHandoffAudit(tmp_path) audit.append( { "pr_number": 146, "head_sha": HEAD_A, "watcher_owner": "dev6", "event": EVENT_TERMINAL_REACHED, "terminal_state": TERMINAL_MERGE_READY, "reason": "MERGE_READY", } ) rec = json.loads(audit.path.read_text(encoding="utf-8").strip()) assert rec["terminal_state"] == TERMINAL_MERGE_READY def test_append_terminal_reached_with_invalid_state_rejected(tmp_path): audit = CiWatchHandoffAudit(tmp_path) with pytest.raises(AuditError, match="terminal_state must be one of"): audit.append( { "pr_number": 146, "head_sha": HEAD_A, "watcher_owner": "dev6", "event": EVENT_TERMINAL_REACHED, "terminal_state": "BOGUS", "reason": "should fail", } ) # ── ANCHOR-3: redaction guard (raw token sentinel 누출 차단) ────────────────── def test_append_rejects_disallowed_keys(tmp_path): audit = CiWatchHandoffAudit(tmp_path) with pytest.raises(AuditError, match="disallowed audit keys"): audit.append( { "pr_number": 146, "head_sha": HEAD_A, "watcher_owner": "dev6", "event": EVENT_POLL_TICK, "ci_status": "PASS", "unexpected_extra_key": "value", } ) def test_append_rejects_token_key_sentinel(tmp_path): audit = CiWatchHandoffAudit(tmp_path) with pytest.raises(AuditError): audit.append( { "pr_number": 146, "head_sha": HEAD_A, "watcher_owner": "dev6", "event": EVENT_POLL_TICK, "ci_status": "PASS", # disallowed key name with token sentinel "token": "ghp_x" * 8, } ) def test_append_rejects_bearer_token_in_value(tmp_path): audit = CiWatchHandoffAudit(tmp_path) # 'reason' is allowed key but value contains Bearer sentinel with pytest.raises(AuditError, match="token sentinel"): audit.append( { "pr_number": 146, "head_sha": HEAD_A, "watcher_owner": "dev6", "event": EVENT_POLL_TICK, "ci_status": "PASS", "reason": "Bearer " + "ghp_xxx", } ) # ── ANCHOR: invalid event 거부 ─────────────────────────────────────────────── def test_append_rejects_unknown_event(tmp_path): audit = CiWatchHandoffAudit(tmp_path) with pytest.raises(AuditError, match="event must be one of"): audit.append( { "pr_number": 146, "head_sha": HEAD_A, "watcher_owner": "dev6", "event": "NOT_AN_EVENT", "reason": "should fail", } ) # ── ANCHOR: records_for_pr_head ────────────────────────────────────────────── def test_records_for_pr_head_filters_pr_and_head(tmp_path): audit = CiWatchHandoffAudit(tmp_path) # PR 146 head A — 2 records for ev in (EVENT_HANDOFF_RECEIVED, EVENT_POLL_TICK): rec = { "pr_number": 146, "head_sha": HEAD_A, "watcher_owner": "dev6", "event": ev, } if ev == EVENT_POLL_TICK: rec["ci_status"] = "PENDING" rec["loop_iterations"] = 1 rec["reason"] = "x" audit.append(rec) # PR 999 head B — noise audit.append( { "pr_number": 999, "head_sha": HEAD_B, "watcher_owner": "other", "event": EVENT_HANDOFF_RECEIVED, "reason": "noise", } ) rows = audit.records_for_pr_head(pr_number=146, head=HEAD_A) assert len(rows) == 2 assert {r["event"] for r in rows} == {EVENT_HANDOFF_RECEIVED, EVENT_POLL_TICK} rows_other = audit.records_for_pr_head(pr_number=999, head=HEAD_B) assert len(rows_other) == 1 def test_audit_creates_parent_directory(tmp_path): nested = tmp_path / "nested" / "deep" audit = CiWatchHandoffAudit(nested) audit.append( { "pr_number": 1, "head_sha": HEAD_A, "watcher_owner": "dev6", "event": EVENT_HANDOFF_RECEIVED, "reason": "x", } ) assert audit.path.exists() assert audit.path.parent.is_dir() def test_allowed_keys_match_documented_set(): assert ALLOWED_AUDIT_KEYS == frozenset( { "schema", "ts_utc", "task_id", "pr_number", "head_sha", "watcher_owner", "watcher_schedule_id", "event", "terminal_state", "auto_remediation_attempts", "loop_iterations", "router_final_state", "ci_status", "callback_prompt_bytes", "reason", } ) def test_append_chair_required_terminal(tmp_path): audit = CiWatchHandoffAudit(tmp_path) audit.append( { "pr_number": 146, "head_sha": HEAD_A, "watcher_owner": "dev6", "event": EVENT_CALLBACK_FIRED, "terminal_state": TERMINAL_CHAIR_REQUIRED, "callback_prompt_bytes": 512, "reason": "chair callback", } ) rec = json.loads(audit.path.read_text(encoding="utf-8").strip()) assert rec["terminal_state"] == TERMINAL_CHAIR_REQUIRED assert rec["callback_prompt_bytes"] == 512