# -*- coding: utf-8 -*- """tests/regression/test_callback_registration_enforcement.py task-2635 — 5 fixture finalize_result 단언 + fail-closed 5값 enum 분기. Spec §5 / §4.1: REGISTERED → success+cancel · NOT_REGISTERED/SENDFILE_ONLY/ REGISTER_FAILED → fail-closed+no-cancel · SKIPPED_WITH_EXPLICIT_REASON → success+(이중잠금시) cancel. Subprocess 실호출 0. fixture evidence.json 을 그대로 finalize_hooks 에 흘려 보내고 expected.json 과 cross-check. """ from __future__ import annotations import json from pathlib import Path import pytest from utils.anu_callback_fallback import decide_fallback_cancel, expected_collector_spawn from utils.callback_envelope_schema import ( NormalCallbackRegistrationStatus, is_callback_complete, is_fail_closed_status, is_success_status, ) FIXTURE_ROOT = ( Path(__file__).resolve().parents[1] / "fixtures" / "normal_callback_registration" ) SCENARIOS = [ "registered_normal", "not_registered_envelope_only", "sendfile_only_no_cron", "register_failed_cli_error", "skipped_explicit_reason_dryrun", ] def _load_pair(scenario: str): sdir = FIXTURE_ROOT / scenario evidence = json.loads((sdir / "evidence.json").read_text(encoding="utf-8")) expected = json.loads((sdir / "expected.json").read_text(encoding="utf-8")) return evidence, expected @pytest.mark.parametrize("scenario", SCENARIOS) def test_fixture_directories_exist_with_three_files(scenario): sdir = FIXTURE_ROOT / scenario assert (sdir / "evidence.json").is_file() assert (sdir / "expected.json").is_file() assert (sdir / "PROVENANCE.md").is_file() @pytest.mark.parametrize("scenario", SCENARIOS) def test_evidence_status_matches_expected_status(scenario): evidence, expected = _load_pair(scenario) assert evidence["registration_status"] == expected["registration_status"] @pytest.mark.parametrize("scenario", SCENARIOS) def test_fail_closed_branching_matches_enum_contract(scenario): evidence, expected = _load_pair(scenario) status = evidence["registration_status"] fail_closed = is_fail_closed_status(status) assert fail_closed is bool(expected["fail_closed"]) # success-status set inverse: a status is success iff not fail-closed and # not unknown. The fixture matrix guarantees no unknown statuses. success = is_success_status(status) assert success is (not expected["fail_closed"]) @pytest.mark.parametrize("scenario", SCENARIOS) def test_fallback_decision_matches_expected(scenario): evidence, expected = _load_pair(scenario) decision = decide_fallback_cancel(evidence) assert decision.cancel_fallback is bool(expected["fallback_cancel_signal"]), ( f"scenario {scenario}: decided cancel={decision.cancel_fallback} " f"(reason={decision.reason!r}) vs expected={expected['fallback_cancel_signal']}" ) if "fallback_decision_reason_contains" in expected: assert expected["fallback_decision_reason_contains"] in decision.reason, ( f"scenario {scenario}: reason mismatch — got {decision.reason!r}" ) @pytest.mark.parametrize("scenario", SCENARIOS) def test_collector_spawn_expected_matches_expected(scenario): evidence, expected = _load_pair(scenario) spawn = expected_collector_spawn(evidence) assert spawn is bool(expected["collector_spawn_expected"]) callback_complete = is_callback_complete(evidence) assert callback_complete is bool(expected["is_callback_complete"]) def test_all_5_enum_values_are_covered_by_fixture_matrix(): """Each enum value appears in at least one fixture.""" used = set() for scenario in SCENARIOS: ev, _ = _load_pair(scenario) used.add(ev["registration_status"]) all_values = {s.value for s in NormalCallbackRegistrationStatus} assert used == all_values, ( f"missing enum values in fixture matrix: {all_values - used}; " f"extra: {used - all_values}" ) def test_finalize_with_callback_registration_success_on_skipped_fixture(): """skipped scenario: helper accepts skip_registration=True + reason and produces success outcome (matches expected.json contract).""" from dispatch.finalize_hooks import finalize_with_callback_registration fr = finalize_with_callback_registration( task_id="task-2635-skip-test", result={ "executor_name": "dispatch-executor-dev6", "result_path": "memory/tasks/task-2635-skip-test.result.json", "report_path": "memory/reports/task-2635-skip-test.md", "commit_sha": "0" * 40, "spec_sha256": "0fbd1dad1e110c49474dfbdf13a21fb3bdd9c7f094128004dba8472840bb832d", }, skip_registration=True, skip_reason="unit-test dry-run pre-approval", seed_envelope_overrides={"explicit_skip_authorizes_fallback_cancel": True}, ) assert fr.finalize_result == "success" assert ( fr.registration_status == NormalCallbackRegistrationStatus.SKIPPED_WITH_EXPLICIT_REASON.value ) # ANCHOR-4 — explicit skip + ouble-lock → fallback cancel allowed. assert fr.fallback_decision.cancel_fallback is True def test_finalize_with_callback_registration_fail_on_register_failed_path(): """Registrar returns REGISTER_FAILED → finalize_result must be 'fail'.""" from dispatch.finalize_hooks import finalize_with_callback_registration def fake_runner(argv, timeout): class _P: returncode = 2 stdout = "" stderr = '{"status":"error","message":"network unreachable"}' return _P() fr = finalize_with_callback_registration( task_id="task-2635-fail-test", result={ "executor_name": "dispatch-executor-dev6", "result_path": "memory/tasks/task-2635-fail-test.result.json", "report_path": "memory/reports/task-2635-fail-test.md", "commit_sha": "0" * 40, "spec_sha256": "0fbd1dad1e110c49474dfbdf13a21fb3bdd9c7f094128004dba8472840bb832d", }, subprocess_runner=fake_runner, cli_exists_check=lambda _: True, ) assert fr.finalize_result == "fail" assert ( fr.registration_status == NormalCallbackRegistrationStatus.REGISTER_FAILED.value ) assert fr.fallback_decision.cancel_fallback is False def test_finalize_with_callback_registration_success_on_registered_path(): """Registrar returns REGISTERED → finalize_result='success' but fallback cancel still requires the durable-success marker (ANCHOR-4 invariant). """ from dispatch.finalize_hooks import finalize_with_callback_registration def fake_runner(argv, timeout): class _P: returncode = 0 stdout = '{"status":"ok","id":"CRON-OK-1"}' stderr = "" return _P() fr = finalize_with_callback_registration( task_id="task-2635-ok-test", result={ "executor_name": "dispatch-executor-dev6", "result_path": "memory/tasks/task-2635-ok-test.result.json", "report_path": "memory/reports/task-2635-ok-test.md", "commit_sha": "0" * 40, "spec_sha256": "0fbd1dad1e110c49474dfbdf13a21fb3bdd9c7f094128004dba8472840bb832d", }, subprocess_runner=fake_runner, cli_exists_check=lambda _: True, ) assert fr.finalize_result == "success" assert fr.registration_status == NormalCallbackRegistrationStatus.REGISTERED.value assert fr.envelope["cron_schedule_id"] == "CRON-OK-1" # No durable marker in seed envelope → fallback cancel should be False. assert fr.fallback_decision.cancel_fallback is False assert "durable-success marker" in fr.fallback_decision.reason def test_dispatch_package_exports_finalize_hook(): """spec §3.3 §6.2 — dispatch/__init__.py 1줄 import 결선 확인.""" import dispatch assert hasattr(dispatch, "finalize_with_callback_registration")