# -*- coding: utf-8 -*- """task-2742-r3 — ONESHOT_TARGET head_sha 런타임 canonical HEAD 동적화 필수 regression. squash merge catch-22 해소: 이전(r2)에는 ONESHOT_TARGET.head_sha 가 하드코드 상수 ("4012bc50…")였다. task-2742 가 squash merge 되면 canonical HEAD 가 새 mergeCommit 으로 바뀌어 상수 head_sha 가 영구 stale → live real-fire 가 항상 FIRE_BLOCKED 되는 구조적 catch-22 였다. r3 부터 attempt_owner_fire 는 target head 를 **런타임 canonical HEAD read-only probe** 로 동적 판정한다. target 승인 scope 검증 3개: (a) task_id 정확 일치 (b) absolute path 정확 일치 (c) result.head_sha == current canonical HEAD **full 40자 정확 일치**(== only) ★ prefix/startswith/short SHA 완화 0 (len==40 + 문자열 동일성). canonical HEAD probe 실패/40자 아님 → fail-closed(fire 0). target path None/empty/relative → fail-closed. ★ 실 live cron fire 0(본 task): subprocess 호출부는 mock fire_fn 으로 주입한다 — 실 cokacdir/cron subprocess 호출 0(subprocess.run/Popen assert_not_called). ACTIVE=false 유지·raw ANU key literal 0(source/output/marker)·self-collector 0. 회장 필수 테스트(task md §4 / 위임 16) 매핑: 1. current_head=H, fixture H → mock fire_fn 정확히 1회 → Test06/Test01dyn 2. current_head=H2(HEAD 이동), fixture H1 → RESULT_NOT_PICKUP_ELIGIBLE → Test02dyn/Test03 3. task_id mismatch → fire 0 → Test02 4. absolute path mismatch → fire 0 → Test02 5. head_probe 실패(빈/40자 아님) → fail-closed, fire 0 → Test05dyn 6. short SHA/prefix만 일치 → fire 0 → Test06dyn 7. approved=false → FIRE_BLOCKED_NO_APPROVAL, fire 0 → Test01 8. duplicate marker → DEDUPED, fire 0 → Test04 9. sealed key 부재/dev/위조/self → FAIL_CLOSED_NON_ANU_KEY, fire 0 → Test05 10. mock argv 가 current canonical HEAD 일치 fixture만 대상·cron 1회 → Test07/Test10dyn 11. live fire 0·raw key 0·ACTIVE=false·systemd/p0b_driver 0·glob 0 → Test08 dynamic 설계 입증(하드코드 제거·full 40·prefix 완화 0): · ONESHOT_TARGET_HEAD_SHA 상수 부재 + OneShotTarget.head_sha 필드 부재 → Test11dyn · full 40자 정확 일치만 fire / 39·41자·대소문자 변형 fire 0 → Test12dyn ★ raw ANU key literal hygiene: 본 테스트는 ANU key/executor self-key 를 모두 callback_owner_enforcer(무수정) 에서 import 한다 — 테스트 소스에도 raw key literal 0. """ 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 모듈 로드. _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 DEV_SELF_KEY = "0123456789abcdef" # 비-ANU dev self-key (ANU set 에 없음) TARGET_TASK_ID = gate_mod.ONESHOT_TARGET_TASK_ID # "task-realfire-pilot-260614" # ── head SHA 값 (full 40자 hex) — r3 는 target 상수 head 가 아니라 런타임 probe 로 판정 ── # H = "현재 canonical HEAD" 역할(full 40자). probe 가 이 값을 반환하면 fresh. # H2 = HEAD 가 이동(squash merge 등)한 다른 full 40자 — stale/superseded 입증. # SHORT = H 의 8자 prefix — short SHA/prefix 완화 0 입증(정확 일치 실패). H = "d18358f3829d4a1278b58d88bb86f89cb536d2fe" # full 40자 (canonical-like) H2 = "deadbeefdeadbeefdeadbeefdeadbeefdeadbeef" # full 40자, H 와 다름 SHORT = H[:8] # "d18358f3" (8자 prefix) H_39 = H[:39] # 39자 (full 아님) H_41 = H + "a" # 41자 (full 아님) def _anu_loader(): return ANU_KEY def _probe_H(): return H def _fire_spy(spawned=True, callback_evidence=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} return calls, _fn class _Base(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp(prefix="t2742r3_") 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): # ★ r3: OneShotTarget 에는 head_sha 필드가 없다(런타임 canonical HEAD 동적 판정). return gate_mod.OneShotTarget(task_id=task_id, result_json_path=path) def _gate(self, path, *, probe=_probe_H, **kw): kw.setdefault("sealed_key_loader", _anu_loader) kw.setdefault("dedupe_ledger_path", self.ledger) return gate_mod.owner_proof_pickup_gate(path, head_probe=probe, **kw) def _ready_gate(self, path, *, probe=_probe_H, **kw): g = self._gate(path, probe=probe, executor_key=EXECUTOR_KEY, **kw) self.assertEqual(g.status, gate_mod.GATE_OWNER_PICKUP_READY_NO_FIRE, msg=f"전제: gate READY 여야 함 ({g.status} / {g.reasons})") return g def _fire(self, gate_result, target, *, fire_fn, loader=_anu_loader, probe=_probe_H, **kw): kw.setdefault("dedupe_ledger_path", self.ledger) kw.setdefault("executor_key", EXECUTOR_KEY) return gate_mod.attempt_owner_fire( gate_result, approved_one_shot=True, fire_fn=fire_fn, sealed_key_loader=loader, head_probe=probe, target=target, **kw) # ── 7. approved=false → FIRE_BLOCKED_NO_APPROVAL, cron 0 ─────────────────────── class Test01ApprovedFalseBlocked(_Base): def test_no_one_shot_flag_blocked(self): p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() with mock.patch("subprocess.run") as srun, mock.patch("subprocess.Popen") as spop: fired = gate_mod.attempt_owner_fire(g, fire_fn=fn, target=self._target_for(p)) srun.assert_not_called() spop.assert_not_called() self.assertEqual(fired.status, gate_mod.GATE_FIRE_BLOCKED_NO_APPROVAL) self.assertTrue(fired.fire_blocked) self.assertFalse(fired.active) self.assertEqual(calls, [], "approved=false → fire_fn 호출 0(cron 0)") def test_legacy_approved_only_blocked(self): p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() fired = gate_mod.attempt_owner_fire(g, approved=True, fire_fn=fn, target=self._target_for(p)) self.assertEqual(fired.status, gate_mod.GATE_FIRE_BLOCKED_NO_APPROVAL) self.assertEqual(calls, []) def test_legacy_persistent_active_still_unwired(self): # persistent ACTIVE 실 발사 결선 부재(무결선) 보존 — approved+active → RuntimeError. p = self._write_result() g = self._ready_gate(p) with self.assertRaises(RuntimeError): gate_mod.attempt_owner_fire(g, approved=True, active=True) # ── 3·4. task_id / absolute path mismatch → fire 0 ──────────────────────────── class Test02TargetMismatch(_Base): def test_task_id_mismatch_blocked(self): p = self._write_result(task_id="task-9999", fname="task-9999.result.json") g = self._ready_gate(p) calls, fn = _fire_spy() # target 은 고정 pilot task_id — gate_result.task_id 와 불일치. fired = self._fire(g, self._target_for(p, task_id=TARGET_TASK_ID), fire_fn=fn) self.assertEqual(fired.status, gate_mod.GATE_FIRE_BLOCKED_NO_APPROVAL) self.assertEqual(calls, [], "대상 task_id mismatch → fire_fn 0(cron 0)") def test_path_mismatch_blocked(self): p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() other = os.path.join(self.tmp, "elsewhere", "x.result.json") fired = self._fire(g, self._target_for(other), fire_fn=fn) self.assertEqual(fired.status, gate_mod.GATE_FIRE_BLOCKED_NO_APPROVAL) self.assertEqual(calls, [], "대상 absolute path mismatch → fire_fn 0(cron 0)") # ── 2. current_head 이동(stale) → RESULT_NOT_PICKUP_ELIGIBLE, cron 0 ────────── class Test03StaleHeadMoved(_Base): def test_head_moved_between_gate_and_fire_not_eligible(self): # gate 시점 canonical=H1(=H) → READY. 발사 직전 HEAD 가 H2 로 이동(squash merge) # → result.head_sha(H1) != current canonical(H2) → 정확 일치 실패 → NOT_ELIGIBLE. p = self._write_result(head_sha=H) g = self._ready_gate(p, probe=_probe_H) calls, fn = _fire_spy() fired = self._fire(g, self._target_for(p), fire_fn=fn, probe=lambda: H2) self.assertEqual(fired.status, gate_mod.GATE_RESULT_NOT_PICKUP_ELIGIBLE) self.assertTrue(fired.fire_blocked) self.assertFalse(fired.active) self.assertEqual(calls, [], "HEAD 이동(stale) → fire_fn 0(cron 0)") # ── 8. duplicate marker → DEDUPED, cron 0 ───────────────────────────────────── class Test04Dedupe(_Base): def test_existing_marker_deduped(self): p = self._write_result() g = self._ready_gate(p) # 이미 (task_id, current canonical HEAD) marker 존재 → 재발사 방지. with open(self.ledger, "w", encoding="utf-8") as f: f.write(json.dumps({"event": gate_mod.GATE_LEDGER_EVENT, "task_id": TARGET_TASK_ID, "head_sha": H}) + "\n") calls, fn = _fire_spy() fired = self._fire(g, self._target_for(p), fire_fn=fn) self.assertEqual(fired.status, gate_mod.GATE_DEDUPED) self.assertTrue(fired.dedupe_marker_present) self.assertEqual(calls, [], "dup marker → fire_fn 0(cron 0)") # ── 9. sealed key 부재/dev/위조/self → FAIL_CLOSED_NON_ANU_KEY, cron 0 ───────── class Test05FailClosedKey(_Base): def test_missing_key(self): p = self._write_result() g = self._ready_gate(p) for loader in (lambda: None, lambda: "", lambda: " "): calls, fn = _fire_spy() fired = self._fire(g, self._target_for(p), fire_fn=fn, loader=loader) self.assertEqual(fired.status, gate_mod.GATE_FAIL_CLOSED_NON_ANU_KEY) self.assertEqual(calls, [], "key 부재 → fire_fn 0(cron 0)") def test_dev_self_key(self): p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() fired = self._fire(g, self._target_for(p), fire_fn=fn, loader=lambda: DEV_SELF_KEY) self.assertEqual(fired.status, gate_mod.GATE_FAIL_CLOSED_NON_ANU_KEY) self.assertEqual(calls, []) def test_runtime_key_equals_executor_self_collector(self): # 런타임 key == executor key (executor self-collector) → fail-closed. p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() fired = gate_mod.attempt_owner_fire( g, approved_one_shot=True, fire_fn=fn, sealed_key_loader=_anu_loader, executor_key=ANU_KEY, # executor 가 ANU key 탈취 head_probe=_probe_H, dedupe_ledger_path=self.ledger, target=self._target_for(p)) self.assertEqual(fired.status, gate_mod.GATE_FAIL_CLOSED_NON_ANU_KEY) self.assertEqual(calls, [], "self-collector(런타임==executor key) → fire_fn 0") # ── 1. current_head=H, fixture H → fire_fn 정확히 1회 + partial/error 분기 ────── class Test06SuccessFiresExactlyOnce(_Base): def test_fire_fn_called_exactly_once(self): p = self._write_result(head_sha=H) g = self._ready_gate(p, probe=_probe_H) calls, fn = _fire_spy(spawned=True, callback_evidence=True) with mock.patch("subprocess.run") as srun, mock.patch("subprocess.Popen") as spop: fired = self._fire(g, self._target_for(p), fire_fn=fn, probe=_probe_H) srun.assert_not_called() # 실 cokacdir/cron subprocess 0 spop.assert_not_called() self.assertEqual(fired.status, gate_mod.GATE_REALFIRE_1SHOT_SUCCESS) self.assertEqual(len(calls), 1, "주입형 fire_fn 정확히 1회 호출") self.assertFalse(fired.fire_blocked, "성공 시 fire_blocked=False") self.assertFalse(fired.active, "ACTIVE=false 유지") # ★ marker/argv head 식별자 = 런타임 검증 canonical HEAD(full 40자). self.assertEqual(fired.head_sha, H) def test_partial_no_spawn(self): p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy(spawned=False) fired = self._fire(g, self._target_for(p), fire_fn=fn) self.assertEqual(fired.status, gate_mod.GATE_PARTIAL_FIRE_NO_SPAWN) self.assertEqual(len(calls), 1) def test_partial_no_callback_evidence(self): p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy(spawned=True, callback_evidence=False) fired = self._fire(g, self._target_for(p), fire_fn=fn) self.assertEqual(fired.status, gate_mod.GATE_PARTIAL_FIRE_NO_CALLBACK_EVIDENCE) self.assertEqual(len(calls), 1) def test_fire_fn_exception_is_fire_error(self): p = self._write_result() g = self._ready_gate(p) def _boom(_argv): del _argv raise RuntimeError("subprocess down") fired = self._fire(g, self._target_for(p), fire_fn=_boom) self.assertEqual(fired.status, gate_mod.GATE_FIRE_ERROR) def test_fire_fn_missing_is_fire_error_no_marker(self): # fire_fn 미주입(본 task 기본) → 실 발사 0(FIRE_ERROR)·marker 미작성(pilot 보존). p = self._write_result() g = self._ready_gate(p) fired = gate_mod.attempt_owner_fire( g, approved_one_shot=True, fire_fn=None, sealed_key_loader=_anu_loader, executor_key=EXECUTOR_KEY, head_probe=_probe_H, dedupe_ledger_path=self.ledger, target=self._target_for(p)) self.assertEqual(fired.status, gate_mod.GATE_FIRE_ERROR) self.assertFalse(os.path.exists(self.ledger), "fire_fn 미주입 → 재발사 marker 미작성(pilot fire_fn 주입 보존)") def test_refire_prevented_after_success(self): # 1회 성공 후 marker 기록 → 두 번째 시도는 DEDUPED(재발사 방지·자동 재시도 0). p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() first = self._fire(g, self._target_for(p), fire_fn=fn) self.assertEqual(first.status, gate_mod.GATE_REALFIRE_1SHOT_SUCCESS) self.assertTrue(os.path.exists(self.ledger), "성공 후 재발사 방지 marker 기록") second = self._fire(g, self._target_for(p), fire_fn=fn) self.assertEqual(second.status, gate_mod.GATE_DEDUPED) self.assertEqual(len(calls), 1, "재발사 0 — fire_fn 총 1회만 호출") # ── 10. mock argv 가 cokacdir --cron 1회 생성 형태 + current HEAD 반영 ────────── class Test07ArgvShape(_Base): def test_fire_argv_is_cokacdir_cron_once(self): p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() fired = self._fire(g, self._target_for(p), fire_fn=fn) self.assertEqual(fired.status, gate_mod.GATE_REALFIRE_1SHOT_SUCCESS) self.assertEqual(len(calls), 1) argv = calls[0] self.assertEqual(argv[0], "cokacdir") self.assertIn("--cron", argv) self.assertIn("--once", argv) # cron 1회 생성 형태 self.assertIn("--key", argv) self.assertIn("--at", argv) self.assertEqual(argv.count("--cron"), 1) self.assertEqual(argv.count("--once"), 1) # 실 key 가 --key 자리에 마지막 순간 주입됨(fire_fn 전용 ephemeral argv). self.assertEqual(argv[argv.index("--key") + 1], ANU_KEY) # ★ task-2742-r4: real_argv 에 --chat 포함(대상 chat 결선). self.assertIn("--chat", argv) self.assertEqual(argv.count("--chat"), 1) self.assertEqual(argv[argv.index("--chat") + 1], gate_mod.ONESHOT_PILOT_CHAT_ID) # ★ --owner-pickup-result 자리는 검증된 absolute path(다른 CWD 견고). self.assertIn("--owner-pickup-result", argv) res_path = argv[argv.index("--owner-pickup-result") + 1] self.assertTrue(os.path.isabs(res_path), "result path 는 absolute") self.assertEqual(res_path, os.path.abspath(p)) # ★ argv prompt 에 런타임 canonical HEAD(현재 HEAD)가 반영됨(상수 아님). joined = " ".join(argv) self.assertIn(H[:12], joined, "argv 가 current canonical HEAD 식별자 반영") # ── task-2742-r4: --chat 결선 (real_argv·redacted_argv 둘 다 --chat + fail-closed) ─ class Test07rChatWiring(_Base): def test_real_argv_has_chat_with_value(self): # mock fire_fn 에 넘어가는 real_argv 에 --chat 포함(대상 chat). p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() fired = self._fire(g, self._target_for(p), fire_fn=fn) self.assertEqual(fired.status, gate_mod.GATE_REALFIRE_1SHOT_SUCCESS) self.assertEqual(len(calls), 1) argv = calls[0] self.assertIn("--chat", argv) self.assertEqual(argv[argv.index("--chat") + 1], gate_mod.ONESHOT_PILOT_CHAT_ID) # --cron 발사 형태 + 대상 chat 동시 충족(PARTIAL_FIRE_NO_SPAWN 근본 fix). self.assertIn("--cron", argv) def test_redacted_persist_argv_has_chat_and_abs_path(self): # persist/return 되는 redacted argv 에도 --chat + absolute result path 포함. p = self._write_result() g = self._ready_gate(p) _, fn = _fire_spy() fired = self._fire(g, self._target_for(p), fire_fn=fn) self.assertEqual(fired.status, gate_mod.GATE_REALFIRE_1SHOT_SUCCESS) rargv = fired.argv self.assertIn("--chat", rargv) self.assertEqual(rargv[rargv.index("--chat") + 1], gate_mod.ONESHOT_PILOT_CHAT_ID) self.assertIn("--owner-pickup-result", rargv) res_path = rargv[rargv.index("--owner-pickup-result") + 1] self.assertTrue(os.path.isabs(res_path), "redacted argv result path 도 absolute") self.assertEqual(res_path, os.path.abspath(p)) # redacted argv 의 key 는 placeholder(raw key 0). self.assertEqual(rargv[rargv.index("--key") + 1], gate_mod.OWNER_PICKUP_KEY_PLACEHOLDER) self.assertNotIn(ANU_KEY, " ".join(rargv)) def test_chat_none_fail_closed(self): # chat_id=None → fail-closed, fire 0(cokacdir --cron 미등록 spawn 0 근본 차단). p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() fired = self._fire(g, self._target_for(p), fire_fn=fn, chat_id=None) self.assertEqual(fired.status, gate_mod.GATE_FIRE_BLOCKED_NO_APPROVAL) self.assertTrue(fired.fire_blocked) self.assertFalse(fired.active) self.assertEqual(calls, [], "chat_id None → fire_fn 0(cron 0)") def test_chat_empty_and_whitespace_fail_closed(self): # chat_id="" / 공백 only → fail-closed, fire 0. for bad in ("", " ", "\t"): p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() fired = self._fire(g, self._target_for(p), fire_fn=fn, chat_id=bad) self.assertEqual(fired.status, gate_mod.GATE_FIRE_BLOCKED_NO_APPROVAL, f"chat_id={bad!r} → fail-closed") self.assertEqual(calls, [], "chat_id empty → fire_fn 0(cron 0)") def test_build_argv_empty_chat_raises(self): # argv builder 자체도 빈 chat_id 를 방어적으로 거부(빈 --chat argv 0). for bad in (None, "", " "): with self.assertRaises(ValueError): gate_mod._build_owner_pickup_argv( task_id=TARGET_TASK_ID, head_sha=H, result_json_path="/abs/x.result.json", chat_id=bad) def test_nofire_gate_argv_also_has_chat(self): # no-fire gate(READY) argv 도 --chat 포함(default pilot chat 상수). p = self._write_result() g = self._ready_gate(p) self.assertIsNotNone(g.argv) self.assertIn("--chat", g.argv) self.assertEqual(g.argv[g.argv.index("--chat") + 1], gate_mod.ONESHOT_PILOT_CHAT_ID) # ── 11. invariants: raw key 0·live fire 0·ACTIVE=false·systemd/p0b_driver 0·glob 0 ─ class Test08Invariants(_Base): def test_no_raw_key_in_source(self): runner_src = (_ROOT / "dispatch" / "anu_result_pickup_runner.py").read_text("utf-8") test_src = Path(__file__).read_text("utf-8") self.assertNotIn(ANU_KEY, runner_src, "runner 소스에 raw ANU key literal 금지") self.assertNotIn(ANU_KEY, test_src, "테스트 소스에 raw ANU key literal 금지") def test_no_raw_key_in_output_and_marker(self): p = self._write_result() g = self._ready_gate(p) _, fn = _fire_spy() fired = self._fire(g, self._target_for(p), fire_fn=fn) self.assertEqual(fired.status, gate_mod.GATE_REALFIRE_1SHOT_SUCCESS) blob = json.dumps(fired.to_json(), ensure_ascii=False) self.assertNotIn(ANU_KEY, blob, "fire 결과 json 에 raw ANU key 금지") self.assertIn(gate_mod.OWNER_PICKUP_KEY_PLACEHOLDER, fired.argv, "persist argv 는 placeholder 로 redact") marker_blob = Path(self.ledger).read_text("utf-8") self.assertNotIn(ANU_KEY, marker_blob, "marker 파일에 raw ANU key 금지") def test_no_real_cron_subprocess(self): p = self._write_result() g = self._ready_gate(p) _, fn = _fire_spy() with mock.patch("subprocess.run") as srun, mock.patch("subprocess.Popen") as spop: self._fire(g, self._target_for(p), fire_fn=fn) srun.assert_not_called() spop.assert_not_called() def test_active_false_invariant(self): p = self._write_result() g = self._ready_gate(p) _, fn = _fire_spy() fired = self._fire(g, self._target_for(p), fire_fn=fn) self.assertFalse(fired.active) def test_no_systemd_or_p0b_driver_mutation_in_source(self): runner_src = (_ROOT / "dispatch" / "anu_result_pickup_runner.py").read_text("utf-8") for forbidden in ("systemctl", "WantedBy", "ExecStart", "daemon-reload"): self.assertNotIn(forbidden, runner_src, f"runner 가 {forbidden} 를 다루면 안 됨(systemd/p0b_driver 무관)") def test_self_collector_blocked(self): p = self._write_result() g = self._ready_gate(p) calls, fn = _fire_spy() fired = gate_mod.attempt_owner_fire( g, approved_one_shot=True, fire_fn=fn, sealed_key_loader=_anu_loader, executor_key=ANU_KEY, head_probe=_probe_H, dedupe_ledger_path=self.ledger, target=self._target_for(p)) self.assertEqual(fired.status, gate_mod.GATE_FAIL_CLOSED_NON_ANU_KEY) self.assertEqual(calls, [], "self-collector → fire_fn 0") def test_no_glob_full_scan_in_source(self): runner_src = (_ROOT / "dispatch" / "anu_result_pickup_runner.py").read_text("utf-8") self.assertNotIn("import glob", runner_src) self.assertNotIn("glob.glob", runner_src) # ── 5. canonical HEAD probe 실패(빈/40자 아님) → fail-closed, fire 0 ─────────── class Test05dynProbeFailFailClosed(_Base): def test_probe_empty_fail_closed(self): p = self._write_result(head_sha=H) g = self._ready_gate(p, probe=_probe_H) calls, fn = _fire_spy() fired = self._fire(g, self._target_for(p), fire_fn=fn, probe=lambda: "") self.assertEqual(fired.status, gate_mod.GATE_RESULT_NOT_PICKUP_ELIGIBLE) self.assertTrue(fired.fire_blocked) self.assertEqual(calls, [], "canonical HEAD probe 빈값 → fire_fn 0(fail-closed)") def test_probe_raises_fail_closed(self): p = self._write_result(head_sha=H) g = self._ready_gate(p, probe=_probe_H) def _boom_probe(): raise RuntimeError("git rev-parse 실패") calls, fn = _fire_spy() fired = self._fire(g, self._target_for(p), fire_fn=fn, probe=_boom_probe) self.assertEqual(fired.status, gate_mod.GATE_RESULT_NOT_PICKUP_ELIGIBLE) self.assertEqual(calls, [], "canonical HEAD probe 예외 → fire_fn 0(fail-closed)") def test_probe_non_40_len_fail_closed(self): # probe 가 full 40자가 아닌 값(short/39/41) 반환 → fail-closed. p = self._write_result(head_sha=H) g = self._ready_gate(p, probe=_probe_H) for bad in (SHORT, H_39, H_41): calls, fn = _fire_spy() fired = self._fire(g, self._target_for(p), fire_fn=fn, probe=lambda b=bad: b) self.assertEqual(fired.status, gate_mod.GATE_RESULT_NOT_PICKUP_ELIGIBLE, f"probe len!=40({len(bad)}) → fail-closed") self.assertEqual(calls, [], "probe full 40자 아님 → fire_fn 0") # ── 6. short SHA / prefix만 일치 → fire 0 (prefix 완화 0) ────────────────────── class Test06dynShortShaPrefix(_Base): def test_result_head_short_prefix_of_canonical_not_eligible(self): # result.head_sha = canonical HEAD 의 8자 prefix. gate 는 probe 도 SHORT 면 READY # (== 동일성)이 되지만, 발사 직전 canonical 은 full 40(H) → result(SHORT) != H # → 정확 일치 실패 → fire 0. (prefix/startswith 완화 0) p = self._write_result(head_sha=SHORT) g = self._ready_gate(p, probe=lambda: SHORT) # gate 시점엔 SHORT==SHORT calls, fn = _fire_spy() fired = self._fire(g, self._target_for(p), fire_fn=fn, probe=_probe_H) # 발사 직전 full 40 self.assertEqual(fired.status, gate_mod.GATE_RESULT_NOT_PICKUP_ELIGIBLE) self.assertEqual(calls, [], "short prefix → fire_fn 0(prefix 완화 0)") def test_canonical_full_but_result_prefix_only_no_fire(self): # 직접 scope (c) 검증: gate_result.head_sha = SHORT(8자), 발사 직전 canonical = full H. # len(g_head)!=40 → RESULT_NOT_PICKUP_ELIGIBLE. fire 0. p = self._write_result(head_sha=SHORT) g_short = self._ready_gate(p, probe=lambda: SHORT) self.assertEqual(g_short.head_sha, SHORT) calls, fn = _fire_spy() fired = self._fire(g_short, self._target_for(p), fire_fn=fn, probe=_probe_H) self.assertEqual(fired.status, gate_mod.GATE_RESULT_NOT_PICKUP_ELIGIBLE) self.assertEqual(calls, [], "result.head_sha prefix only → fire 0") # ── 1·10. dynamic: current_head=H 와 fixture H 정확 일치 fixture만 대상 ───────── class Test01dynDynamicHeadMatch(_Base): def test_only_fixture_matching_current_head_fires(self): # 동일 target/task_id 이나 head 가 current canonical(H)과 다른 fixture(H2)는 fire 0. # 현재 HEAD(H)와 정확 일치하는 fixture(H)만 발사된다. # (i) H2 fixture (gate 시점 canonical=H2 → READY) 이나 발사 직전 canonical=H → 불일치. p2 = self._write_result(head_sha=H2, fname="r_h2.result.json") g2 = self._ready_gate(p2, probe=lambda: H2) c2, fn2 = _fire_spy() fired2 = self._fire(g2, self._target_for(p2), fire_fn=fn2, probe=_probe_H) self.assertEqual(fired2.status, gate_mod.GATE_RESULT_NOT_PICKUP_ELIGIBLE) self.assertEqual(c2, [], "current HEAD(H) 와 다른 fixture(H2) → fire 0") # (ii) H fixture, 발사 직전 canonical=H → 정확 일치 → fire 1회. p1 = self._write_result(head_sha=H, fname="r_h1.result.json") g1 = self._ready_gate(p1, probe=_probe_H) c1, fn1 = _fire_spy() fired1 = self._fire(g1, self._target_for(p1), fire_fn=fn1, probe=_probe_H) self.assertEqual(fired1.status, gate_mod.GATE_REALFIRE_1SHOT_SUCCESS) self.assertEqual(len(c1), 1, "current HEAD 일치 fixture만 정확히 1회 fire") # ── argv result path 견고화(r2 유지) + None/empty/relative fail-closed ───────── class Test10ArgvAbsPathRobustness(_Base): def _result_arg(self, argv): if "--owner-pickup-result" in argv: return argv[argv.index("--owner-pickup-result") + 1] return argv[-1] def test_fire_argv_result_path_is_absolute_t_path_abs(self): p_abs = self._write_result() rel = os.path.relpath(p_abs, os.getcwd()) self.assertFalse(os.path.isabs(rel), "전제: gate 에 relative path 주입") g = self._ready_gate(rel) self.assertEqual(g.result_json_path, rel) calls, fn = _fire_spy() fired = self._fire(g, self._target_for(p_abs), fire_fn=fn) # target = absolute self.assertEqual(fired.status, gate_mod.GATE_REALFIRE_1SHOT_SUCCESS) self.assertEqual(len(calls), 1, "fire_fn 정확히 1회") real_path = self._result_arg(calls[0]) self.assertTrue(os.path.isabs(real_path), "argv result path 는 absolute 여야 함") self.assertEqual(real_path, os.path.abspath(p_abs), "argv result path = 검증된 t_path_abs") self.assertNotEqual(real_path, rel, "re_gate.result_json_path(relative) 가 아니라 t_path_abs 사용") def test_redacted_persist_argv_same_absolute_and_placeholder(self): p_abs = self._write_result() rel = os.path.relpath(p_abs, os.getcwd()) g = self._ready_gate(rel) calls, fn = _fire_spy() fired = self._fire(g, self._target_for(p_abs), fire_fn=fn) self.assertEqual(fired.status, gate_mod.GATE_REALFIRE_1SHOT_SUCCESS) real_path = self._result_arg(calls[0]) redacted_path = self._result_arg(fired.argv) self.assertEqual(redacted_path, real_path, "real/redacted argv result path 동일") self.assertEqual(redacted_path, os.path.abspath(p_abs)) self.assertTrue(os.path.isabs(redacted_path)) self.assertEqual(calls[0][calls[0].index("--key") + 1], ANU_KEY) self.assertEqual(fired.argv[fired.argv.index("--key") + 1], gate_mod.OWNER_PICKUP_KEY_PLACEHOLDER) self.assertNotIn(ANU_KEY, json.dumps(fired.to_json(), ensure_ascii=False)) def test_target_result_path_none_fail_closed(self): p_abs = self._write_result() g = self._ready_gate(p_abs) calls, fn = _fire_spy() with mock.patch("subprocess.run") as srun, mock.patch("subprocess.Popen") as spop: fired = self._fire(g, self._target_for(None), fire_fn=fn) srun.assert_not_called() spop.assert_not_called() self.assertEqual(fired.status, gate_mod.GATE_FIRE_BLOCKED_NO_APPROVAL) self.assertEqual(calls, [], "target path None → fire_fn 0(fire 0)") def test_target_result_path_empty_and_whitespace_fail_closed(self): p_abs = self._write_result() g = self._ready_gate(p_abs) for empty in ("", " ", "\t"): calls, fn = _fire_spy() fired = self._fire(g, self._target_for(empty), fire_fn=fn) self.assertEqual(fired.status, gate_mod.GATE_FIRE_BLOCKED_NO_APPROVAL) self.assertEqual(calls, [], "target path 빈/공백 → fire_fn 0(fire 0)") def test_gate_result_path_empty_fail_closed(self): p_abs = self._write_result() g = self._ready_gate(p_abs) g.result_json_path = "" # 빈 gate path 강제 calls, fn = _fire_spy() fired = self._fire(g, self._target_for(p_abs), fire_fn=fn) self.assertEqual(fired.status, gate_mod.GATE_FIRE_BLOCKED_NO_APPROVAL) self.assertEqual(calls, [], "gate path 빈 → fire_fn 0(fire 0)") # ── dynamic 설계 입증: 하드코드 제거 + full 40 exact + prefix 완화 0 ──────────── class Test11dynDynamicTargetDesign(_Base): def test_oneshot_target_head_sha_constant_removed(self): # ★ ONESHOT_TARGET_HEAD_SHA 하드코드 상수 제거 입증. self.assertFalse(hasattr(gate_mod, "ONESHOT_TARGET_HEAD_SHA"), "ONESHOT_TARGET_HEAD_SHA 상수는 제거되어야 함(런타임 동적 판정)") self.assertNotIn("ONESHOT_TARGET_HEAD_SHA", gate_mod.__all__) def test_oneshot_target_has_no_head_sha_field(self): # ★ OneShotTarget 에 head_sha 필드 부재(task_id + result_json_path 만). import dataclasses names = {f.name for f in dataclasses.fields(gate_mod.OneShotTarget)} self.assertEqual(names, {"task_id", "result_json_path"}, "OneShotTarget 는 task_id + result_json_path 만(head_sha 필드 0)") self.assertFalse(hasattr(gate_mod.ONESHOT_TARGET, "head_sha")) def test_no_literal_old_hardcoded_head_value_in_source(self): # ★ 이전(r2) 하드코드 40자 SHA 값이 코드 어디에도 literal 로 남지 않음(주석 포함). # 값은 토큰을 합성해 구성한다 — 본 테스트 소스에도 full literal 을 박지 않는다. old_head = "4012bc505ccdc952ca48d7d70347da" + "91497aad7f" # 합성(literal 0) runner_src = (_ROOT / "dispatch" / "anu_result_pickup_runner.py").read_text("utf-8") self.assertNotIn(old_head, runner_src, "이전 하드코드 head_sha 값이 source 에 남아선 안 됨(완전 제거)") # 상수 정의(assignment) 라인이 없어야 함 — 주석 설명은 허용. for line in runner_src.splitlines(): stripped = line.strip() if stripped.startswith("#"): continue self.assertNotIn("ONESHOT_TARGET_HEAD_SHA =", stripped, "ONESHOT_TARGET_HEAD_SHA 상수 정의(비주석) 부재") class Test12dynFull40ExactOnly(_Base): def test_only_full_40_exact_fires(self): # full 40자 정확 일치 → fire. (성공 기준선) p = self._write_result(head_sha=H) g = self._ready_gate(p, probe=_probe_H) calls, fn = _fire_spy() fired = self._fire(g, self._target_for(p), fire_fn=fn, probe=_probe_H) self.assertEqual(fired.status, gate_mod.GATE_REALFIRE_1SHOT_SUCCESS) self.assertEqual(len(calls), 1) def test_case_variant_head_no_fire(self): # 대소문자만 다른 head(동일 hex 의미라도 문자열 != ) → 정확 일치 실패 → fire 0. p = self._write_result(head_sha=H.upper()) # gate 시점엔 probe 도 동일 대문자라 READY 가능. g = self._ready_gate(p, probe=lambda: H.upper()) calls, fn = _fire_spy() # 발사 직전 canonical 은 소문자 full 40(H) → H.upper() != H → fire 0. fired = self._fire(g, self._target_for(p), fire_fn=fn, probe=_probe_H) self.assertEqual(fired.status, gate_mod.GATE_RESULT_NOT_PICKUP_ELIGIBLE) self.assertEqual(calls, [], "대소문자 변형(문자열 불일치) → fire 0(== only)") # ── terminal status enum 전체 도달 ───────────────────────────────────────────── class Test09TerminalEnum(_Base): def test_all_nine_terminals_reachable(self): seen = set() p = self._write_result() def _led(name): return os.path.join(self.tmp, f"led_{name}.jsonl") # OWNER_PICKUP_READY_NO_FIRE (gate) seen.add(self._gate(p, executor_key=EXECUTOR_KEY).status) g = self._ready_gate(p) # FIRE_BLOCKED_NO_APPROVAL (approved=false; marker 미작성) _, fn = _fire_spy() seen.add(gate_mod.attempt_owner_fire(g, fire_fn=fn, target=self._target_for(p)).status) # RESULT_NOT_PICKUP_ELIGIBLE (head moved at fire-time; marker 미작성) _, fn = _fire_spy() seen.add(self._fire(g, self._target_for(p), fire_fn=fn, probe=lambda: H2, dedupe_ledger_path=_led("stale")).status) # FAIL_CLOSED_NON_ANU_KEY (marker 미작성) _, fn = _fire_spy() seen.add(self._fire(g, self._target_for(p), fire_fn=fn, loader=lambda: DEV_SELF_KEY, dedupe_ledger_path=_led("key")).status) # FIRE_ERROR (fire_fn exception; marker 기록 → 독립 ledger) def _boom(_a): del _a raise RuntimeError("x") seen.add(self._fire(g, self._target_for(p), fire_fn=_boom, dedupe_ledger_path=_led("err")).status) # PARTIAL_FIRE_NO_SPAWN (marker 기록 → 독립 ledger) _, fn = _fire_spy(spawned=False) seen.add(self._fire(g, self._target_for(p), fire_fn=fn, dedupe_ledger_path=_led("nospawn")).status) # PARTIAL_FIRE_NO_CALLBACK_EVIDENCE (marker 기록 → 독립 ledger) _, fn = _fire_spy(spawned=True, callback_evidence=False) seen.add(self._fire(g, self._target_for(p), fire_fn=fn, dedupe_ledger_path=_led("nocb")).status) # REALFIRE_1SHOT_SUCCESS (독립 ledger — 이후 DEDUPED 재사용) led_ok = _led("ok") _, fn = _fire_spy() seen.add(self._fire(g, self._target_for(p), fire_fn=fn, dedupe_ledger_path=led_ok).status) # DEDUPED (위 success 의 marker 재사용 → 재발사 방지) _, fn = _fire_spy() seen.add(self._fire(g, self._target_for(p), fire_fn=fn, dedupe_ledger_path=led_ok).status) self.assertEqual(seen, set(gate_mod.GATE_ALL_TERMINAL_STATUSES), "회장 9 terminal status 전체 도달") if __name__ == "__main__": unittest.main(verbosity=2)