# -*- coding: utf-8 -*- """tests.test_dispatch_spawn_verifier_2645 — task-2645 spawn gate regression.""" from __future__ import annotations import json from pathlib import Path import pytest from utils.dispatch_spawn_verifier import ( HISTORY_ABSENT_AFTER_COMPLETION, HISTORY_ABSENT_IN_PROGRESS, HISTORY_ABSENT_SILENT_DROP, HISTORY_PRESENT, SpawnSignals, collect_signals, dispatched_dict_is_unverified, evaluate_signals, is_silent_drop_confirmed, verify_spawn, ) from utils.dispatch_status_enum import ( DISPATCH_SILENT_DROP_HOLD, DISPATCH_SUBMITTED_UNVERIFIED, DISPATCH_VERIFIED_SPAWN, ) FIXTURE_DIR = Path(__file__).parent / "fixtures" / "dispatch_gate" def _signals(**overrides: bool) -> SpawnSignals: defaults = dict( workspace_dir_exists=True, workspace_dir_has_activity=True, executor_process_present=True, schedule_visible_in_cron_list=False, schedule_accepted_marker_present=True, schedule_history_log_present=True, ) defaults.update(overrides) return SpawnSignals(**defaults) # ── 4 항목 verification 라벨 검증 ─────────────────────────────────────── def test_signals_dataclass_has_all_4_items() -> None: s = _signals() j = s.to_json() for k in ( "workspace_dir_exists", "workspace_dir_has_activity", "executor_process_present", "schedule_visible_in_cron_list", "schedule_accepted_marker_present", "schedule_history_log_present", ): assert k in j # ── §17.11 spawn verification 4 항목 + verified ───────────────────────── def test_verified_spawn_from_fixture_success() -> None: fix = json.loads((FIXTURE_DIR / "spawn_success.json").read_text()) s = SpawnSignals(**fix["input"]["signals"]) r = evaluate_signals(fix["input"]["cron_id"], s) assert r.verified is True assert r.silent_drop is False assert r.history_state == fix["expected"]["history_state"] == HISTORY_PRESENT assert r.next_status == fix["expected"]["next_status"] == DISPATCH_VERIFIED_SPAWN assert is_silent_drop_confirmed(r) is False # ── §17.12 silent drop 4신호 모두 hit → DISPATCH_SILENT_DROP_HOLD ─────── def test_silent_drop_confirmed_from_673aa5a6_fixture() -> None: fix = json.loads((FIXTURE_DIR / "spawn_silent_drop.json").read_text()) s = SpawnSignals(**fix["input"]["signals"]) r = evaluate_signals(fix["input"]["cron_id"], s) assert r.verified is False assert r.silent_drop is True assert r.history_state == fix["expected"]["history_state"] == HISTORY_ABSENT_SILENT_DROP assert r.next_status == fix["expected"]["next_status"] == DISPATCH_SILENT_DROP_HOLD assert is_silent_drop_confirmed(r) is True # ── 4신호 중 일부만 hit 시 silent drop 단정 금지 ──────────────────────── @pytest.mark.parametrize( "override,expected_status,expected_silent_drop", [ # workspace 활동 hit 1개 — spawn verified ({"workspace_dir_exists": True, "workspace_dir_has_activity": True, "executor_process_present": False, "schedule_history_log_present": False, "schedule_accepted_marker_present": False, "schedule_visible_in_cron_list": False}, DISPATCH_VERIFIED_SPAWN, False), # executor process hit 1개 — spawn verified ({"workspace_dir_exists": False, "workspace_dir_has_activity": False, "executor_process_present": True, "schedule_history_log_present": False, "schedule_accepted_marker_present": False, "schedule_visible_in_cron_list": False}, DISPATCH_VERIFIED_SPAWN, False), # cron 가시만 hit (활동/process 0) — submitted_unverified 재폴링 ({"workspace_dir_exists": False, "workspace_dir_has_activity": False, "executor_process_present": False, "schedule_history_log_present": False, "schedule_accepted_marker_present": False, "schedule_visible_in_cron_list": True}, DISPATCH_SUBMITTED_UNVERIFIED, False), ], ) def test_partial_signals_do_not_falsely_confirm_silent_drop( override: dict, expected_status: str, expected_silent_drop: bool, ) -> None: s = SpawnSignals(**override) r = evaluate_signals("CRON-PARTIAL", s) assert r.next_status == expected_status assert r.silent_drop is expected_silent_drop # ── schedule_history 부재 3 상태 분리 (단독 단정 금지 doctrine) ───────── def test_history_absent_3_states_separation() -> None: # IN_PROGRESS: activity + executor + no log s_inp = SpawnSignals( workspace_dir_exists=True, workspace_dir_has_activity=True, executor_process_present=True, schedule_visible_in_cron_list=False, schedule_accepted_marker_present=False, schedule_history_log_present=False, ) r1 = evaluate_signals("C-INP", s_inp) assert r1.history_state == HISTORY_ABSENT_IN_PROGRESS # AFTER_COMPLETION: accepted marker + no log + cron 비가시 + 활동/process 0 # → silent drop 분기 회피를 위해 activity True 로 둔다 (1회성 자동삭제 후 워크스페이스 잔존) s_done = SpawnSignals( workspace_dir_exists=True, workspace_dir_has_activity=True, executor_process_present=False, schedule_visible_in_cron_list=False, schedule_accepted_marker_present=True, schedule_history_log_present=False, ) r2 = evaluate_signals("C-DONE", s_done) assert r2.history_state == HISTORY_ABSENT_AFTER_COMPLETION # SILENT_DROP: 4신호 모두 hit s_drop = SpawnSignals( workspace_dir_exists=False, workspace_dir_has_activity=False, executor_process_present=False, schedule_visible_in_cron_list=False, schedule_accepted_marker_present=False, schedule_history_log_present=False, ) r3 = evaluate_signals("C-DROP", s_drop) assert r3.history_state == HISTORY_ABSENT_SILENT_DROP # PRESENT r4 = evaluate_signals("C-OK", _signals()) assert r4.history_state == HISTORY_PRESENT # ── live filesystem collect (tmp_path 격리) ───────────────────────────── def test_collect_signals_reads_filesystem(tmp_path: Path) -> None: ws_root = tmp_path / "ws" hist_root = tmp_path / "hist" (ws_root / "ABCDEF12").mkdir(parents=True) (ws_root / "ABCDEF12" / "marker").write_text("x") (hist_root).mkdir() (hist_root / "ABCDEF12.log").write_text("run record") s = collect_signals( "ABCDEF12", workspace_root=str(ws_root), schedule_history_root=str(hist_root), cron_list_ids=("ABCDEF12",), executor_process_ids=("ABCDEF12",), accepted_marker_ids=(), ) assert s.workspace_dir_exists is True assert s.workspace_dir_has_activity is True assert s.schedule_history_log_present is True assert s.executor_process_present is True assert s.schedule_visible_in_cron_list is True def test_collect_signals_silent_drop_when_nothing_present(tmp_path: Path) -> None: s = collect_signals( "MISSING123", workspace_root=str(tmp_path / "ws"), schedule_history_root=str(tmp_path / "hist"), cron_list_ids=(), executor_process_ids=(), accepted_marker_ids=(), ) r = evaluate_signals("MISSING123", s) assert r.silent_drop is True assert r.next_status == DISPATCH_SILENT_DROP_HOLD def test_verify_spawn_composite_silent_drop(tmp_path: Path) -> None: r = verify_spawn( "DEAD1234", workspace_root=str(tmp_path / "ws"), schedule_history_root=str(tmp_path / "hist"), cron_list_ids=(), executor_process_ids=(), accepted_marker_ids=(), ) assert is_silent_drop_confirmed(r) is True # ── dispatched_dict_is_unverified — false ok 차단 게이트 ──────────────── def test_dispatched_dict_legacy_is_treated_as_unverified() -> None: d = {"status": "dispatched", "cron_response": {"status": "ok", "id": "673AA5A6"}} assert dispatched_dict_is_unverified(d) is True d2 = {"status": DISPATCH_SUBMITTED_UNVERIFIED} assert dispatched_dict_is_unverified(d2) is True # spawn_verification 키 있으면 더 이상 unverified 아님 d3 = { "status": "dispatched", "spawn_verification": {"verified": True, "next_status": DISPATCH_VERIFIED_SPAWN}, } assert dispatched_dict_is_unverified(d3) is False def test_dispatched_dict_handles_non_dict() -> None: assert dispatched_dict_is_unverified(None) is False assert dispatched_dict_is_unverified("ok") is False assert dispatched_dict_is_unverified(["ok"]) is False # ── JSON schema serialization ─────────────────────────────────────────── def test_result_json_schema_id() -> None: s = _signals() r = evaluate_signals("X", s) j = r.to_json() assert j["schema"] == "dispatch.spawn_verification_result.v1" assert "signals" in j assert "next_status" in j