# -*- coding: utf-8 -*- """task-2733 — P0B_MINIMUM event strategy 보정(전용 inbox) 회귀. 회장 재정의 범위(P0B_MINIMUM) 내 event strategy 결함 수정 검증: ① systemd anu-pickup.path 가 전용 inbox(memory/events/p0b_inbox/task-*.result.json)만 watch 하고 memory/events 직속 legacy 를 watch 하지 않는다(잔존 폭주 0, 설계 검증). ② driver scan 기본 glob 이 inbox 만 대상으로 한다(INBOX_DIR_REL). ③ memory/events 직속 legacy result.json 은 scan 후에도 byte-identical 무변경 (이동/삭제 0 — ANCHOR-2). ④ 처리(CLOSEOUT_DONE) 후 result.json 이 inbox 밖(processed)으로 이동 → inbox 의 *.result.json 0건(재트리거 0 — 설계 검증). ⑤ reporting path(task-2732) 재사용: driver record → PASS/HOLD/CALLBACK_MISSING. ⑥ launcher_fn=None 기본(wake/real launch 0) 회귀 · ANU key literal 0. 네트워크 0, 전부 isolated tmp root + fake sender/mock. canary(실제 inbox 작성)·systemd 설치·merge 는 회장 승인 전 금지 — 본 테스트는 설계/dry-run 만 검증한다. """ from __future__ import annotations import glob as _glob import json import os import re import subprocess import sys import tempfile import types import unittest from datetime import datetime, timezone from pathlib import Path _REPO_ROOT = Path(__file__).resolve().parents[1] if str(_REPO_ROOT) not in sys.path: sys.path.insert(0, str(_REPO_ROOT)) # tests/dispatch/__init__.py(테스트용 빈 패키지) 가 prepend 되어 실제 dispatch 패키지를 # 가릴 수 있으므로, 실제 dispatch 패키지를 파일 위치로 직접 로드해 sys.modules 에 고정한 뒤 # 서브모듈을 import 한다(test_anu_pickup_driver_2721.py 와 동일 패턴). import importlib.util as _ilu # noqa: E402 _real_init = _REPO_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(_REPO_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 import anu_pickup_reporter as rpt # noqa: E402 _PATH_FILE = _REPO_ROOT / "deploy" / "systemd" / "anu-pickup.path" _SERVICE_FILE = _REPO_ROOT / "deploy" / "systemd" / "anu-pickup.service" _RUNBOOK_FILE = _REPO_ROOT / "docs" / "p0b_driver_runbook_260601.md" _CLOCK = lambda: datetime(2026, 6, 11, 6, 0, 0, tzinfo=timezone.utc) # noqa: E731 _NO_SLEEP = lambda *a, **k: None # noqa: E731 VALID_PAYLOAD = { "task_id": "task-999", "completion_signal": "EXECUTOR_RESULT_WRITTEN", "collector_envelope": {"task_id": "task-999", "schedule_id": "sch-1"}, "report_path": "r.md", "sha256": "deadbeefcafe0000", } # ── helper ──────────────────────────────────────────────────────────────────── def _inbox(root: str) -> str: return os.path.join(root, "memory", "events", "p0b_inbox") def _events(root: str) -> str: return os.path.join(root, "memory", "events") def _make_dirs(root: str) -> None: os.makedirs(_inbox(root), exist_ok=True) os.makedirs(os.path.join(root, "memory", "state"), exist_ok=True) def _enable(root: str) -> None: flag = os.path.join(root, "memory", "state", "p0b_driver_enabled") with open(flag, "w", encoding="utf-8") as fh: fh.write("enabled\n") def _write(dir_path: str, name: str, payload=None) -> str: os.makedirs(dir_path, exist_ok=True) p = os.path.join(dir_path, name) with open(p, "w", encoding="utf-8") as fh: json.dump(VALID_PAYLOAD if payload is None else payload, fh) return p def _age(path: str, seconds: float = 30.0) -> None: # mtime 을 driver 가 readiness 판정에 쓰는 _CLOCK 기준 과거로 설정한다. # driver 는 age = clock().timestamp() - mtime 으로 안정성을 본다(anu_pickup_driver._check_readiness). # real time.time() 을 쓰면 실제 시각이 _CLOCK(고정 06:00 UTC=15:00 KST)을 지나는 순간 # mtime(now) > clock → age 음수 → recent_mtime DEFER 로 4 fail 이 발생하는 시간 의존 # time-bomb 이 된다. _CLOCK 기준으로 통일하여 시각 무관 결정성을 확보한다. past = _CLOCK().timestamp() - seconds os.utime(path, (past, past)) def _pickup_mock(verdict="CLOSEOUT_DONE"): calls = [] def _p(*a, **k): calls.append((a, k)) return types.SimpleNamespace( verdict=verdict, ok=True, argv=["x"], task_id="task-999", reasons=[], ) _p.calls = calls # type: ignore[attr-defined] return _p def _verify_mock(verdict="AUTHORITATIVE"): def _v(*a, **k): return types.SimpleNamespace(verdict=verdict, ok=True, classification="", reasons=[]) return _v class _FakeSender: def __init__(self, ok=True): self.ok = ok self.calls = [] def __call__(self, file_path): try: with open(file_path, "r", encoding="utf-8") as fh: contents = fh.read() except OSError: contents = "" self.calls.append((file_path, contents)) return self.ok # ═══════════════════════════════════════════════════════════════════════════════ # ① systemd .path 가 전용 inbox 만 watch (설계 검증 — 잔존 폭주 0) # ═══════════════════════════════════════════════════════════════════════════════ class TestSystemdPathInboxOnly(unittest.TestCase): def test_path_watches_inbox_glob_only(self): content = _PATH_FILE.read_text(encoding="utf-8") # 활성 directive(라인 시작) 만 추출 — 주석(#) 제외. directives = [ ln.strip() for ln in content.splitlines() if ln.strip().startswith("PathExistsGlob=") ] self.assertEqual( directives, ["PathExistsGlob=%h/workspace/memory/events/p0b_inbox/task-*.result.json"], "anu-pickup.path 의 활성 PathExistsGlob 은 전용 inbox 하나뿐이어야 한다", ) # memory/events 직속 legacy glob 를 활성 directive 로 감시하면 안 됨(잔존 폭주 재발). self.assertNotIn( "PathExistsGlob=%h/workspace/memory/events/task-*.result.json", content, ) # ═══════════════════════════════════════════════════════════════════════════════ # ② driver scan 기본 glob = inbox 전용 # ═══════════════════════════════════════════════════════════════════════════════ class TestScanGlobsInboxOnly(unittest.TestCase): def test_inbox_dir_constant(self): self.assertEqual(drv.INBOX_DIR_REL, "memory/events/p0b_inbox") # EVENTS_DIR_REL 은 무변경(legacy 위치 참조 보존). self.assertEqual(drv.EVENTS_DIR_REL, "memory/events") # inbox 는 events 하위지만 quarantine/processed/evidence 는 events 밖(watched 밖). self.assertTrue(drv.INBOX_DIR_REL.startswith("memory/events/")) self.assertFalse(drv.QUARANTINE_DIR_REL.startswith("memory/events")) self.assertFalse(drv.PROCESSED_DIR_REL.startswith("memory/events")) def test_scan_default_targets_inbox_not_legacy(self): with tempfile.TemporaryDirectory() as td: _make_dirs(td) _enable(td) # 신규 result → inbox / legacy result → memory/events 직속 inbox_p = _write(_inbox(td), "task-999.result.json") _age(inbox_p) legacy_p = _write(_events(td), "task-legacy-1.result.json") _age(legacy_p) records = drv.scan_once( td, pickup_fn=_pickup_mock("CLOSEOUT_DONE"), verify_fn=_verify_mock(), write_evidence=False, clock=_CLOCK, sleep_fn=_NO_SLEEP, ) scanned = [r for r in records if r.result_path] # inbox 파일만 scan 대상 — legacy 는 record 에 등장 0. self.assertEqual(len(scanned), 1) self.assertTrue(scanned[0].result_path.endswith( os.path.join("p0b_inbox", "task-999.result.json"))) self.assertNotIn( "task-legacy-1", " ".join(r.result_path for r in scanned), ) # ═══════════════════════════════════════════════════════════════════════════════ # ③ memory/events 직속 legacy 무변경 (이동/삭제 0) # ═══════════════════════════════════════════════════════════════════════════════ class TestLegacyUntouched(unittest.TestCase): def test_legacy_events_byte_identical_after_scan(self): with tempfile.TemporaryDirectory() as td: _make_dirs(td) _enable(td) # 12개 legacy result.json 을 memory/events 직속에 작성(inbox 아님). legacy = {} for i in range(12): payload = dict(VALID_PAYLOAD) payload["task_id"] = f"task-legacy-{i}" p = _write(_events(td), f"task-legacy-{i}.result.json", payload) _age(p) legacy[p] = Path(p).read_bytes() before = sorted(os.listdir(_events(td))) drv.scan_once( td, pickup_fn=_pickup_mock("CLOSEOUT_DONE"), verify_fn=_verify_mock(), write_evidence=False, clock=_CLOCK, sleep_fn=_NO_SLEEP, ) # 파일 목록·내용 모두 무변경(이동/삭제 0). self.assertEqual(sorted(os.listdir(_events(td))), before) for p, content in legacy.items(): self.assertTrue(os.path.exists(p), f"legacy 삭제됨: {p}") self.assertEqual(Path(p).read_bytes(), content, f"legacy 변형됨: {p}") # quarantine/processed 미생성(legacy 미처리). self.assertFalse(os.path.exists(os.path.join(td, "memory", "p0b_state", "processed"))) self.assertFalse(os.path.exists(os.path.join(td, "memory", "p0b_state", "quarantine"))) # ═══════════════════════════════════════════════════════════════════════════════ # ④ 처리 후 inbox 비움 → 재트리거 0 (설계 검증) # ═══════════════════════════════════════════════════════════════════════════════ class TestInboxEmptiedNoRetrigger(unittest.TestCase): def test_inbox_no_result_json_after_processing(self): with tempfile.TemporaryDirectory() as td: _make_dirs(td) _enable(td) p = _write(_inbox(td), "task-999.result.json") _age(p) records = drv.scan_once( td, pickup_fn=_pickup_mock("CLOSEOUT_DONE"), verify_fn=_verify_mock(), write_evidence=False, clock=_CLOCK, sleep_fn=_NO_SLEEP, ) self.assertEqual(records[0].verdict, drv.VERDICT_CLOSEOUT_DONE) # inbox 의 *.result.json 0건 → systemd PathExistsGlob 재매칭 0(재트리거 0). remaining = _glob.glob(os.path.join(_inbox(td), "*.result.json")) self.assertEqual(remaining, [], f"inbox 에 result.json 잔존: {remaining}") # 원본은 inbox 밖 processed 로 이동(수거 완료). processed = _glob.glob(os.path.join( td, "memory", "p0b_state", "processed", "task-999.result.json*")) self.assertEqual(len(processed), 1) def test_second_scan_idempotent_zero_records(self): # inbox 비운 뒤 재 scan → 처리 대상 0(무한 재트리거 소멸). with tempfile.TemporaryDirectory() as td: _make_dirs(td) _enable(td) p = _write(_inbox(td), "task-999.result.json") _age(p) drv.scan_once(td, pickup_fn=_pickup_mock("CLOSEOUT_DONE"), verify_fn=_verify_mock(), write_evidence=False, clock=_CLOCK, sleep_fn=_NO_SLEEP) records2 = drv.scan_once(td, pickup_fn=_pickup_mock("CLOSEOUT_DONE"), verify_fn=_verify_mock(), write_evidence=False, clock=_CLOCK, sleep_fn=_NO_SLEEP) self.assertEqual([r for r in records2 if r.result_path], []) # ═══════════════════════════════════════════════════════════════════════════════ # ⑤ reporting path(task-2732) 재사용 → PASS/HOLD/CALLBACK_MISSING # ═══════════════════════════════════════════════════════════════════════════════ class TestReportingPathReused(unittest.TestCase): def test_closeout_record_maps_to_pass(self): with tempfile.TemporaryDirectory() as td: _make_dirs(td) _enable(td) p = _write(_inbox(td), "task-999.result.json") _age(p) records = drv.scan_once( td, pickup_fn=_pickup_mock("CLOSEOUT_DONE"), verify_fn=_verify_mock(), write_evidence=False, clock=_CLOCK, sleep_fn=_NO_SLEEP) sender = _FakeSender(ok=True) decisions = rpt.report_records(records, root=td, sender=sender, clock=_CLOCK) sent = [d for d in decisions if d.decision == rpt.REPORT_PASS] self.assertEqual(len(sent), 1) self.assertTrue(sent[0].sent) self.assertIn("task-999: PASS", sender.calls[0][1]) def test_quarantine_record_maps_to_hold(self): with tempfile.TemporaryDirectory() as td: sender = _FakeSender(ok=True) dec = rpt.report_one( drv.VERDICT_QUARANTINE, os.path.join(_inbox(td), "task-77.result.json"), root=td, sender=sender, result_data={}, clock=_CLOCK) self.assertEqual(dec.decision, rpt.REPORT_HOLD) self.assertIn("task-77: HOLD", sender.calls[0][1]) def test_callback_missing_mapping(self): with tempfile.TemporaryDirectory() as td: sender = _FakeSender(ok=True) dec = rpt.report_one( drv.VERDICT_CLOSEOUT_DONE, os.path.join(_inbox(td), "task-88.result.json"), root=td, sender=sender, result_data={"callback_schedule_created": True}, clock=_CLOCK) self.assertEqual(dec.decision, rpt.REPORT_CALLBACK_MISSING) self.assertIn("task-88: CALLBACK_MISSING", sender.calls[0][1]) # ═══════════════════════════════════════════════════════════════════════════════ # ⑥ launcher_fn=None 기본(wake/real launch 0) 회귀 # ═══════════════════════════════════════════════════════════════════════════════ class TestWakeZeroDefault(unittest.TestCase): def test_scan_default_no_real_launch(self): with tempfile.TemporaryDirectory() as td: _make_dirs(td) _enable(td) p = _write(_inbox(td), "task-999.result.json") _age(p) # pickup_fn 이 WAKE_BUILT 를 반환해도 launcher_fn/relay_fn 미주입(기본 None) # → 실제 launch 0(fire_cron_id None). records = drv.scan_once( td, pickup_fn=_pickup_mock("WAKE_BUILT"), verify_fn=_verify_mock(), write_evidence=False, clock=_CLOCK, sleep_fn=_NO_SLEEP) wake = [r for r in records if r.verdict == drv.VERDICT_WAKE_BUILT] self.assertEqual(len(wake), 1) self.assertIsNone(wake[0].fire_cron_id, "launcher_fn=None 인데 real launch(fire_cron_id) 발생") # ═══════════════════════════════════════════════════════════════════════════════ # ⑥ ANU key literal 0 (분할 조합으로만 — 완성 literal 미노출) # ═══════════════════════════════════════════════════════════════════════════════ class TestNoKeyLiteral(unittest.TestCase): def test_no_anu_key_literal_in_sources(self): forbidden = "c119085" + "addb0f8b7" for rel in ("dispatch/anu_pickup_driver.py", "deploy/systemd/anu-pickup.path", "tests/test_anu_pickup_inbox_2733.py"): src = (_REPO_ROOT / rel).read_text(encoding="utf-8") self.assertNotIn(forbidden, src, f"{rel} 에 ANU key literal 노출") # ═══════════════════════════════════════════════════════════════════════════════ # ⑦ inbox 디렉토리 사전생성 보장 (task-2733-r2 — MEDIUM dir preflight) # activation 전 preflight 가 p0b_inbox 부모 디렉토리 생성을 보장한다(설치 안정성). # PathExistsGlob 부모(p0b_inbox) 부재 시 path 유닛 시작이 흔들리는 문제 해소. # ═══════════════════════════════════════════════════════════════════════════════ class TestInboxDirPreflight(unittest.TestCase): def test_runbook_mkdir_preflight_before_path_enable(self): """런북 enable 절차에 inbox mkdir preflight 가 path enable 보다 앞서 존재한다. path 유닛 시작(systemctl enable --now anu-pickup.path) 시점에 부모 디렉토리 (memory/events/p0b_inbox)가 존재하도록, 그 직전에 mkdir -p preflight 가 명시돼야 함. """ text = _RUNBOOK_FILE.read_text(encoding="utf-8") # inbox 를 대상으로 한 mkdir preflight 라인. mkdir_idx = text.find("mkdir -p") # 같은 라인이 inbox 를 가리키는지 확인. self.assertNotEqual(mkdir_idx, -1, "런북에 inbox mkdir preflight 가 없습니다") mkdir_line = next( (ln for ln in text.splitlines() if "mkdir -p" in ln and "memory/events/p0b_inbox" in ln), None, ) self.assertIsNotNone( mkdir_line, "런북 mkdir preflight 가 memory/events/p0b_inbox 를 대상으로 하지 않습니다") assert mkdir_line is not None # type narrowing # path enable 명령 위치. enable_idx = text.find("enable --now anu-pickup.path") self.assertNotEqual(enable_idx, -1, "런북에 path enable 절차가 없습니다") # mkdir(inbox 대상) 라인 위치가 path enable 보다 앞서야 한다. inbox_mkdir_idx = text.find(mkdir_line) self.assertLess( inbox_mkdir_idx, enable_idx, "런북에서 inbox mkdir preflight 가 path enable 보다 뒤에 있습니다(부모 디렉토리 보장 실패)") def test_service_execstartpre_creates_inbox(self): """service 유닛에 inbox 를 mkdir -p 하는 ExecStartPre 보조 안전망이 있다.""" content = _SERVICE_FILE.read_text(encoding="utf-8") pre = [ ln.strip() for ln in content.splitlines() if ln.strip().startswith("ExecStartPre=") ] self.assertTrue(pre, "anu-pickup.service 에 ExecStartPre 보조 안전망이 없습니다") self.assertTrue( any("mkdir -p" in ln and "memory/events/p0b_inbox" in ln for ln in pre), f"ExecStartPre 가 inbox mkdir -p 를 수행하지 않습니다: {pre}") # ExecStartPre(보조)가 ExecStart(driver 호출)보다 앞서야 systemd 가 먼저 실행한다. exec_pre_idx = content.find("ExecStartPre=") exec_start_idx = content.find("ExecStart=/bin/bash") self.assertLess(exec_pre_idx, exec_start_idx, "ExecStartPre 가 ExecStart 보다 뒤에 정의되었습니다") def test_preflight_creates_inbox_from_absent_parent(self): """부모 디렉토리 부재 상태에서 service ExecStartPre mkdir 명령이 inbox 를 생성한다(기능 검증). service 유닛에서 ExecStartPre mkdir 경로를 추출, %h 를 tmp HOME 으로 치환한 뒤 실제 실행하여 p0b_inbox 가 absent → 생성됨을 확인. 재실행 시 idempotent(에러 0)도 확인. """ content = _SERVICE_FILE.read_text(encoding="utf-8") m = re.search(r"^ExecStartPre=/bin/mkdir\s+-p\s+(\S+)\s*$", content, re.MULTILINE) self.assertIsNotNone(m, "ExecStartPre mkdir -p 형식을 찾지 못했습니다") assert m is not None # type narrowing target_tmpl = m.group(1) self.assertIn("%h/workspace/memory/events/p0b_inbox", target_tmpl) with tempfile.TemporaryDirectory() as home: # %h → tmp HOME 치환. 부모(memory/events)까지만 만들고 p0b_inbox 는 일부러 부재. os.makedirs(os.path.join(home, "workspace", "memory", "events"), exist_ok=True) target = target_tmpl.replace("%h", home) inbox = os.path.join(home, "workspace", "memory", "events", "p0b_inbox") self.assertFalse(os.path.exists(inbox), "사전조건: inbox 가 미리 존재하면 안 됨") # 실제 mkdir -p 실행 (systemd ExecStartPre 와 동일 명령). r1 = subprocess.run(["/bin/mkdir", "-p", target], capture_output=True) self.assertEqual(r1.returncode, 0, f"mkdir 실패: {r1.stderr!r}") self.assertTrue(os.path.isdir(inbox), "preflight 후 inbox 디렉토리가 생성되지 않았습니다") # idempotent: 재실행해도 에러 0(이미 존재). r2 = subprocess.run(["/bin/mkdir", "-p", target], capture_output=True) self.assertEqual(r2.returncode, 0, f"재실행 mkdir 실패(idempotent 위반): {r2.stderr!r}") self.assertTrue(os.path.isdir(inbox)) def test_runbook_mkdir_command_creates_inbox(self): """런북에 적힌 mkdir 명령(~/workspace/...)을 HOME=tmp 로 실행해 inbox 생성 보장(기능 검증).""" text = _RUNBOOK_FILE.read_text(encoding="utf-8") # fenced 코드블록 안의 inbox mkdir 명령 추출. cmd = next( (ln.strip() for ln in text.splitlines() if ln.strip().startswith("mkdir -p") and "memory/events/p0b_inbox" in ln), None, ) self.assertIsNotNone(cmd, "런북 코드블록에서 inbox mkdir 명령을 찾지 못했습니다") assert cmd is not None # type narrowing with tempfile.TemporaryDirectory() as home: inbox = os.path.join(home, "workspace", "memory", "events", "p0b_inbox") self.assertFalse(os.path.exists(inbox)) # HOME 을 tmp 로 한정하여 ~ 확장이 tmp 안에서만 일어나게 함(시스템 오염 0). env = dict(os.environ, HOME=home) r = subprocess.run(["/bin/bash", "-c", cmd], capture_output=True, env=env) self.assertEqual(r.returncode, 0, f"런북 mkdir 명령 실패: {r.stderr!r}") self.assertTrue(os.path.isdir(inbox), "런북 mkdir 명령 실행 후 inbox 가 생성되지 않았습니다") def test_path_glob_unchanged_inbox_only(self): """preflight 추가 후에도 PathExistsGlob 은 여전히 inbox 전용(legacy 직속 감시 0).""" content = _PATH_FILE.read_text(encoding="utf-8") directives = [ ln.strip() for ln in content.splitlines() if ln.strip().startswith("PathExistsGlob=") ] self.assertEqual( directives, ["PathExistsGlob=%h/workspace/memory/events/p0b_inbox/task-*.result.json"], "preflight 보정이 PathExistsGlob 을 변경했습니다(inbox 전용 불변 위반)") self.assertNotIn( "PathExistsGlob=%h/workspace/memory/events/task-*.result.json", content) if __name__ == "__main__": unittest.main()