# -*- coding: utf-8 -*- """tests/regression/test_p0b_callback_launch_signature_2763.py task-2763 — callback_launch_fn call-time 시그니처 회귀 테스트. 수정 대상: dispatch/anu_pickup_driver.py scan_live_inbox_once (line ~1826) 가 callback_launch_fn 을 단일 dict positional 로 호출하던 결함을, _fn(result_json_path, *, task_id, sha256, executor_key="", ...) 시그니처에 맞게 명시 kwarg 전달로 수정했다: callback_launch_fn(p, task_id=task_id, sha256=sha, executor_key=executor_key) 회귀 핵심: - 기존 mock/spy(아무거나 받는 lambda) 는 결함을 가린다. - 본 테스트는 **실제 build_callback_launch_fn factory** 를 사용해 결함 노출/방지한다. - 수정 전 코드(dict positional)면 TypeError → FIRE_FAILED 로 이 테스트 FAIL. - 수정 후에는 TypeError 없이 PASS. ★ 모든 테스트: runner_fire_fn=fake(실 cron/systemd/subprocess 발사 0). canonical root 미접촉 · tmp_path hermetic. """ from __future__ import annotations import hashlib import json import os import sys import tempfile import unittest from pathlib import Path from typing import List # ── sys.path / dispatch 패키지 부트스트랩 ────────────────────────────────────── # 기존 2762/2760 테스트와 동일 패턴 — dispatch.__file__ 이 worktree root 를 가리켜야 함. _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) from dispatch import anu_pickup_driver as drv # noqa: E402 from dispatch.anu_callback_launch_audit import ( # noqa: E402 build_callback_launch_fn, LAUNCH_NOOP_DISABLED, LAUNCH_ENFORCED, LAUNCH_NO_RESULT_JSON, LAUNCH_FAIL, ) # ── fake runner_fire_fn ──────────────────────────────────────────────────────── class _FakeFireResult: """anu_runner_pickup_and_fire 가 반환하는 ok+argv mock.""" def __init__(self): self.ok = True self.argv = ["fake-argv"] self.reasons: list = [] class _FakeFireFn: """실 cron/subprocess 발사 0 — 기록만 남기는 fake runner_fire_fn.""" def __init__(self): self.calls: List[dict] = [] def __call__(self, *, result_json_path, executor_key, anu_key, anu_keys, **kwargs): self.calls.append({ "result_json_path": result_json_path, "executor_key": executor_key, "anu_key": anu_key, }) return _FakeFireResult() # ── helpers ─────────────────────────────────────────────────────────────────── def _enabled(): """activation flag_reader: ACTIVE 활성 모사(실파일 미생성).""" return drv.ACTIVATION_ENABLED def _cb_flag_enabled(): """callback launch ACTIVE flag reader: enabled 반환.""" from dispatch.anu_callback_launch_audit import CALLBACK_LAUNCH_ENABLED return CALLBACK_LAUNCH_ENABLED def _cb_flag_disabled(): """callback launch ACTIVE flag reader: disabled 반환.""" return "" class _Gate: """owner_gate_fn 결과 모사. ready=True → ANU owner proof 통과.""" def __init__(self, ready: bool): self.ready = ready self.status = drv.GATE_READY_STATUS if ready else "OWNER_PICKUP_BLOCKED" def __call__(self, path, *, executor_key=""): return self def _write_live_result(root: str, task_id: str, body: str = "ok") -> str: """tmp root 의 LIVE_INBOX 에 task-XXXX.result.json 작성 → 경로 반환.""" d = os.path.join(root, *drv.LIVE_INBOX_DIR_REL.split("/")) os.makedirs(d, exist_ok=True) p = os.path.join(d, f"{task_id}.result.json") with open(p, "w", encoding="utf-8") as fh: fh.write(json.dumps({"task_id": task_id, "body": body}, ensure_ascii=False)) return p def _sha256_of(path: str) -> str: h = hashlib.sha256() with open(path, "rb") as fh: for chunk in iter(lambda: fh.read(65536), b""): h.update(chunk) return h.hexdigest() # ════════════════════════════════════════════════════════════════════════════════ class TestCallbackLaunchSignature2763(unittest.TestCase): """callback_launch_fn 시그니처 회귀 — 실 factory build_callback_launch_fn 사용.""" # ── 1. 실 factory _fn 시그니처 — positional path + keyword id/sha 수용 ────── def test_real_factory_fn_signature_accepts_positional_path_and_keyword_id_sha(self): """실제 build_callback_launch_fn 이 반환하는 _fn 을 _fn(result_json_path, task_id=..., sha256=...) 형태로 호출 가능한지 검증. ACTIVE flag OFF 환경 → NOOP_DISABLED 반환(no-op) — 시그니처 호환만 확인. 실 cron 발사 0(flag OFF + runner_fire_fn=fake). """ fake_fire = _FakeFireFn() # flag OFF: NOOP_DISABLED 경로 — 시그니처 체크만 _fn = build_callback_launch_fn( root="/nonexistent-fake-root", flag_reader=_cb_flag_disabled, runner_fire_fn=fake_fire, ) with tempfile.TemporaryDirectory() as tmp: p = os.path.join(tmp, "task-2763-x.result.json") with open(p, "w") as fh: fh.write('{"task_id":"task-2763-x"}') try: result = _fn(p, task_id="task-2763-x", sha256="aabbcc") except TypeError as e: self.fail( f"build_callback_launch_fn 반환 _fn(path, task_id=, sha256=) 가 " f"TypeError: {e}\n→ 시그니처 회귀." ) # flag OFF → NOOP_DISABLED 확인(실 fire 없음) self.assertEqual( result.verdict, LAUNCH_NOOP_DISABLED, f"flag OFF → NOOP_DISABLED 기대, got={result.verdict}", ) self.assertEqual(len(fake_fire.calls), 0, "flag OFF → 실 fire 호출 0") # ── 2. scan_live_inbox_once 에 실 factory fn 주입 → TypeError 없이 도달 ──── def test_scan_live_inbox_reaches_real_factory_without_typeerror(self): """scan_live_inbox_once 에 실제 build_callback_launch_fn factory 결과를 주입하고, activation ON + owner_gate ready(fake) + 유효한 result.json 1건 배치. scan 실행 시 TypeError 없이 callback_launch_fn 까지 도달해야 한다. ★ 회귀 핵심: 수정 전 코드(dict positional 호출)면 TypeError → FIRE_FAILED 로 이 테스트 FAIL. 수정 후엔 PASS (결과 verdict != FIRE_FAILED). """ fake_fire = _FakeFireFn() with tempfile.TemporaryDirectory() as root: # callback launch flag ON (fake flag_reader 주입) _fn = build_callback_launch_fn( root=root, flag_reader=_cb_flag_enabled, runner_fire_fn=fake_fire, audit_path=os.path.join(root, "fake_audit.jsonl"), ) # result.json 배치 p = _write_live_result(root, "task-2763-y") ledger = os.path.join(root, "ledger.jsonl") pdir = os.path.join(root, "processed") gate = _Gate(ready=True) try: records = drv.scan_live_inbox_once( root, flag_reader=_enabled, owner_gate_fn=gate, callback_launch_fn=_fn, paths=[p], ledger_path=ledger, processed_dir=pdir, ) except TypeError as e: self.fail( f"scan_live_inbox_once 내 callback_launch_fn 호출에서 TypeError: {e}\n" "→ callback_launch_fn(p, task_id=...) 호출 시그니처 회귀." ) verdicts = [r.verdict for r in records] # FIRE_FAILED 가 TypeError 에서 비롯된 것이 아님을 확인(시그니처 결함 아님). # (flag ON + ANU key loader 없으면 SELF_KEY_REFUSED 또는 LAUNCH_ENFORCED 가능. # 중요: FIRE_FAILED with "TypeError" 가 아니어야 한다.) for rec in records: if rec.verdict == drv.VERDICT_FIRE_FAILED: err = getattr(rec, "error", "") or "" self.assertNotIn( "TypeError", err, f"FIRE_FAILED verdict 의 error 에 TypeError 포함 — 시그니처 회귀: {err}", ) # scan 은 적어도 1개 record 를 반환해야 한다 self.assertGreater(len(records), 0, "scan 이 record 를 반환하지 않음") # ── 3. callback_launch_fn 호출 시 올바른 인자 전달 확인 ────────────────── def test_callback_launch_called_with_correct_args(self): """scan_live_inbox_once 가 callback_launch_fn 에 전달하는 인자를 캡처하여 검증. result_json_path == 실제 result 파일 경로(p) task_id == _legacy_skip_task_id(p) 값 sha256 == _sha256_file(p) 값 executor_key == scan 에 전달된 executor_key 실 _fn 을 감싸 인자를 기록 후 위임하는 wrapper 사용(단순 spy 아님 — 실 factory 경유). """ fake_fire = _FakeFireFn() captured_calls: List[dict] = [] with tempfile.TemporaryDirectory() as root: # callback launch flag OFF — NOOP_DISABLED 반환, 하지만 인자는 _fn 에 도달함. # flag OFF 여도 _fn 은 호출되고 인자가 전달됨 — NOOP 는 내부에서 early return. _real_fn = build_callback_launch_fn( root=root, flag_reader=_cb_flag_disabled, # no-op 이면 충분(fire 0) runner_fire_fn=fake_fire, audit_path=os.path.join(root, "fake_audit.jsonl"), ) def _capturing_wrapper(result_json_path, *, task_id, sha256, executor_key="", schedule_id="", terminal_intent="", relay_required=False): """실 _fn 을 감싸 인자 캡처 후 위임.""" captured_calls.append({ "result_json_path": result_json_path, "task_id": task_id, "sha256": sha256, "executor_key": executor_key, }) return _real_fn( result_json_path, task_id=task_id, sha256=sha256, executor_key=executor_key, schedule_id=schedule_id, terminal_intent=terminal_intent, relay_required=relay_required, ) p = _write_live_result(root, "task-2763-z") expected_task_id = drv._legacy_skip_task_id(p) expected_sha = drv._sha256_file(p) ledger = os.path.join(root, "ledger.jsonl") pdir = os.path.join(root, "processed") gate = _Gate(ready=True) try: drv.scan_live_inbox_once( root, flag_reader=_enabled, owner_gate_fn=gate, callback_launch_fn=_capturing_wrapper, executor_key="fake-executor-key", paths=[p], ledger_path=ledger, processed_dir=pdir, ) except TypeError as e: self.fail(f"wrapper 호출 중 TypeError: {e}") # _capturing_wrapper 가 호출됐는지 확인 self.assertEqual(len(captured_calls), 1, "callback_launch_fn 이 정확히 1회 호출돼야 한다") call = captured_calls[0] # result_json_path == 실제 파일 경로 self.assertEqual( call["result_json_path"], p, f"result_json_path 불일치: expected={p}, got={call['result_json_path']}", ) # task_id == _legacy_skip_task_id(p) self.assertEqual( call["task_id"], expected_task_id, f"task_id 불일치: expected={expected_task_id}, got={call['task_id']}", ) # sha256 == _sha256_file(p) (파일 이동 전이므로 가능; 이미 move 됐으면 pdir 에서 확인) self.assertEqual( call["sha256"], expected_sha, f"sha256 불일치: expected={expected_sha}, got={call['sha256']}", ) # executor_key == 전달된 값 self.assertEqual( call["executor_key"], "fake-executor-key", f"executor_key 불일치, got={call['executor_key']}", ) # ── 4. fake runner_fire_fn — 실 cron/subprocess 발사 0 보장 ────────────── def test_fire_fn_is_fake_no_real_cron(self): """모든 테스트에서 runner_fire_fn 이 fake 임을 보장하는 가드 테스트. fake_fire.calls 로 호출 기록만 남기고, 실 cron/systemd/subprocess 는 0. _FakeFireFn 이 실 anu_runner_pickup_and_fire 와 다름을 증명한다. """ import dispatch.anu_owned_callback_enforcement as _real_module real_fire = _real_module.anu_runner_pickup_and_fire fake_fire = _FakeFireFn() # fake 는 실 fire fn 과 다른 객체 self.assertIsNot( fake_fire, real_fire, "fake_fire 는 실 anu_runner_pickup_and_fire 와 다른 객체이어야 한다", ) self.assertNotEqual( type(fake_fire).__name__, type(real_fire).__name__, "fake_fire 는 _FakeFireFn 이고 실 fire 와 타입이 달라야 한다", ) with tempfile.TemporaryDirectory() as root: # flag ON + fake_fire 주입으로 빌드 _fn = build_callback_launch_fn( root=root, flag_reader=_cb_flag_enabled, runner_fire_fn=fake_fire, audit_path=os.path.join(root, "fake_audit.jsonl"), ) p = os.path.join(root, "task-2763-guard.result.json") with open(p, "w") as fh: fh.write('{"task_id":"task-2763-guard"}') result = _fn(p, task_id="task-2763-guard", sha256="deadbeef00") # fake_fire 가 호출됐으면 calls 에 기록이 있음 (실 subprocess 0) if result.verdict == LAUNCH_ENFORCED: self.assertEqual(len(fake_fire.calls), 1, "LAUNCH_ENFORCED → fake_fire 1회 호출(실 cron 0)") # fake_fire.calls 에 result_json_path/executor_key/anu_key 기록만 call = fake_fire.calls[0] self.assertIn("result_json_path", call) self.assertIn("anu_key", call) # 실 subprocess/schedule 등록 없음 — 기록만 있고 부작용 0 # NOOP/FAIL 케이스도 fake_fire 가 실 발사 안 했음을 구조적으로 보장 # (anu_runner_pickup_and_fire 대신 _FakeFireFn 주입됐으므로) self.assertIs( fake_fire.__class__, _FakeFireFn, "runner_fire_fn 은 _FakeFireFn 이어야 한다(실 cron 발사 구조적 차단)", ) # launch 가 일어났어도 호출 카운트가 1 이하(중복 발사 0) self.assertLessEqual( len(fake_fire.calls), 1, f"fake_fire 가 2회 이상 호출됨(중복 발사 위험): {len(fake_fire.calls)}", ) if __name__ == "__main__": unittest.main()