# -*- coding: utf-8 -*- """task-2673 RS-2670-C — callback registrar invariants. 수정 목표 3: terminal state 도달 시 ANU normal callback 발사 보장. spec §11: envelope UTF-8 ≤ 3900 bytes / absolute timestamp / ANU key 단일출처. """ from __future__ import annotations import re import subprocess import sys from pathlib import Path from types import SimpleNamespace sys.path.insert(0, str(Path(__file__).resolve().parents[2])) import pytest # noqa: E402 from utils.pr_watcher_terminal_state_classifier import ( # noqa: E402 ANU_CHAT_ID_DEFAULT, ANU_KEY_DEFAULT, CALLBACK_DELAY_SEC, COKACDIR_BIN, ENVELOPE_MAX_BYTES, HOLD_FOR_CHAIR, LOOP_BOUNDARY, MERGE_READY, TERMINAL_STATES, build_callback_envelope, register_terminal_callback, ) def _ok(*_args, **_kwargs): return SimpleNamespace(returncode=0, stdout="ok", stderr="") def _capture(): captured = {} def runner(cmd, **kwargs): captured["cmd"] = cmd captured["kwargs"] = kwargs return SimpleNamespace(returncode=0, stdout="ok", stderr="") return runner, captured def test_envelope_within_byte_limit_for_all_states(): """envelope UTF-8 ≤ ENVELOPE_MAX_BYTES (모든 enum).""" for state in TERMINAL_STATES: env = build_callback_envelope( task_id="2667", pr_number=149, terminal_state=state, reason="test reason " * 5, polls_completed=12, elapsed_sec=1336, last_snapshot={ "head_match_expected": True, "mergeStateStatus": "BLOCKED", "unresolved_thread_count": 3, "latest_gemini_review": {"submittedAt": "2026-05-25T13:39:36Z"}, }, ) assert len(env.encode("utf-8")) <= ENVELOPE_MAX_BYTES, state assert state in env def test_envelope_truncates_when_extras_oversize(): """extras 가 거대해도 잘려서 ≤ ENVELOPE_MAX_BYTES.""" extras = {f"k{i}": "x" * 200 for i in range(40)} env = build_callback_envelope( task_id="2667", pr_number=149, terminal_state=HOLD_FOR_CHAIR, reason="x", polls_completed=1, elapsed_sec=1, extras=extras, ) assert len(env.encode("utf-8")) <= ENVELOPE_MAX_BYTES def test_envelope_rejects_invalid_state(): with pytest.raises(ValueError): build_callback_envelope( task_id="2667", pr_number=149, terminal_state="MERGE_GREEN_LIGHT", # not in enum reason="x", polls_completed=1, elapsed_sec=1, ) def test_register_calls_cokacdir_with_required_args(): """RS-2670-C 본체 — subprocess 인자 검증.""" runner, captured = _capture() env = build_callback_envelope( task_id="2667", pr_number=149, terminal_state=HOLD_FOR_CHAIR, reason="poll_12_silent_fall_through", polls_completed=12, elapsed_sec=1336, ) result = register_terminal_callback( envelope=env, anu_key=ANU_KEY_DEFAULT, chat_id=ANU_CHAT_ID_DEFAULT, runner=runner, ) assert result.fired is True assert result.envelope_bytes == len(env.encode("utf-8")) cmd = captured["cmd"] assert cmd[0] == COKACDIR_BIN assert cmd[1] == "--cron" assert cmd[2] == env assert "--at" in cmd assert "--chat" in cmd assert "--key" in cmd assert "--once" in cmd # ANU key 단일출처 검증 (self-key 차단) key_idx = cmd.index("--key") assert cmd[key_idx + 1] == ANU_KEY_DEFAULT chat_idx = cmd.index("--chat") assert cmd[chat_idx + 1] == str(ANU_CHAT_ID_DEFAULT) def test_register_uses_absolute_timestamp_not_relative(): """task-2661 Phase 2b — absolute "YYYY-MM-DD HH:MM:SS" 강제.""" runner, captured = _capture() env = build_callback_envelope( task_id="2667", pr_number=149, terminal_state=LOOP_BOUNDARY, reason="elapsed 3600s", polls_completed=30, elapsed_sec=3600, ) result = register_terminal_callback( envelope=env, anu_key=ANU_KEY_DEFAULT, chat_id=ANU_CHAT_ID_DEFAULT, delay_seconds=CALLBACK_DELAY_SEC, runner=runner, ) at_idx = captured["cmd"].index("--at") fire_at = captured["cmd"][at_idx + 1] # absolute "YYYY-MM-DD HH:MM:SS" assert re.match(r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$", fire_at) # relative 형식은 금지 assert not fire_at.endswith("s") assert not fire_at.endswith("m") assert result.fire_at == fire_at def test_register_skips_when_envelope_empty(): runner, captured = _capture() result = register_terminal_callback( envelope="", anu_key=ANU_KEY_DEFAULT, chat_id=ANU_CHAT_ID_DEFAULT, runner=runner, ) assert result.fired is False assert result.skipped_reason == "empty_envelope" assert "cmd" not in captured # subprocess 미호출 def test_register_skips_when_envelope_oversize(): runner, captured = _capture() oversize = "x" * (ENVELOPE_MAX_BYTES + 100) result = register_terminal_callback( envelope=oversize, anu_key=ANU_KEY_DEFAULT, chat_id=ANU_CHAT_ID_DEFAULT, runner=runner, ) assert result.fired is False assert "envelope_oversize" in result.skipped_reason assert "cmd" not in captured def test_register_silent_on_subprocess_exception(): """RS-2670-C-fail — timeout/exception 시 watcher 정상 종료.""" def bad_runner(*_a, **_k): raise subprocess.TimeoutExpired(cmd="cokacdir", timeout=30) env = build_callback_envelope( task_id="2667", pr_number=149, terminal_state=MERGE_READY, reason="ok", polls_completed=15, elapsed_sec=1800, ) result = register_terminal_callback( envelope=env, anu_key=ANU_KEY_DEFAULT, chat_id=ANU_CHAT_ID_DEFAULT, runner=bad_runner, ) assert result.fired is False assert "TimeoutExpired" in result.skipped_reason def test_register_uses_default_key_when_env_and_explicit_missing(monkeypatch): """ANU key fail-closed default — explicit/env 모두 없으면 ANU_KEY_DEFAULT 로 fallback (spec §11 "ANU key 단일출처 유지" 정합). """ monkeypatch.delenv("ANU_KEY", raising=False) runner, captured = _capture() env = build_callback_envelope( task_id="2667", pr_number=149, terminal_state=HOLD_FOR_CHAIR, reason="x", polls_completed=1, elapsed_sec=1, ) result = register_terminal_callback( envelope=env, anu_key=None, # 명시적 None chat_id=ANU_CHAT_ID_DEFAULT, runner=runner, ) assert result.fired is True key_idx = captured["cmd"].index("--key") assert captured["cmd"][key_idx + 1] == ANU_KEY_DEFAULT def test_register_non_zero_returncode_not_fired(): def runner(*_a, **_k): return SimpleNamespace(returncode=1, stdout="", stderr="err") env = build_callback_envelope( task_id="2667", pr_number=149, terminal_state=LOOP_BOUNDARY, reason="x", polls_completed=30, elapsed_sec=3600, ) result = register_terminal_callback( envelope=env, anu_key=ANU_KEY_DEFAULT, chat_id=ANU_CHAT_ID_DEFAULT, runner=runner, ) assert result.fired is False assert result.skipped_reason == "non_zero_exit" assert result.returncode == 1