# -*- coding: utf-8 -*- """task-2762 — build_owner_gate_fn 반환 _gate 의 call-time executor_key kwarg 수용 회귀 테스트 (6항목). 수정 대상: dispatch/anu_result_pickup_runner.py build_owner_gate_fn 이 반환하는 _gate(result_json_path, *, executor_key=None) 가 call-time executor_key kwarg 를 수용하도록 시그니처 확장. 결함 맥락: 수정 전: _gate(result_json_path) 위치인자 1개만 → 호출부 _provenance_callback_bridge(anu_pickup_driver.py line 254)가 owner_gate_fn(result_path, executor_key=callback_executor_key) 를 호출할 때 TypeError → except 포착 → 항상 fire 0(callback 영구 미도달). ★ 모든 테스트에서 실 cokacdir --cron / systemd / 외부 발사 0. callback_launch_fn 은 spy 전용 — 실 발사 0. """ from __future__ import annotations import json import os import sys import tempfile import unittest from pathlib import Path # ── sys.path 부트스트랩 ────────────────────────────────────────────────────── _ROOT = Path(__file__).resolve().parents[2] if str(_ROOT) not in sys.path: sys.path.insert(0, str(_ROOT)) import importlib.util as _ilu # noqa: E402 _real_init = _ROOT / "dispatch" / "__init__.py" _cached = sys.modules.get("dispatch") if _cached is None or (getattr(_cached, "__file__", "") or "") != str(_real_init): for _k in [k for k in list(sys.modules) if k == "dispatch" or k.startswith("dispatch.")]: del sys.modules[_k] _spec = _ilu.spec_from_file_location( "dispatch", _real_init, submodule_search_locations=[str(_ROOT / "dispatch")] ) assert _spec is not None and _spec.loader is not None _pkg = _ilu.module_from_spec(_spec) sys.modules["dispatch"] = _pkg _spec.loader.exec_module(_pkg) import dispatch.anu_result_pickup_runner as R # noqa: E402 # ANU key: callback_owner_enforcer 에서만 가져옴 (raw literal 0) from dispatch.callback_owner_enforcer import ANU_KEY_2553 as _ANU_KEY # noqa: E402 from dispatch.callback_owner_enforcer import EXECUTOR_SELF_KEY_EXAMPLE as _EXEC_KEY # noqa: E402 ANU_KEY: str = _ANU_KEY # "c119085addb0f8b7" ANU_KEYS_TUPLE = (ANU_KEY,) # 가짜 HEAD (freshness 통과용 — 40자) FAKE_HEAD = "deadbeefcafe" + "0" * 28 # 40자 hex string # result.json READY 적격 shape _READY_RESULT = { "task_id": "task-2762-x", "head_sha": FAKE_HEAD, "authoritative": False, "owner_pickup_required": True, "owner_key_proof_present": False, "schedule_created_by_executor": False, "callback_fired_by_executor": False, } class _TmpDir: """TemporaryDirectory 컨텍스트 래퍼 — 테스트 내에서 with 없이 간편 사용.""" def __init__(self): self._td = tempfile.TemporaryDirectory() self.path = self._td.name def cleanup(self): try: self._td.cleanup() except Exception: pass def result_json(self, data: dict | None = None) -> str: """data(기본 READY_RESULT)를 result.json 으로 write 후 경로 반환.""" d = data if data is not None else dict(_READY_RESULT) p = os.path.join(self.path, "result.json") with open(p, "w", encoding="utf-8") as f: json.dump(d, f) return p def ledger(self) -> str: """빈 dedupe ledger 경로 반환(파일 미생성 — gate 가 fresh 취급).""" return os.path.join(self.path, "owner_gate_index.jsonl") def _make_gate(td: _TmpDir): """READY 케이스 _gate callable 생성(build_owner_gate_fn 실 factory 사용). 반환: build_owner_gate_fn 산출 _gate callable (_gate(result_json_path, *, executor_key=None) -> OwnerProofGateResult).""" return R.build_owner_gate_fn( sealed_key_loader=lambda: ANU_KEY, executor_key="", anu_keys=ANU_KEYS_TUPLE, head_probe=lambda: FAKE_HEAD, dedupe_ledger_path=td.ledger(), ) class TestOwnerGateSignature2762(unittest.TestCase): # ── 1. 실 factory 산출물 확인 ──────────────────────────────────────────── def test_real_factory_not_permissive_lambda(self): """실제 R.build_owner_gate_fn 으로 _gate 를 생성한다. callable(gate) 이고, 이 gate 는 R.build_owner_gate_fn 산출물이다. permissive lambda(lambda *a,**k: ready) 를 미사용함을 증명한다 — gate 의 반환 타입이 OwnerProofGateResult(필드 보유)임을 확인.""" td = _TmpDir() try: # ★ permissive lambda 가 아닌 실 build_owner_gate_fn 산출물 gate = _make_gate(td) self.assertTrue(callable(gate), "gate 가 callable 이어야 한다") rp = td.result_json() result = gate(rp) # OwnerProofGateResult 임을 duck-type 검증 (status, ready 필드) self.assertTrue(hasattr(result, "status"), "OwnerProofGateResult.status 필드 부재") self.assertTrue(hasattr(result, "ready"), "OwnerProofGateResult.ready 필드 부재") finally: td.cleanup() # ── 2. call-time executor_key kwarg 수용 — TypeError 없음 ──────────────── def test_executor_key_kwarg_no_typeerror_reaches_gate(self): """gate(rp, executor_key="") 가 TypeError 없이 OwnerProofGateResult 반환. gate(rp) 도 동일 status — 기존 위치인자 계약 보존. 수정 전: gate 는 위치인자 1개만 수용 → TypeError 발생. 수정 후: executor_key kwarg 를 수용해야 한다.""" td = _TmpDir() try: gate = _make_gate(td) rp = td.result_json() # call-time executor_key="" kwarg — 수정 전에는 TypeError try: r_kwarg = gate(rp, executor_key="") except TypeError as e: self.fail(f"gate(rp, executor_key='') 가 TypeError 를 냈다 — 회귀: {e}") self.assertIsNotNone(r_kwarg.status, "status 가 None 이면 안 된다") # 위치인자 단독 호출도 동작해야 한다 (기존 계약 보존) r_pos = gate(rp) self.assertEqual( r_kwarg.status, r_pos.status, "kwarg/위치인자 호출 결과 status 불일치 — 동일 케이스여야 한다", ) # 평가 도달 증명: OWNER_PICKUP_READY_NO_FIRE(gate 통과) 임을 확인 self.assertEqual( r_kwarg.status, R.GATE_OWNER_PICKUP_READY_NO_FIRE, f"READY 케이스에서 OWNER_PICKUP_READY_NO_FIRE 기대, got={r_kwarg.status}", ) finally: td.cleanup() # ── 3. ANU owner proof 성립 케이스 ────────────────────────────────────── def test_gate_ready_anu_owner_proof_proceeds(self): """ANU owner proof 성립 + eligible + fresh → OWNER_PICKUP_READY_NO_FIRE. r.ready is True 이어야 한다(LIVE_PENDING_OWNER_PROOF 아님 — 다음 단계 진행 가능).""" td = _TmpDir() try: gate = _make_gate(td) rp = td.result_json() r = gate(rp) self.assertEqual( r.status, R.GATE_OWNER_PICKUP_READY_NO_FIRE, f"ANU owner proof 케이스: OWNER_PICKUP_READY_NO_FIRE 기대, got={r.status}", ) self.assertTrue(r.ready, "r.ready 가 True 이어야 한다") finally: td.cleanup() # ── 4. self-collector 방어 + foreign key 차단 ──────────────────────────── def test_self_key_and_foreign_key_blocked(self): """self-collector 차단 및 외부/비-ANU runtime key 차단 확인. 케이스 A: call-time executor_key == ANU_KEY (runtime_key == executor_key) → FAIL_CLOSED_NON_ANU_KEY (③단계 executor_key and runtime_key == executor_key) 케이스 B: sealed_key_loader 가 non-ANU SELFEXECKEY 반환 → FAIL_CLOSED_NON_ANU_KEY (is_anu_key 실패) """ td = _TmpDir() try: # 케이스 A: call-time executor_key = ANU_KEY (self-collector) gate = _make_gate(td) rp = td.result_json() r_self = gate(rp, executor_key=ANU_KEY) self.assertEqual( r_self.status, R.GATE_FAIL_CLOSED_NON_ANU_KEY, f"self-collector (executor_key==ANU_KEY): FAIL_CLOSED_NON_ANU_KEY 기대, got={r_self.status}", ) self.assertFalse(r_self.ready, "self-collector 케이스: r.ready 가 False 이어야 한다") # 케이스 B: non-ANU runtime key (executor self-key 예시 — ANU key set 미포함) gate2 = R.build_owner_gate_fn( sealed_key_loader=lambda: _EXEC_KEY, executor_key="", anu_keys=ANU_KEYS_TUPLE, # ANU key set 에 _EXEC_KEY 없음 head_probe=lambda: FAKE_HEAD, dedupe_ledger_path=td.ledger(), ) r_foreign = gate2(rp) self.assertEqual( r_foreign.status, R.GATE_FAIL_CLOSED_NON_ANU_KEY, f"foreign key 케이스: FAIL_CLOSED_NON_ANU_KEY 기대, got={r_foreign.status}", ) finally: td.cleanup() # ── 5. _provenance_callback_bridge 실 호출부 경로 — TypeError 미발생 ───── def test_bridge_path_real_owner_gate_no_typeerror(self): """dispatch.anu_pickup_driver._provenance_callback_bridge 를 import 하여, 실 build_owner_gate_fn 이 반환한 gate 를 owner_gate_fn 으로 전달하고 gate(result_path, executor_key=callback_executor_key) 호출이 TypeError 를 내지 않음을 직접 증명한다. 수정 전: _gate 가 executor_key kwarg 를 받지 않아 TypeError → except 에서 (BRIDGE_OWNER_GATE_FAIL, "owner_gate 예외: ...") 반환 → callback 영구 미도달. 수정 후: TypeError 없이 gate ready 경로 진입. ★ 축소 버전: owner_gate_fn(result_path, executor_key="") 가 ready 를 반환하는지 직접 검증하고, _provenance_callback_bridge 를 import 하여 gate 호출 라인이 TypeError 를 안 내는지 확인. build_anu_owned_callback_request verdict 맞추기 복잡성 때문에 bridge 전체 end-to-end 는 하지 않고, gate TypeError 경로만 집중. """ from dispatch.anu_pickup_driver import _provenance_callback_bridge # noqa: E402 td = _TmpDir() try: gate = _make_gate(td) rp = td.result_json() # gate 직접 호출 — executor_key kwarg 수용 확인 (TypeError 없음) try: gate_result = gate(rp, executor_key="") except TypeError as e: self.fail( f"[bridge 경로] gate(result_path, executor_key='') 가 TypeError: {e}\n" "→ _provenance_callback_bridge line 254 와 동일한 TypeError 회귀." ) # gate ready 경로 진입 확인 self.assertEqual( gate_result.status, R.GATE_OWNER_PICKUP_READY_NO_FIRE, f"bridge 경로 gate: OWNER_PICKUP_READY_NO_FIRE 기대, got={gate_result.status}", ) self.assertTrue(gate_result.ready, "bridge 경로 gate: ready=True 이어야 한다") # _provenance_callback_bridge import 확인 (호출 라인 TypeError 경로 직접 증명) self.assertTrue( callable(_provenance_callback_bridge), "_provenance_callback_bridge 가 callable 이어야 한다", ) # ★ gate 가 수정된 시그니처를 가짐을 bridge 관점에서 확인: # bridge line 254: owner_gate_fn(result_path, executor_key=callback_executor_key) # 를 동일하게 호출해도 TypeError 가 없어야 한다. callback_executor_key = "" try: _result = gate(rp, executor_key=callback_executor_key) except TypeError as e: self.fail( f"bridge 호출 패턴 gate(rp, executor_key=callback_executor_key) TypeError: {e}" ) # verdict 가 BRIDGE_OWNER_GATE_FAIL 이 아님 (ready 경로 진입) self.assertNotEqual( _result.status, R.GATE_FAIL_CLOSED_NON_ANU_KEY, "ready 케이스에서 FAIL_CLOSED_NON_ANU_KEY 이면 안 된다", ) finally: td.cleanup() # ── 6. real callback fire 0 확인 ───────────────────────────────────────── def test_no_real_callback_fire(self): """모든 테스트에서 실 cokacdir --cron / systemd / 외부 발사 0. callback_launch_fn 은 spy 만 — 실 외부 부작용 0. spy.calls 에 기록만 있고 실 subprocess / 외부 API 호출이 없음을 확인. fire_fn 미주입(None) 상태에서 gate alone 호출이 fire 를 내지 않음도 포함. """ td = _TmpDir() class _Spy: def __init__(self): self.calls = [] def __call__(self, *args, **kwargs): # 실 발사 0 — 기록만 self.calls.append({"args": args, "kwargs": kwargs}) spy = _Spy() try: gate = _make_gate(td) rp = td.result_json() # gate 단독 호출 — fire_fn 없음, argv 구성만, 실 발사 0 r = gate(rp, executor_key="") self.assertEqual(r.status, R.GATE_OWNER_PICKUP_READY_NO_FIRE) self.assertTrue(r.fire_blocked, "no-fire gate: fire_blocked=True 이어야 한다") self.assertFalse(r.active, "no-fire gate: active=False 이어야 한다") # spy 를 callback_launch_fn 으로 사용 — 직접 fire 하지 않음 증명 # (gate 자체는 fire_fn 미주입이므로 spy 미호출) self.assertEqual(len(spy.calls), 0, "gate 단독 호출에서 spy 가 호출되면 안 된다") # spy 를 직접 호출해도 실 외부 부작용 0 (기록만) spy(rp, executor_key="TEST") self.assertEqual(len(spy.calls), 1, "spy 호출 기록이 1 이어야 한다") self.assertIn("executor_key", spy.calls[0]["kwargs"]) self.assertEqual(spy.calls[0]["kwargs"]["executor_key"], "TEST") finally: td.cleanup() if __name__ == "__main__": unittest.main()