# -*- coding: utf-8 -*- """tests.regression.test_spawn_callback_contract_validator — task-2640 Track A. 회장 verbatim unfork #1/#2 정합 검증 — dispatch.spawn_callback_contract_validator 6 fixture parametrized PASS/FAIL 단언. 본 회귀는 Layer A / NO-CRON: subprocess / cokacdir / merge / cron 호출 0. """ from __future__ import annotations import importlib.util import json import sys from pathlib import Path import pytest # task-2640 — tests/dispatch (테스트 sub-package, __init__.py 빈 파일) 가 # 최상위 dispatch namespace 를 sys.path 우선순위로 그림자처럼 가린다 (pytest 가 # tests/ 자체를 sys.path 에 추가하기 때문). 본 회귀는 worktree 의 실 dispatch # 패키지를 로드해야 하므로 tests/ 항목을 sys.path 에서 제거한 뒤 worktree 루트를 # 선두에 둔다. tests.regression 패키지 자체는 이미 import 된 상태이므로 후속 # 영향 없음. 기존 test_callback_runtime_enforcement_2626.py 의 _load_real 패턴은 # 부분 모듈만 로드하지만 본 회귀는 dispatch.core (heavy facade) 도 필요해 전체 # 패키지 reload 가 필요. _ROOT = Path(__file__).resolve().parents[2] _TESTS_DIR = str(_ROOT / "tests") sys.path[:] = [p for p in sys.path if p != _TESTS_DIR] if str(_ROOT) not in sys.path: sys.path.insert(0, str(_ROOT)) # 캐싱된 그림자 dispatch 가 있으면 제거 → worktree 의 dispatch 로 fresh import. for _m in [ k for k in list(sys.modules) if k == "dispatch" or k.startswith("dispatch.") ]: _cached = sys.modules.get(_m) _f = getattr(_cached, "__file__", "") or "" if "/tests/dispatch" in _f or _f == "": del sys.modules[_m] def _load_real(modname: str, relpath: str): existing = sys.modules.get(modname) if existing is not None and getattr(existing, "__file__", "") and str(existing.__file__).endswith(relpath): return existing spec = importlib.util.spec_from_file_location(modname, _ROOT / relpath) assert spec is not None and spec.loader is not None, f"spec not found: {relpath}" mod = importlib.util.module_from_spec(spec) sys.modules[modname] = mod spec.loader.exec_module(mod) return mod # dispatch 패키지 자체를 worktree 의 __init__.py 로 강제 로딩 (heavy 하지만 본 # 회귀는 dispatch.core / dispatch.prompt facade 도 필요). _load_real("dispatch", "dispatch/__init__.py") _load_real("dispatch.callback_owner_enforcer", "dispatch/callback_owner_enforcer.py") _load_real("dispatch.normal_fallback_callback_helper", "dispatch/normal_fallback_callback_helper.py") _load_real( "dispatch.spawn_callback_contract_validator", "dispatch/spawn_callback_contract_validator.py", ) _load_real("dispatch.core", "dispatch/core.py") _load_real("dispatch.prompt", "dispatch/prompt.py") WORKSPACE = Path(__file__).resolve().parent.parent.parent FIXTURE_ROOT = WORKSPACE / "tests" / "fixtures" / "self_collector_enforcement" # Scenarios that exercise validate_spawn_callback_contract directly. SPAWN_VALIDATOR_SCENARIOS = ( "spawn_contract_pass_anu_key", "spawn_contract_fail_self_key", "spawn_contract_fail_no_prompt_doctrine", ) def _load_fixture(scenario: str) -> tuple[dict, dict]: fdir = FIXTURE_ROOT / scenario evidence = json.loads((fdir / "evidence.json").read_text(encoding="utf-8")) expected = json.loads((fdir / "expected.json").read_text(encoding="utf-8")) return evidence, expected @pytest.mark.parametrize("scenario", SPAWN_VALIDATOR_SCENARIOS) def test_validate_spawn_callback_contract_fixture(scenario: str) -> None: """validate_spawn_callback_contract 가 fixture verdict 와 정합한다.""" from dispatch.spawn_callback_contract_validator import ( VALIDATOR_SCHEMA, validate_spawn_callback_contract, ) evidence, expected = _load_fixture(scenario) result = validate_spawn_callback_contract( task_id=evidence["task_id"], executor_key=evidence["executor_key"], anu_key=evidence["anu_key"], prompt_text=evidence["prompt_text"], anu_keys=tuple(evidence.get("anu_keys", ())), anu_keys_resolvable=evidence.get("anu_keys_resolvable", True), ) assert result.schema == VALIDATOR_SCHEMA assert result.verdict == expected["verdict"], ( f"{scenario}: verdict 불일치 (got={result.verdict}, " f"expected={expected['verdict']}, reasons={result.reasons})" ) assert result.ok == expected["ok"] assert result.prompt_has_anu_key == expected["prompt_has_anu_key"] assert result.prompt_has_collector_role == expected["prompt_has_collector_role"] assert result.prompt_has_required_doctrine == expected["prompt_has_required_doctrine"] expected_primary = expected.get("primary_classification") if expected_primary is None: assert result.primary_classification is None assert result.classifications == [] else: assert result.primary_classification == expected_primary, ( f"{scenario}: primary classification 불일치 " f"(got={result.primary_classification}, " f"expected={expected_primary})" ) must_contain = expected.get("must_contain_classifications", []) for cls in must_contain: assert cls in result.classifications, ( f"{scenario}: classification {cls!r} 누락 " f"(got={result.classifications})" ) def test_dispatch_to_bot_with_contract_pass_path() -> None: """dispatch.core.dispatch_to_bot_with_contract PASS path — 정상 ANU key.""" from dispatch.core import dispatch_to_bot_with_contract evidence, _ = _load_fixture("spawn_contract_pass_anu_key") out = dispatch_to_bot_with_contract( task_id=evidence["task_id"], executor_key=evidence["executor_key"], prompt=evidence["prompt_text"], anu_key=evidence["anu_key"], ) assert out["status"] == "ok" assert out["verdict"] == "PASS" assert out["classifications"] == [] assert "no_op_reason" not in out def test_dispatch_to_bot_with_contract_block_self_key() -> None: """dispatch.core.dispatch_to_bot_with_contract block — executor self-key.""" from dispatch.core import ( NO_OP_SPAWN_CONTRACT_FAILED, dispatch_to_bot_with_contract, ) evidence, _ = _load_fixture("spawn_contract_fail_self_key") out = dispatch_to_bot_with_contract( task_id=evidence["task_id"], executor_key=evidence["executor_key"], prompt=evidence["prompt_text"], anu_key=evidence["anu_key"], ) assert out["status"] == "blocked" assert out["verdict"] == "FAIL" assert out["no_op_reason"] == NO_OP_SPAWN_CONTRACT_FAILED assert out["primary_classification"] == "SELF_COLLECTOR_FORBIDDEN" def test_dispatch_to_bot_with_contract_block_no_doctrine() -> None: """dispatch.core.dispatch_to_bot_with_contract block — prompt doctrine 부재.""" from dispatch.core import ( NO_OP_SPAWN_CONTRACT_FAILED, dispatch_to_bot_with_contract, ) evidence, _ = _load_fixture("spawn_contract_fail_no_prompt_doctrine") out = dispatch_to_bot_with_contract( task_id=evidence["task_id"], executor_key=evidence["executor_key"], prompt=evidence["prompt_text"], anu_key=evidence["anu_key"], ) assert out["status"] == "blocked" assert out["verdict"] == "FAIL" assert out["no_op_reason"] == NO_OP_SPAWN_CONTRACT_FAILED assert "PROMPT_DOCTRINE_MISSING" in out["classifications"] def test_validator_hold_for_chair_when_anu_keys_unresolvable() -> None: """ANU key 집합 resolution 불가 시 HOLD_FOR_CHAIR (silent pass 금지).""" from dispatch.spawn_callback_contract_validator import ( validate_spawn_callback_contract, ) result = validate_spawn_callback_contract( task_id="task-fixture-hold", executor_key="1e41a2324a3ccdd0", anu_key="c119085addb0f8b7", prompt_text="dummy", anu_keys=(), anu_keys_resolvable=False, ) assert result.verdict == "HOLD_FOR_CHAIR" assert result.classifications == []