# -*- coding: utf-8 -*- """task-2774+8 — D-2 fire wiring (실발사 가능하게 하는 안전 결선) 필수 regression. 목적: attempt_owner_fire 의 production 실발사 fire_fn 결선 공백 + ONESHOT_TARGET 하드코딩을 닫은 안전 결선(run_approved_oneshot_fire 전용 entrypoint + build_oneshot_cokacdir_fire_fn production fire_fn factory)이 **우회 없이** 동작하되 본 PR 에서는 real fire 0 임을 입증한다. ★ 이 PR 은 real fire 를 쏘지 않는다 — 본 테스트는 mock fire_fn / mock flag_reader / mock runner 만 사용한다. 실제 cokacdir --cron subprocess 호출 0(subprocess.run/Popen 미호출 검증 포함). ACTIVE gate 는 함수인자(flag_reader)로만 판정 — 전역 flag 파일 생성/읽기 0. ★ raw ANU key literal hygiene: ANU key / executor self-key 는 callback_owner_enforcer (무수정)에서 import 한다 — 테스트 소스에 raw key literal 0. 회장 필수 테스트(task md §4 11종) 매핑: 1. fire_fn 미주입 → FIRE_BLOCKED_NO_FIRE_FN → Test01 2. target 정확 일치 1건만 → mock fire_fn 정확히 1회 호출 → Test02 3. 동일 target 2회차 → DEDUP/BLOCK, fire_fn 미호출 → Test03 4. target 미일치(task_id/head_sha/abs path 중 1개라도) → fire 0 → Test04 5. ACTIVE gate OFF → NOOP_DISABLED, fire_fn 미호출 → Test05 6. owner-proof gate 미통과 → fire_fn 미호출 → Test06 7. raw key 가 argv/log/audit 에 포함되면 FAIL(노출 0 검증) → Test07 8. 전역 flag 파일 생성 0(함수인자 격리 검증) → Test08 9. canonical backlog 13건 → 처리 대상 0(target 외 fire 0) → Test09 10. target manifest/인자 누락 시 fail-closed(FIRE_BLOCKED_NO_APPROVAL) → Test10 11. 실제 cokacdir --cron 호출 0 — mock fire_fn 으로만·real fire 0 → Test11 + wiring factory 입증: build_oneshot_cokacdir_fire_fn --once 강제·runner 격리 → Test12 """ from __future__ import annotations import importlib.util import json import os import sys import tempfile import unittest from pathlib import Path from unittest import mock _ROOT = Path(__file__).resolve().parent.parent.parent if str(_ROOT) not in sys.path: sys.path.insert(0, str(_ROOT)) def _load(modname: str, relpath: str): """실 dispatch 모듈을 canonical 이름으로 직접 로드(tests/dispatch shadowing 회피).""" if modname in sys.modules: return sys.modules[modname] spec = importlib.util.spec_from_file_location(modname, _ROOT / relpath) assert spec is not None and spec.loader is not None mod = importlib.util.module_from_spec(spec) sys.modules[modname] = mod spec.loader.exec_module(mod) return mod # 의존 chain 선로딩 → gate 모듈 로드(2742 패턴 동일). _ENF = _load("dispatch.callback_owner_enforcer", "dispatch/callback_owner_enforcer.py") _load("dispatch.normal_fallback_callback_helper", "dispatch/normal_fallback_callback_helper.py") _load("dispatch.anu_owned_callback_enforcement", "dispatch/anu_owned_callback_enforcement.py") _load("dispatch.anu_collector_result", "dispatch/anu_collector_result.py") gate_mod = _load("dispatch.anu_result_pickup_runner", "dispatch/anu_result_pickup_runner.py") # 테스트용 key (raw ANU key literal 0 — import 상수로만 취득) ANU_KEY = _ENF.ANU_KEY_2553 EXECUTOR_KEY = _ENF.EXECUTOR_SELF_KEY_EXAMPLE TARGET_TASK_ID = gate_mod.ONESHOT_TARGET_TASK_ID # head SHA 값(full 40자 hex). H = 현재 canonical HEAD 역할. H2 = HEAD 이동(stale). H = "d18358f3829d4a1278b58d88bb86f89cb536d2fe" H2 = "deadbeefdeadbeefdeadbeefdeadbeefdeadbeef" SHORT = H[:8] def _anu_loader(): return ANU_KEY def _probe_H(): return H def _flag_on(): return True def _flag_off(): return False def _fire_spy(spawned=True, callback_evidence=True, cron_registered=True): """mock fire_fn — 호출 argv 를 calls 에 기록(실 cokacdir/cron subprocess 0).""" calls = [] def _fn(argv): calls.append(list(argv)) return {"spawned": spawned, "callback_evidence": callback_evidence, "cron_registered": cron_registered, "schedule_id": "sched-mock-1"} return calls, _fn class _Base(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp(prefix="t2774p8_") self.addCleanup(lambda: __import__("shutil").rmtree(self.tmp, ignore_errors=True)) self.ledger = os.path.join(self.tmp, "owner_pickup_gate_index.jsonl") def _write_result(self, *, task_id=TARGET_TASK_ID, head_sha=H, authoritative=False, owner_pickup_required=True, owner_key_proof_present=False, schedule_created_by_executor=False, callback_fired_by_executor=False, fname=None): payload = { "task_id": task_id, "head_sha": head_sha, "authoritative": authoritative, "owner_pickup_required": owner_pickup_required, "owner_key_proof_present": owner_key_proof_present, "schedule_created_by_executor": schedule_created_by_executor, "callback_fired_by_executor": callback_fired_by_executor, } p = os.path.join(self.tmp, fname or f"{task_id}.result.json") with open(p, "w", encoding="utf-8") as f: json.dump(payload, f, ensure_ascii=False) return p def _target_for(self, path, *, task_id=TARGET_TASK_ID): return gate_mod.OneShotTarget(task_id=task_id, result_json_path=path) def _ready_gate(self, path, *, probe=_probe_H, **kw): kw.setdefault("sealed_key_loader", _anu_loader) kw.setdefault("dedupe_ledger_path", self.ledger) kw.setdefault("executor_key", EXECUTOR_KEY) g = gate_mod.owner_proof_pickup_gate(path, head_probe=probe, **kw) self.assertEqual(g.status, gate_mod.GATE_OWNER_PICKUP_READY_NO_FIRE, msg=f"전제: gate READY 여야 함 ({g.status} / {g.reasons})") return g def _gate_raw(self, path, *, probe=_probe_H, **kw): kw.setdefault("sealed_key_loader", _anu_loader) kw.setdefault("dedupe_ledger_path", self.ledger) kw.setdefault("executor_key", EXECUTOR_KEY) return gate_mod.owner_proof_pickup_gate(path, head_probe=probe, **kw) def _run(self, gate_result, *, flag_reader, fire_fn, target, probe=_probe_H, loader=_anu_loader, **kw): kw.setdefault("dedupe_ledger_path", self.ledger) kw.setdefault("executor_key", EXECUTOR_KEY) return gate_mod.run_approved_oneshot_fire( gate_result, flag_reader=flag_reader, fire_fn=fire_fn, target=target, sealed_key_loader=loader, head_probe=probe, **kw) # ── Test01: fire_fn 미주입 → FIRE_BLOCKED_NO_FIRE_FN (gate ON·fire_fn None) ─────── class Test01_NoFireFn(_Base): def test_no_fire_fn_fail_closed(self): p = self._write_result() g = self._ready_gate(p) out = self._run(g, flag_reader=_flag_on, fire_fn=None, target=self._target_for(p)) self.assertEqual(out.status, gate_mod.GATE_FIRE_BLOCKED_NO_FIRE_FN) self.assertTrue(out.fire_blocked) self.assertIsNone(out.argv, "차단 시 argv 미구성(raw key 0)") self.assertFalse(os.path.exists(self.ledger), "fire_fn 미주입 → fire 0·재발사 marker 미작성") # ── Test02: target 정확 일치 1건 → mock fire_fn 정확히 1회 호출 ─────────────────── class Test02_ExactTargetFiresOnce(_Base): def test_exact_match_fires_once(self): p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() out = self._run(g, flag_reader=_flag_on, fire_fn=fn, target=self._target_for(p)) self.assertEqual(out.status, gate_mod.GATE_REALFIRE_1SHOT_SUCCESS) self.assertEqual(len(calls), 1, "주입형 mock fire_fn 정확히 1회 호출") self.assertFalse(out.fire_blocked) # ── Test03: 동일 target 2회차 → DEDUP/BLOCK, fire_fn 미호출 ─────────────────────── class Test03_SecondCallDeduped(_Base): def test_refire_deduped(self): p = self._write_result() g1 = self._ready_gate(p) calls1, fn1 = _fire_spy() first = self._run(g1, flag_reader=_flag_on, fire_fn=fn1, target=self._target_for(p)) self.assertEqual(first.status, gate_mod.GATE_REALFIRE_1SHOT_SUCCESS) self.assertEqual(len(calls1), 1) # 2회차: 동일 (task_id, head_sha) marker 존재 → gate 자체가 DEDUPED 반환. # run_approved_oneshot_fire → attempt_owner_fire 가 DEDUPED 전파·fire_fn 미호출. g2 = self._gate_raw(p) self.assertEqual(g2.status, gate_mod.GATE_DEDUPED, "전제: 1회 소진 후 gate DEDUPED") calls2, fn2 = _fire_spy() second = self._run(g2, flag_reader=_flag_on, fire_fn=fn2, target=self._target_for(p)) self.assertEqual(second.status, gate_mod.GATE_DEDUPED) self.assertEqual(calls2, [], "2회차 → DEDUP/BLOCK·fire_fn 미호출(재발사 0)") # ── Test04: target 미일치(task_id/path/head 1개라도) → fail-closed, fire_fn 미호출 ─ class Test04_TargetMismatchBlocked(_Base): def test_task_id_mismatch_blocked(self): p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() # target.task_id 다름 → (a) 불일치 → FIRE_BLOCKED_NO_APPROVAL, fire 0. out = self._run(g, flag_reader=_flag_on, fire_fn=fn, target=self._target_for(p, task_id="task-other-999")) self.assertEqual(out.status, gate_mod.GATE_FIRE_BLOCKED_NO_APPROVAL) self.assertEqual(calls, [], "task_id mismatch → fire_fn 미호출(cron 0)") def test_abs_path_mismatch_blocked(self): p = self._write_result() other = self._write_result(fname="other.result.json") g = self._ready_gate(p) calls, fn = _fire_spy() # target.result_json_path 다름 → (b) 불일치 → FIRE_BLOCKED_NO_APPROVAL, fire 0. out = self._run(g, flag_reader=_flag_on, fire_fn=fn, target=self._target_for(other)) self.assertEqual(out.status, gate_mod.GATE_FIRE_BLOCKED_NO_APPROVAL) self.assertEqual(calls, [], "abs path mismatch → fire_fn 미호출(cron 0)") def test_head_sha_mismatch_blocked(self): # result.head_sha != current canonical HEAD(probe=H2) → (c) 불일치. # attempt_owner_fire(무수정·task-2742) 의 stale/superseded 정밀 enum 은 # RESULT_NOT_PICKUP_ELIGIBLE 이다(== only·prefix 완화 0). 어느 enum 이든 fire 0. p = self._write_result(head_sha=H) g = self._ready_gate(p, probe=_probe_H) calls, fn = _fire_spy() out = self._run(g, flag_reader=_flag_on, fire_fn=fn, target=self._target_for(p), probe=lambda: H2) self.assertIn(out.status, (gate_mod.GATE_RESULT_NOT_PICKUP_ELIGIBLE, gate_mod.GATE_FIRE_BLOCKED_NO_APPROVAL)) self.assertTrue(out.fire_blocked) self.assertEqual(calls, [], "head_sha mismatch(stale) → fire_fn 미호출(cron 0)") def test_short_sha_prefix_blocked(self): # result.head_sha 가 SHORT(8자 prefix) — len==40 정확 일치 위반 → fire 0. p = self._write_result(head_sha=SHORT) g = self._gate_raw(p, probe=_probe_H) calls, fn = _fire_spy() out = self._run(g, flag_reader=_flag_on, fire_fn=fn, target=self._target_for(p)) self.assertTrue(out.fire_blocked) self.assertEqual(calls, [], "short SHA/prefix → fire_fn 미호출(완화 0·cron 0)") # ── Test05: ACTIVE gate OFF → NOOP_DISABLED, fire_fn 미호출 ─────────────────────── class Test05_ActiveGateOff(_Base): def test_gate_off_noop(self): p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() out = self._run(g, flag_reader=_flag_off, fire_fn=fn, target=self._target_for(p)) self.assertEqual(out.status, gate_mod.GATE_NOOP_DISABLED) self.assertEqual(calls, [], "ACTIVE gate OFF → fire_fn 미호출(실 발사 0)") self.assertIsNone(out.argv) def test_flag_reader_absent_noop(self): # flag_reader 부재(None) → 보수적 OFF → NOOP_DISABLED. p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() out = self._run(g, flag_reader=None, fire_fn=fn, target=self._target_for(p)) self.assertEqual(out.status, gate_mod.GATE_NOOP_DISABLED) self.assertEqual(calls, [], "flag_reader 부재 → fire_fn 미호출") def test_flag_reader_exception_noop(self): # flag_reader 예외 → 보수적 OFF(fail-closed) → NOOP_DISABLED. def _boom(): raise RuntimeError("flag read error") p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() out = self._run(g, flag_reader=_boom, fire_fn=fn, target=self._target_for(p)) self.assertEqual(out.status, gate_mod.GATE_NOOP_DISABLED) self.assertEqual(calls, [], "flag_reader 예외 → fail-closed·fire_fn 미호출") # ── Test06: owner-proof gate 미통과 → fire_fn 미호출 ───────────────────────────── class Test06_OwnerProofFail(_Base): def test_ineligible_result_no_fire(self): # authoritative=true → eligibility 위반 → gate RESULT_NOT_PICKUP_ELIGIBLE. # run_approved_oneshot_fire(gate ON·target 일치)도 attempt_owner_fire 재확인에서 # non-READY 전파 → fire_fn 미호출. p = self._write_result(authoritative=True) g = self._gate_raw(p) self.assertNotEqual(g.status, gate_mod.GATE_OWNER_PICKUP_READY_NO_FIRE) calls, fn = _fire_spy() out = self._run(g, flag_reader=_flag_on, fire_fn=fn, target=self._target_for(p)) self.assertTrue(out.fire_blocked) self.assertEqual(calls, [], "owner-proof 미통과 → fire_fn 미호출") def test_self_collector_key_no_fire(self): # 런타임 sealed key == executor key → self-collector → FAIL_CLOSED_NON_ANU_KEY. p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() out = self._run(g, flag_reader=_flag_on, fire_fn=fn, target=self._target_for(p), loader=lambda: EXECUTOR_KEY) self.assertTrue(out.fire_blocked) self.assertEqual(calls, [], "self-collector(runtime==executor key) → fire_fn 미호출") # ── Test07: raw key 가 argv/log/audit 에 포함되면 FAIL(노출 0 검증) ────────────── class Test07_RawKeyHygiene(_Base): def test_no_raw_key_in_persisted_result_or_ledger(self): p = self._write_result() g = self._ready_gate(p) captured, fn = _fire_spy() out = self._run(g, flag_reader=_flag_on, fire_fn=fn, target=self._target_for(p)) self.assertEqual(out.status, gate_mod.GATE_REALFIRE_1SHOT_SUCCESS) # persisted result(to_json) 에 raw ANU key literal 0 — placeholder 만. persisted = json.dumps(out.to_json(), ensure_ascii=False) self.assertNotIn(ANU_KEY, persisted, "persisted result.argv/reasons 에 raw key 노출 0") self.assertIn(gate_mod.OWNER_PICKUP_KEY_PLACEHOLDER, persisted, "persist argv 는 placeholder key 사용") # dedupe ledger marker 에도 raw key literal 0. with open(self.ledger, encoding="utf-8") as f: ledger_txt = f.read() self.assertNotIn(ANU_KEY, ledger_txt, "ledger marker 에 raw key 노출 0") # 경계 검증: 실 key 는 fire_fn(주입 subprocess 경계)에만 도달한다 — redaction 경계 입증. self.assertEqual(len(captured), 1) self.assertIn(ANU_KEY, captured[0], "실 key 는 fire_fn real_argv 경계에만 — persist 와 분리됨") # ── Test08: 전역 flag 파일 생성 0(함수인자 격리 검증) ──────────────────────────── class Test08_NoGlobalFlagFile(_Base): def test_function_arg_gate_creates_no_global_flag(self): # ACTIVE gate 는 함수인자 flag_reader 로만 판정 — 전역 flag 파일 생성/읽기 0. forbidden = [ os.path.join(gate_mod.CANONICAL_ROOT, "memory", "state", "p0b_callback_launch_enabled"), os.path.join(gate_mod.CANONICAL_ROOT, "memory", "state", "p0b_driver_enabled"), os.path.join(gate_mod.CANONICAL_ROOT, "memory", "state", "p0b_real_wake_enabled"), ] before = {fp: os.path.exists(fp) for fp in forbidden} spy = {"calls": 0} def _flag_spy(): spy["calls"] += 1 return True p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() out = self._run(g, flag_reader=_flag_spy, fire_fn=fn, target=self._target_for(p)) self.assertEqual(out.status, gate_mod.GATE_REALFIRE_1SHOT_SUCCESS) self.assertEqual(len(calls), 1, "함수인자 gate 통과 → mock fire_fn 1회") self.assertGreaterEqual(spy["calls"], 1, "ACTIVE gate 는 함수인자 flag_reader 로만 판정") # 본 호출이 전역 flag 파일을 새로 만들지 않았는지(존재 상태 불변) 검증. for fp in forbidden: self.assertEqual(os.path.exists(fp), before[fp], f"전역 flag 파일 생성 0 위반: {fp}") # ── Test09: canonical backlog 13건 → 처리 대상 0(target 외 fire 0) ──────────────── class Test09_BacklogNotProcessed(_Base): def test_only_explicit_target_fires_backlog_blocked(self): # 승인 target 1건만 fire 가능. 그 외 backlog 항목(다른 task_id)은 전부 차단. approved_path = self._write_result(task_id=TARGET_TASK_ID, fname="approved.result.json") target = self._target_for(approved_path, task_id=TARGET_TASK_ID) for i in range(13): bp = self._write_result(task_id=f"task-backlog-{i:04d}", fname=f"backlog-{i:04d}.result.json") g = self._ready_gate(bp) calls, fn = _fire_spy() out = self._run(g, flag_reader=_flag_on, fire_fn=fn, target=target) self.assertEqual(out.status, gate_mod.GATE_FIRE_BLOCKED_NO_APPROVAL, f"backlog[{i}] target 외 → FIRE_BLOCKED_NO_APPROVAL") self.assertEqual(calls, [], f"backlog[{i}] → fire_fn 미호출(target 외 fire 0)") # ── Test10: target 명시 주입 부재 → FIRE_BLOCKED_NO_APPROVAL(하드코딩 자동발사 차단) ─ class Test10_MissingTargetFailClosed(_Base): def test_target_none_blocked(self): p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() out = self._run(g, flag_reader=_flag_on, fire_fn=fn, target=None) self.assertEqual(out.status, gate_mod.GATE_FIRE_BLOCKED_NO_APPROVAL) self.assertEqual(calls, [], "target 미주입 → fire_fn 미호출(하드코딩 자동발사 차단)") # FIRE_BLOCKED_NO_APPROVAL 은 gate 의 placeholder argv 를 보존한다(raw key 0). self.assertNotIn(ANU_KEY, json.dumps(out.to_json(), ensure_ascii=False), "차단 결과에 raw key 노출 0(placeholder 만)") # ── Test11: 실제 cokacdir --cron 호출 0 — mock fire_fn 으로만·real fire 0 ───────── class Test11_NoRealCokacdirCall(_Base): def test_full_success_path_does_not_spawn_subprocess(self): # 전체 success 경로(gate ON·target 일치·mock fire_fn)에서 subprocess.run/Popen # 호출 0 — 실 cokacdir --cron 발사 0 입증. p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() with mock.patch("subprocess.run") as m_run, mock.patch("subprocess.Popen") as m_popen: out = self._run(g, flag_reader=_flag_on, fire_fn=fn, target=self._target_for(p)) self.assertEqual(out.status, gate_mod.GATE_REALFIRE_1SHOT_SUCCESS) self.assertEqual(len(calls), 1, "mock fire_fn 1회만") m_run.assert_not_called() m_popen.assert_not_called() def test_blocked_paths_do_not_spawn_subprocess(self): p = self._write_result() g = self._ready_gate(p) _, fn = _fire_spy() with mock.patch("subprocess.run") as m_run, mock.patch("subprocess.Popen") as m_popen: self._run(g, flag_reader=_flag_off, fire_fn=fn, target=self._target_for(p)) self._run(g, flag_reader=_flag_on, fire_fn=None, target=self._target_for(p)) self._run(g, flag_reader=_flag_on, fire_fn=fn, target=None) m_run.assert_not_called() m_popen.assert_not_called() # ── Test12: build_oneshot_cokacdir_fire_fn factory 입증(--once 강제·runner 격리) ── class Test12_FireFnFactory(_Base): def test_factory_recurring_guard_no_once(self): # argv 에 --once 부재 → recurring 위험 → 실행 0·fail-closed(cron_registered=False). runner_calls = [] fn = gate_mod.build_oneshot_cokacdir_fire_fn(runner=lambda a: runner_calls.append(a)) out = fn(["cokacdir", "--cron", "x", "--at", "2m"]) # --once 없음 self.assertFalse(out["cron_registered"]) self.assertEqual(out.get("error"), "ONESHOT_FIRE_ARGV_INVALID") self.assertEqual(runner_calls, [], "--once 부재 → runner 미호출(실행 0)") def test_factory_runner_injection_no_real_subprocess(self): # runner 주입 → 실 subprocess 대신 mock 사용. --cron + --once 형태에서만 위임. seen = [] def _runner(argv): seen.append(list(argv)) return {"spawned": True, "cron_registered": True, "callback_evidence": True, "schedule_id": "sid-x"} fn = gate_mod.build_oneshot_cokacdir_fire_fn(runner=_runner) with mock.patch("subprocess.run") as m_run, mock.patch("subprocess.Popen") as m_popen: out = fn(["cokacdir", "--cron", "x", "--at", "2m", "--once", "--key", ANU_KEY]) self.assertEqual(len(seen), 1, "runner 정확히 1회 위임") self.assertTrue(out["cron_registered"]) m_run.assert_not_called() m_popen.assert_not_called() def test_factory_stdout_parse_no_key_leak(self): # cokacdir JSON stdout 파싱 → schedule_id 만 취득(stdout 전체 미보존·raw key 0). out = gate_mod._parse_cokacdir_cron_stdout( '{"status":"ok","id":"sched-42","prompt":"[redacted]"}') self.assertTrue(out["cron_registered"]) self.assertEqual(out["schedule_id"], "sched-42") self.assertFalse(out["spawned"], "등록 시점 spawn 미관측(late spawn → reconcile)") # ── Test13: cron 등록 fail-closed (task-2774+9) — id 부재/빈값/None → registered=False ── # status=="ok" 이라도 registration response 의 id 가 실제 존재+비어있지 않을 때만 등록 성공. # real fire 직전 미등록을 성공으로 기록하면 dedupe/duplicate fire/teardown 판단이 흐려진다. class Test13_CronIdFailClosed(_Base): def test_id_key_absent_fail_closed(self): # id 키 자체 없음 → registered=False·FIRE_BLOCKED_NO_CRON_ID·schedule_id 빈값. out = gate_mod._parse_cokacdir_cron_stdout('{"status":"ok"}') self.assertFalse(out["cron_registered"], "id 부재 → fail-closed") self.assertEqual(out["schedule_id"], "") self.assertEqual(out.get("error"), gate_mod.FIRE_BLOCKED_NO_CRON_ID) def test_id_empty_string_fail_closed(self): out = gate_mod._parse_cokacdir_cron_stdout('{"status":"ok","id":""}') self.assertFalse(out["cron_registered"], "id='' → fail-closed") self.assertEqual(out["schedule_id"], "") self.assertEqual(out.get("error"), gate_mod.FIRE_BLOCKED_NO_CRON_ID) def test_id_whitespace_only_fail_closed(self): # 공백뿐인 id → strip 후 빈값 → fail-closed(빈 등록 false-positive 차단). out = gate_mod._parse_cokacdir_cron_stdout('{"status":"ok","id":" "}') self.assertFalse(out["cron_registered"]) self.assertEqual(out["schedule_id"], "") self.assertEqual(out.get("error"), gate_mod.FIRE_BLOCKED_NO_CRON_ID) def test_id_none_fail_closed(self): out = gate_mod._parse_cokacdir_cron_stdout('{"status":"ok","id":null}') self.assertFalse(out["cron_registered"], "id=None → fail-closed") self.assertEqual(out["schedule_id"], "") self.assertEqual(out.get("error"), gate_mod.FIRE_BLOCKED_NO_CRON_ID) def test_valid_id_registered_true(self): # 정상 id 있을 때만 registered=True·error 키 없음(기존 성공 경로 유지). out = gate_mod._parse_cokacdir_cron_stdout('{"status":"ok","id":"sched-9"}') self.assertTrue(out["cron_registered"], "유효 id → 성공 경로 유지") self.assertEqual(out["schedule_id"], "sched-9") self.assertNotIn("error", out, "성공 경로엔 fail-closed 라벨 없음") def test_status_not_ok_fail_closed(self): # status!=ok → FIRE_REGISTRATION_FAILED(비정상 응답 fail-closed). out = gate_mod._parse_cokacdir_cron_stdout('{"status":"error","message":"x"}') self.assertFalse(out["cron_registered"]) self.assertEqual(out.get("error"), gate_mod.FIRE_REGISTRATION_FAILED) def test_unparseable_stdout_fail_closed(self): out = gate_mod._parse_cokacdir_cron_stdout("not-json-at-all") self.assertFalse(out["cron_registered"]) self.assertEqual(out["schedule_id"], "") self.assertEqual(out.get("error"), gate_mod.FIRE_REGISTRATION_FAILED) def test_failclosed_normalized_by_read_fire_outcome(self): # _read_fire_outcome 정규화 단계에서도 cron_registered=False 유지(오판 0). raw = gate_mod._parse_cokacdir_cron_stdout('{"status":"ok"}') outcome = gate_mod._read_fire_outcome(raw) cron_registered, schedule_id = outcome[2], outcome[3] self.assertFalse(cron_registered, "id 부재 응답이 등록 성공으로 오판되지 않음") self.assertEqual(schedule_id, "") def test_id_absent_not_misjudged_as_fire_success(self): # 전체 흐름: id 없는 ok 응답 → PARTIAL_FIRE_NO_SPAWN(terminal)·REALFIRE 아님· # fire_blocked=True. dedupe/fire success 오판 0. 실 cokacdir subprocess 0. p = self._write_result() g = self._ready_gate(p) seen = [] def _runner(argv): seen.append(list(argv)) # cokacdir 가 id 없이 ok 만 반환한 상황 시뮬레이션(실 subprocess 0). return gate_mod._parse_cokacdir_cron_stdout('{"status":"ok"}') fn = gate_mod.build_oneshot_cokacdir_fire_fn(runner=_runner) with mock.patch("subprocess.run") as m_run, mock.patch("subprocess.Popen") as m_popen: out = self._run(g, flag_reader=_flag_on, fire_fn=fn, target=self._target_for(p)) self.assertEqual(out.status, gate_mod.GATE_PARTIAL_FIRE_NO_SPAWN, "id 없는 등록은 PARTIAL_FIRE_NO_SPAWN(등록 실패)로 기록") self.assertNotEqual(out.status, gate_mod.GATE_REALFIRE_1SHOT_SUCCESS, "미등록을 fire 성공으로 기록 금지(duplicate fire 0)") self.assertTrue(out.fire_blocked, "미등록 → fire_blocked=True(성공 기록 0)") self.assertEqual(len(seen), 1, "runner 정확히 1회 위임(중복 발사 0)") m_run.assert_not_called() m_popen.assert_not_called() def test_no_raw_key_leak_in_failclosed_outcome(self): # fail-closed outcome 에 stdout/원문(키 유사 토큰) 미포함 — 고정 라벨·빈 schedule_id 만. leaky = '{"status":"ok","prompt":"--key SECRET_KEY_TOKEN_XYZ"}' out = gate_mod._parse_cokacdir_cron_stdout(leaky) self.assertFalse(out["cron_registered"]) for v in out.values(): self.assertNotIn("SECRET_KEY_TOKEN_XYZ", str(v), "raw stdout 미노출(raw key 0)") self.assertEqual(out.get("error"), gate_mod.FIRE_BLOCKED_NO_CRON_ID) # ── Test14: stdout bytes 인코딩 fail-closed (task-2774+10) ─────────────────────────── # subprocess.run 에서 text=True 를 제거하고 stdout 을 raw bytes 로 캡처하는 변경이 # _parse_cokacdir_cron_stdout(bytes + errors="replace" fail-closed)에 위임되어 # 회귀 없이 동작하는지 bytes 입력 전 경로를 직접 검증한다. class Test14_StdoutBytesEncodingFailClosed(_Base): def test_bytes_stdout_valid_id_parses(self): # bytes stdout 에서 정상 id 파싱 성공 — text=True 제거 후 bytes 경로 핵심 검증. out = gate_mod._parse_cokacdir_cron_stdout(b'{"status":"ok","id":"sched-bytes-1"}') self.assertTrue(out["cron_registered"], "bytes 입력에서도 정상 id → registered=True") self.assertEqual(out["schedule_id"], "sched-bytes-1") self.assertNotIn("error", out, "성공 경로엔 fail-closed 라벨 없음") def test_bytes_stdout_invalid_utf8_no_decode_error(self): # UTF-8 디코딩 불가 바이트를 앞에 두고 개행으로 분리 → 마지막 라인이 유효 JSON 이 됨. # errors="replace" 로 UnicodeDecodeError 없이 정상 처리 — fire_fn 을 깨지 않음을 입증. bad = b'\xff\xfe\n' + b'{"status":"ok","id":"sched-good"}' out = gate_mod._parse_cokacdir_cron_stdout(bad) # 반환이 dict 이고 cron_registered 키가 있어야 함 — 예외 없이 반환 입증. self.assertIsInstance(out, dict, "invalid UTF-8 바이트가 있어도 예외 없이 dict 반환") self.assertIn("cron_registered", out, "fail-closed 표준 키 보존") # 마지막 라인(유효 JSON)이 파싱되어 registered=True·id 취득 — 강한 입증. self.assertTrue(out["cron_registered"], "invalid 바이트 앞줄 후 유효 JSON 마지막 라인 → registered=True(오류 없음)") self.assertEqual(out["schedule_id"], "sched-good", "errors='replace' 처리 후 마지막 라인 id 정상 취득") def test_non_json_bytes_fail_closed(self): # bytes 이지만 JSON 파싱 불가 → FIRE_REGISTRATION_FAILED fail-closed. out = gate_mod._parse_cokacdir_cron_stdout(b'not-json-bytes-at-all') self.assertFalse(out["cron_registered"], "비JSON bytes → fail-closed(registered=False)") self.assertEqual(out.get("error"), gate_mod.FIRE_REGISTRATION_FAILED, "비JSON bytes → FIRE_REGISTRATION_FAILED 고정 라벨") def test_empty_bytes_fail_closed(self): # 빈 bytes → 라인 없음 → FIRE_REGISTRATION_FAILED fail-closed. out = gate_mod._parse_cokacdir_cron_stdout(b'') self.assertFalse(out["cron_registered"], "빈 bytes → fail-closed(registered=False)") self.assertEqual(out.get("error"), gate_mod.FIRE_REGISTRATION_FAILED, "빈 bytes stdout → FIRE_REGISTRATION_FAILED 고정 라벨") def test_bytes_id_absent_registered_false(self): # bytes 입력·status=ok 이나 id 없음 → FIRE_BLOCKED_NO_CRON_ID fail-closed. out = gate_mod._parse_cokacdir_cron_stdout(b'{"status":"ok"}') self.assertFalse(out["cron_registered"], "bytes 입력 id 부재 → registered=False") self.assertEqual(out.get("error"), gate_mod.FIRE_BLOCKED_NO_CRON_ID, "id 없음 → FIRE_BLOCKED_NO_CRON_ID(미등록 성공 오판 차단)") self.assertEqual(out["schedule_id"], "", "id 부재 시 schedule_id 빈값") def test_fire_fn_with_bytes_runner_no_real_subprocess(self): # runner 가 bytes stdout 을 시뮬레이션: _parse_cokacdir_cron_stdout(bytes) 반환. # build_oneshot_cokacdir_fire_fn(runner=_runner) 경유 시 bytes 처리 정상 동작 검증. runner_calls = [] def _runner(argv): runner_calls.append(list(argv)) # bytes stdout 시뮬레이션: parser 를 bytes 입력으로 직접 호출해 반환. return gate_mod._parse_cokacdir_cron_stdout(b'{"status":"ok","id":"sched-b9"}') fn = gate_mod.build_oneshot_cokacdir_fire_fn(runner=_runner) argv = ["cokacdir", "--cron", "x", "--at", "2m", "--once", "--key", ANU_KEY] with mock.patch("subprocess.run") as m_run, mock.patch("subprocess.Popen") as m_popen: out = fn(argv) # 실 subprocess 0 — runner 로만 처리. m_run.assert_not_called() m_popen.assert_not_called() # runner 정확히 1회, 결과 cron_registered=True. self.assertEqual(len(runner_calls), 1, "runner 정확히 1회 호출(중복 발사 0)") self.assertTrue(out["cron_registered"], "bytes runner 통과 후 registered=True(bytes 경로 정상 연결)") self.assertEqual(out["schedule_id"], "sched-b9") def test_stderr_bytes_no_raw_key_leak(self): # stderr 에 bytes(키 유사 토큰 포함) 상황 시뮬레이션 — raw key 노출 0 검증. # _parse_cokacdir_cron_stdout 가 prompt/raw 필드를 return 에 담지 않음을 입증. leaky_bytes = b'{"status":"ok","prompt":"--key SECRET_BYTES_TOKEN_ZZZ"}' out = gate_mod._parse_cokacdir_cron_stdout(leaky_bytes) # id 부재(prompt 필드만) → fail-closed(FIRE_BLOCKED_NO_CRON_ID). self.assertFalse(out["cron_registered"], "id 부재 bytes 입력 → fail-closed(성공 오판 차단)") self.assertEqual(out.get("error"), gate_mod.FIRE_BLOCKED_NO_CRON_ID, "id 없는 ok 응답(bytes) → FIRE_BLOCKED_NO_CRON_ID 고정 라벨") # 핵심: 반환 dict 의 어떤 value 에도 raw key 유사 토큰 미포함(raw key hygiene). for v in out.values(): self.assertNotIn("SECRET_BYTES_TOKEN_ZZZ", str(v), f"bytes stdout 내 토큰이 반환 value 에 노출됨: {v!r}") # ── Test_InvalidFlagReader (task-2774+11): non-callable flag_reader → 명시 fail-closed ── class Test09_InvalidFlagReader(_Base): def test_flag_reader_true_invalid(self): # flag_reader=True (non-callable) → silent NOOP 금지 → INVALID_FLAG_READER. p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() out = self._run(g, flag_reader=True, fire_fn=fn, target=self._target_for(p)) self.assertEqual(out.status, gate_mod.GATE_INVALID_FLAG_READER) self.assertTrue(out.fire_blocked) self.assertIsNone(out.argv, "non-callable flag_reader → argv 미구성(raw key 0)") self.assertEqual(calls, [], "non-callable flag_reader → fire_fn 미호출(실 발사 0)") def test_flag_reader_object_invalid(self): # flag_reader=object() (non-callable) → INVALID_FLAG_READER. p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() out = self._run(g, flag_reader=object(), fire_fn=fn, target=self._target_for(p)) self.assertEqual(out.status, gate_mod.GATE_INVALID_FLAG_READER) self.assertEqual(calls, [], "object() flag_reader → fire_fn 미호출") def test_flag_reader_int_zero_invalid(self): # flag_reader=0 (falsy 이지만 non-callable·None 아님) → silent NOOP 금지 → INVALID. p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() out = self._run(g, flag_reader=0, fire_fn=fn, target=self._target_for(p)) self.assertEqual(out.status, gate_mod.GATE_INVALID_FLAG_READER) self.assertEqual(calls, [], "0(falsy non-callable) → fire_fn 미호출") def test_invalid_flag_reader_no_subprocess(self): # non-callable flag_reader → 내부 real-fire subprocess.run 미호출(assert_not_called). p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() with mock.patch("subprocess.run") as m_run, mock.patch("subprocess.Popen") as m_popen: out = self._run(g, flag_reader=True, fire_fn=fn, target=self._target_for(p)) self.assertEqual(out.status, gate_mod.GATE_INVALID_FLAG_READER) m_run.assert_not_called() m_popen.assert_not_called() self.assertEqual(calls, [], "non-callable → subprocess·fire_fn 모두 미호출") def test_invalid_flag_reader_no_raw_key(self): # non-callable flag_reader 차단 결과(persist)에 raw key 노출 0. p = self._write_result() g = self._ready_gate(p) _, fn = _fire_spy() out = self._run(g, flag_reader=True, fire_fn=fn, target=self._target_for(p)) persisted = json.dumps(out.to_json(), ensure_ascii=False) self.assertNotIn(ANU_KEY, persisted, "INVALID 차단 결과에 raw key 노출 0") def test_invalid_flag_reader_in_wrapper_statuses(self): # 신규 status 가 wrapper terminal status tuple 에 포함(정합성). self.assertIn(gate_mod.GATE_INVALID_FLAG_READER, gate_mod.GATE_ONESHOT_WRAPPER_STATUSES) def test_invalid_flag_reader_no_enum_collision(self): # 신규 status 가 기존 gate/oneshot/all-terminal enum set 과 충돌 0(별개 wrapper 전용). self.assertNotIn(gate_mod.GATE_INVALID_FLAG_READER, gate_mod.GATE_TERMINAL_STATUSES) self.assertNotIn(gate_mod.GATE_INVALID_FLAG_READER, gate_mod.GATE_ONESHOT_FIRE_STATUSES) self.assertNotIn(gate_mod.GATE_INVALID_FLAG_READER, gate_mod.GATE_ALL_TERMINAL_STATUSES) # 값 자체도 기존 wrapper status 2종과 구별됨. self.assertNotEqual(gate_mod.GATE_INVALID_FLAG_READER, gate_mod.GATE_NOOP_DISABLED) self.assertNotEqual(gate_mod.GATE_INVALID_FLAG_READER, gate_mod.GATE_FIRE_BLOCKED_NO_FIRE_FN) # ── Test15: None-safety regression — _fire_blocked_no_approval / _oneshot_wrapper_block ── # # 배경(task-2774+8 백엔드 수정): # _fire_blocked_no_approval 과 _oneshot_wrapper_block 의 wrapper-block 빌더에 # `dict(gate_result.eligibility or {})` / `list(gate_result.reasons or [])` 가 추가됐다. # 수정 전이라면 eligibility=None / reasons=None 으로 생성한 OwnerProofGateResult 를 # 두 함수에 전달할 때 dict(None) / list(None) 로 TypeError 가 발생했을 것이다. # 현재 소스에서 아래 테스트가 모두 PASS 함 = None-safety 수정이 실제로 작동함을 입증. class Test15_NoneFieldsSafetyRegression(unittest.TestCase): """_fire_blocked_no_approval / _oneshot_wrapper_block 의 None 안전성 회귀 테스트. 두 빌더 함수는 fire_fn / subprocess 를 호출하지 않는 순수 block-result 빌더이므로 실 발사 경로를 일절 건드리지 않는다(Test15_8 에서 명시 입증). raw ANU key 노출 0 도 검증(Test15_9). """ # ── 내부 헬퍼 ──────────────────────────────────────────────────────────────── def _make_gate(self, *, eligibility=None, reasons=None): # -> OwnerProofGateResult """eligibility / reasons 를 파라미터로 받는 더미 OwnerProofGateResult 생성. 필수 필드(schema, status, task_id, result_json_path, head_sha, argv, fire_blocked, active, owner_key_proof_present, schedule_owner_proof, eligibility)는 모두 더미 값 — eligibility/reasons 만 파라미터화. None 을 그대로 넘겨 None-safety 를 검증하는 것이 이 헬퍼의 목적. """ return gate_mod.OwnerProofGateResult( schema=gate_mod.GATE_SCHEMA, status=gate_mod.GATE_OWNER_PICKUP_READY_NO_FIRE, task_id=TARGET_TASK_ID, result_json_path="/tmp/dummy.result.json", head_sha=H, argv=None, fire_blocked=False, active=False, owner_key_proof_present=True, schedule_owner_proof=gate_mod.GATE_OWNER_PROOF_ANU, eligibility=eligibility, # 의도적 None 허용 — None-safety 검증 대상 reasons=reasons, # 의도적 None 허용 — None-safety 검증 대상 ) # ── _fire_blocked_no_approval 테스트 ────────────────────────────────────── def test_01_fire_blocked_no_approval_eligibility_none(self): # eligibility=None → TypeError 없이 반환, out.eligibility 는 빈 dict. # (수정 전: dict(None) → TypeError. 현재 `or {}` 가 이를 막는다.) g = self._make_gate(eligibility=None, reasons=["r1"]) out = gate_mod._fire_blocked_no_approval(g, active=False) self.assertIsInstance(out.eligibility, dict) self.assertEqual(out.eligibility, {}) self.assertEqual(out.status, gate_mod.GATE_FIRE_BLOCKED_NO_APPROVAL) def test_02_fire_blocked_no_approval_reasons_none(self): # reasons=None → TypeError 없이 반환, out.reasons 는 list. # extra_reasons=["X"] 를 주면 "X" 가 포함돼야 한다. # (수정 전: list(None) → TypeError. 현재 `or []` 가 이를 막는다.) g = self._make_gate(eligibility={}, reasons=None) out = gate_mod._fire_blocked_no_approval(g, active=False, extra_reasons=["X"]) self.assertIsInstance(out.reasons, list) self.assertIn("X", out.reasons) def test_03_fire_blocked_no_approval_both_none(self): # eligibility=None AND reasons=None 동시 → TypeError 없이 안전. g = self._make_gate(eligibility=None, reasons=None) out = gate_mod._fire_blocked_no_approval(g, active=True, extra_reasons=["Y"]) self.assertIsInstance(out.eligibility, dict) self.assertIsInstance(out.reasons, list) self.assertIn("Y", out.reasons) self.assertEqual(out.status, gate_mod.GATE_FIRE_BLOCKED_NO_APPROVAL) # ── _oneshot_wrapper_block 테스트 ───────────────────────────────────────── def test_04_oneshot_wrapper_block_eligibility_none(self): # eligibility=None → TypeError 없이 반환, out.eligibility == {}. # argv None(차단), fire_blocked True, active False 확인. g = self._make_gate(eligibility=None, reasons=["r1"]) out = gate_mod._oneshot_wrapper_block( g, status=gate_mod.GATE_FIRE_BLOCKED_NO_APPROVAL, reason="test_reason" ) self.assertIsInstance(out.eligibility, dict) self.assertEqual(out.eligibility, {}) self.assertIsNone(out.argv) self.assertTrue(out.fire_blocked) self.assertFalse(out.active) def test_05_oneshot_wrapper_block_reasons_none(self): # reasons=None → TypeError 없이 반환, reason 인자가 out.reasons 에 포함. # (수정 전: list(None) → TypeError. 현재 `or []` 가 이를 막는다.) g = self._make_gate(eligibility={}, reasons=None) out = gate_mod._oneshot_wrapper_block( g, status=gate_mod.GATE_NOOP_DISABLED, reason="block_because_none" ) self.assertIsInstance(out.reasons, list) self.assertIn("block_because_none", out.reasons) def test_06_oneshot_wrapper_block_both_none(self): # eligibility=None AND reasons=None 동시 → 안전. g = self._make_gate(eligibility=None, reasons=None) out = gate_mod._oneshot_wrapper_block( g, status=gate_mod.GATE_FIRE_BLOCKED_NO_FIRE_FN, reason="both_none" ) self.assertIsInstance(out.eligibility, dict) self.assertIsInstance(out.reasons, list) self.assertIn("both_none", out.reasons) def test_07_normal_values_preserved(self): # 정상 경로 의미 무변경 입증: eligibility / reasons 정상 입력 시 두 함수 모두 보존. # None 정규화가 정상값을 훼손하지 않음. elig = {"k": "v", "flag": True} rsns = ["r1", "r2"] g = self._make_gate(eligibility=elig, reasons=rsns) out_blocked = gate_mod._fire_blocked_no_approval(g, active=False, extra_reasons=["extra"]) self.assertEqual(out_blocked.eligibility, elig) self.assertIn("r1", out_blocked.reasons) self.assertIn("r2", out_blocked.reasons) self.assertIn("extra", out_blocked.reasons) out_wrapper = gate_mod._oneshot_wrapper_block( g, status=gate_mod.GATE_FIRE_BLOCKED_NO_APPROVAL, reason="wrap_reason" ) self.assertEqual(out_wrapper.eligibility, elig) self.assertIn("r1", out_wrapper.reasons) self.assertIn("r2", out_wrapper.reasons) self.assertIn("wrap_reason", out_wrapper.reasons) def test_08_none_inputs_do_not_invoke_subprocess(self): # 두 wrapper 함수는 fire_fn / subprocess 를 호출하지 않는 순수 block 빌더임을 입증. # unittest.mock.patch 로 subprocess.run / subprocess.Popen 을 패치하고 # None 입력으로 두 함수를 호출 후 미호출 검증. g = self._make_gate(eligibility=None, reasons=None) with ( mock.patch("subprocess.run") as mock_run, mock.patch("subprocess.Popen") as mock_popen, ): gate_mod._fire_blocked_no_approval(g, active=False, extra_reasons=["test"]) gate_mod._oneshot_wrapper_block( g, status=gate_mod.GATE_NOOP_DISABLED, reason="pure_builder" ) mock_run.assert_not_called() mock_popen.assert_not_called() def test_09_none_input_no_raw_key_in_json(self): # None 입력으로 만든 결과를 json.dumps(out.to_json()) 했을 때 # ANU_KEY 가 포함되지 않음을 검증(raw key 노출 0). g = self._make_gate(eligibility=None, reasons=None) out_blocked = gate_mod._fire_blocked_no_approval(g, active=False) blocked_json = json.dumps(out_blocked.to_json(), ensure_ascii=False) self.assertNotIn(ANU_KEY, blocked_json, "_fire_blocked_no_approval 결과에 raw key 노출 0") out_wrapper = gate_mod._oneshot_wrapper_block( g, status=gate_mod.GATE_FIRE_BLOCKED_NO_APPROVAL, reason="no_key" ) wrapper_json = json.dumps(out_wrapper.to_json(), ensure_ascii=False) self.assertNotIn(ANU_KEY, wrapper_json, "_oneshot_wrapper_block 결과에 raw key 노출 0") if __name__ == "__main__": unittest.main()