"""tests/regression/test_p0b_event_strategy_2760.py task-2760+2 — P0B event strategy move-first 최소 replacement PR regression. 설계근거: memory/plans/pr242_hold_ledger_axis_split_required_packet_260617.md (회장 승인). 상태: P0B_EVENT_STRATEGY_MOVE_FIRST_MINIMAL_REPLACEMENT_PR_CREATED_ACTIVE_FALSE · ACTIVE=false. ★ 범위 축소 replacement: PR #242 의 ledger compaction/robustness 축(whack-a-mole)을 추가 bounded fix 가 아니라 **들어내어**(제거) 범위를 좁힌다. 본 PR 은 "ledger 완벽 수정"이 아니라 "move-first 불변식만 안전 보존"한다. ledger compaction/self-healing/robustness 는 event strategy PR 범위에서 분리(DESIGN_BACKLOG). 검증 대상은 dispatch.anu_pickup_driver.scan_live_inbox_once 의 "신규 이벤트(live inbox)" 감시 전략이다. PathExistsGlob 가 retained legacy result.json 의 "파일 존재"만으로 반복 발화(retrigger)하던 문제를, 전용 LIVE_INBOX 만 scan + (task_id, source_result_sha256) processed ledger idempotency + move-first 로 구조적으로 제거한다. ★ 본 테스트는 fixture(tmp dir) regression 전용이다 — real callback/wake/systemd 실행 0· production activation 0·ACTIVE=true 0·canary 실행 0. callback 은 in-memory spy 로만 관측. 기타(task-2760 기본 불변식, 각각 1 테스트): 1. retained legacy result 가 scan 안 됨 2. live inbox 신규 result 1개만 감지 3. processed ledger 중복 방지 4. 동일 (task_id, source_result_sha256) 재처리 0 5. busy 시 skip 아닌 delayed/pending 잔존 6. self-collector 차단(ANU key 만 fire, executor self-key 차단) 7. ACTIVE=false 유지 8. 기존 task-2758 non-matching thin gate 상태 불변(import/동작) 9. matching_axis DESIGN_REVIEW_REQUIRED 유지(matching 재도입 0) move-first 핵심 불변식(HIGH 1, task-2760+2): 10. move 성공 시 완료 ledger 기록(완료 entry 존재 ⟺ move 성공) 11. move 실패 시 완료 ledger 미기록 12. move 실패 시 VERDICT_LIVE_PENDING_MOVE 13. move 실패 시 원본 inbox 잔존(retry pending) 14. move 실패가 stuck-complete dedupe-skip 되지 않음 + 복구 시 재시도 완료 compaction/robustness 제거 검증(범위 축소 replacement): 15. LIVE_LEDGER_MAX_ENTRIES 상수 없음 16. _rewrite_ledger 함수 없음 17. _load_processed_ledger compaction trim 없음(헬퍼 자체 부재) 18. ledger robustness 신규 방어 로직 없음(단순 append-only 읽기/쓰기만) """ import json import os import sys import tempfile import unittest from pathlib import Path # regression/conftest.py 가 worktree root 를 sys.path[0] 에 보장하지만, 단독 실행/순서 # 변동 대비로 보강한다. 또한 tests/dispatch/__init__.py(테스트용 빈 패키지)가 실제 # dispatch 패키지를 가리지 않도록, 워크트리 root 의 실제 dispatch 를 파일 위치로 직접 # 로드해 sys.modules 에 고정한다(test_anu_pickup_driver_2721.py 와 동일 패턴). _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 # ── helpers ───────────────────────────────────────────────────────────────────── def _enabled(): """activation flag_reader: ACTIVE 활성 모사(ACTIVE=true 실파일 미생성 — 주입만).""" return drv.ACTIVATION_ENABLED def _disabled(): """activation flag_reader: ACTIVE=false 모사.""" return "" class _Gate: """기존 owner_proof gate 결과 모사(신규 gate 구현 아님 — owner_gate_fn 주입 인터페이스만 재현). ready=True → ANU owner proof 통과. ready=False → self/foreign(차단).""" def __init__(self, ready): self.ready = ready self.status = drv.GATE_READY_STATUS if ready else "OWNER_PICKUP_BLOCKED" class _Spy: """callback_launch_fn in-memory spy — 실 callback/wake 발사 0. task-2763: 실 build_callback_launch_fn 반환 _fn 계약 시그니처 (result_json_path positional + task_id/sha256 keyword-only)에 정합.""" def __init__(self): self.calls = [] def __call__(self, result_json_path, *, task_id, sha256, executor_key="", schedule_id="", terminal_intent="", relay_required=False): self.calls.append({ "result_json_path": 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, }) def _write_live_result(root, task_id, body="ok"): """LIVE_INBOX 에 task-.result.json (final) 작성.""" 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 _write_legacy_result(root, task_id, body="legacy"): """memory/events 직속(legacy) result.json — 감시 경로 밖.""" d = os.path.join(root, "memory", "events") 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 _ledger_lines(root): lp = os.path.join(root, *drv.LIVE_LEDGER_REL.split("/")) if not os.path.isfile(lp): return [] with open(lp, "r", encoding="utf-8") as fh: return [json.loads(x) for x in fh if x.strip()] def _verdicts(records): return [r.verdict for r in records] class P0BEventStrategy2760(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.root = self._tmp.name def tearDown(self): self._tmp.cleanup() # ── 1. retained legacy result 가 scan 안 됨 ──────────────────────────────── def test_legacy_result_not_scanned(self): legacy = _write_legacy_result(self.root, "task-2553") # LIVE_INBOX 는 비어 있음. recs = drv.scan_live_inbox_once( self.root, flag_reader=_enabled, owner_gate_fn=lambda *a, **k: _Gate(True), ) # legacy 는 후보 자체가 아님 → 처리 record 0 (move/ledger/fire 0). self.assertEqual(recs, [], "live inbox 가 비면 처리 record 0 이어야 한다") self.assertTrue(os.path.isfile(legacy), "legacy result 는 무변경(이동/삭제 0)") self.assertEqual(_ledger_lines(self.root), [], "ledger entry 0") # legacy 의 result_path 가 어떤 record 에도 등장하지 않음. self.assertNotIn(legacy, [r.result_path for r in recs]) # ── 2. live inbox 신규 result 1개만 감지 ────────────────────────────────── def test_live_inbox_single_new_detected(self): live = _write_live_result(self.root, "task-2760") _write_legacy_result(self.root, "task-9999") # 잡음(legacy) — 무시돼야 함 spy = _Spy() recs = drv.scan_live_inbox_once( self.root, flag_reader=_enabled, owner_gate_fn=lambda *a, **k: _Gate(True), callback_launch_fn=spy, ) processed = [r for r in recs if r.verdict == drv.VERDICT_LIVE_PROCESSED] self.assertEqual(len(processed), 1, "신규 live result 정확히 1개만 처리") self.assertEqual(os.path.basename(processed[0].result_path), "task-2760.result.json") self.assertFalse(os.path.isfile(live), "성공 시 원본은 processed 로 이동(inbox 비움)") moved = os.path.join(self.root, *drv.LIVE_PROCESSED_DIR_REL.split("/"), "task-2760.result.json") self.assertTrue(os.path.isfile(moved), "원본은 p0b_processed 로 이동") self.assertEqual(len(_ledger_lines(self.root)), 1, "ledger append 1건") self.assertEqual(len(spy.calls), 1, "ANU gate ready → callback 1회 fire") # ── 3. processed ledger 중복 방지 ───────────────────────────────────────── def test_processed_ledger_dedupe(self): # 1차: 처리 → ledger 1건 + move. _write_live_result(self.root, "task-2760", body="X") recs1 = drv.scan_live_inbox_once( self.root, flag_reader=_enabled, owner_gate_fn=lambda *a, **k: _Gate(True), ) self.assertIn(drv.VERDICT_LIVE_PROCESSED, _verdicts(recs1)) self.assertEqual(len(_ledger_lines(self.root)), 1) # 2차: 동일 task_id+동일 내용(=동일 sha) 파일이 다시 inbox 에 나타남 → 재처리 0. _write_live_result(self.root, "task-2760", body="X") spy = _Spy() recs2 = drv.scan_live_inbox_once( self.root, flag_reader=_enabled, owner_gate_fn=lambda *a, **k: _Gate(True), callback_launch_fn=spy, ) self.assertEqual(_verdicts(recs2), [drv.VERDICT_LIVE_SKIP_DEDUPE], "ledger 존재 → SKIP_DEDUPE(중복 방지)") self.assertEqual(len(_ledger_lines(self.root)), 1, "ledger 중복 append 0") self.assertEqual(len(spy.calls), 0, "dedupe 시 fire 0") # ── 4. 동일 (task_id, source_result_sha256) 재처리 0 ────────────────────── def test_same_task_sha_reprocess_zero(self): live = _write_live_result(self.root, "task-2760", body="payload") sha = drv._sha256_file(live) # ledger 에 (task_id, sha) 를 선적재 → 이미 처리된 것으로 간주. lp = os.path.join(self.root, *drv.LIVE_LEDGER_REL.split("/")) os.makedirs(os.path.dirname(lp), exist_ok=True) with open(lp, "w", encoding="utf-8") as fh: fh.write(json.dumps({ "task_id": "task-2760", "source_result_sha256": sha, "processed_at": "seed", "runner_id": drv.LIVE_RUNNER_ID, "outcome": drv.VERDICT_LIVE_PROCESSED, }) + "\n") spy = _Spy() recs = drv.scan_live_inbox_once( self.root, flag_reader=_enabled, owner_gate_fn=lambda *a, **k: _Gate(True), callback_launch_fn=spy, ) self.assertEqual(_verdicts(recs), [drv.VERDICT_LIVE_SKIP_DEDUPE]) self.assertEqual(len(spy.calls), 0, "(task_id,sha256) ledger hit → 재처리/재발사 0") self.assertEqual(len(_ledger_lines(self.root)), 1, "ledger entry 증가 0") self.assertTrue(os.path.isfile(live), "재처리 0 → 원본 이동 없음") # ── 5. busy 시 skip 아닌 delayed/pending 잔존 ───────────────────────────── def test_busy_yields_pending_not_skip(self): live = _write_live_result(self.root, "task-2760") spy = _Spy() recs = drv.scan_live_inbox_once( self.root, flag_reader=_enabled, busy_fn=lambda: True, owner_gate_fn=lambda *a, **k: _Gate(True), callback_launch_fn=spy, ) self.assertEqual(_verdicts(recs), [drv.VERDICT_LIVE_PENDING_BUSY], "busy 는 누락(skip)이 아니라 지연(PENDING)") self.assertTrue(os.path.isfile(live), "busy → inbox 잔존(다음 idle cycle 재감지)") self.assertEqual(_ledger_lines(self.root), [], "busy → ledger append 0") self.assertEqual(len(spy.calls), 0, "busy → fire 0") # ── 6. self-collector 차단 (ANU key 만 fire) ────────────────────────────── def test_self_collector_blocked_only_anu_fires(self): # self 시나리오: owner_proof gate 미통과(ready=False) → fire 0, inbox 잔존. live_self = _write_live_result(self.root, "task-2760", body="self") spy_self = _Spy() recs_self = drv.scan_live_inbox_once( self.root, flag_reader=_enabled, owner_gate_fn=lambda *a, **k: _Gate(False), callback_launch_fn=spy_self, ) self.assertEqual(_verdicts(recs_self), [drv.VERDICT_LIVE_PENDING_OWNER], "self/foreign → owner_proof gate 차단(PENDING)") self.assertEqual(len(spy_self.calls), 0, "self-collector → fire 0") self.assertTrue(os.path.isfile(live_self), "차단 시 inbox 잔존(move 0)") # ANU 시나리오: gate ready → 정확히 1회 fire. spy_anu = _Spy() recs_anu = drv.scan_live_inbox_once( self.root, flag_reader=_enabled, owner_gate_fn=lambda *a, **k: _Gate(True), callback_launch_fn=spy_anu, ) self.assertIn(drv.VERDICT_LIVE_PROCESSED, _verdicts(recs_anu)) self.assertEqual(len(spy_anu.calls), 1, "ANU owner proof 만 fire") # task-2763: collector_role 은 더 이상 호출 인자가 아님(실 _fn 내부 ANU key 강제). # 호출부가 _fn 계약대로 result_json_path/task_id/sha256 을 명시 전달했는지 검증. call0 = spy_anu.calls[0] self.assertEqual(call0["task_id"], "task-2760") self.assertTrue(call0["result_json_path"].endswith("task-2760.result.json")) self.assertTrue(call0["sha256"], "canonical sha256 이 비어있지 않게 전달") # ── 7. ACTIVE=false 유지 ────────────────────────────────────────────────── def test_active_false_full_noop(self): live = _write_live_result(self.root, "task-2760") spy = _Spy() recs = drv.scan_live_inbox_once( self.root, flag_reader=_disabled, # ACTIVE=false owner_gate_fn=lambda *a, **k: _Gate(True), callback_launch_fn=spy, ) self.assertEqual(_verdicts(recs), [drv.VERDICT_NOOP_DISABLED], "ACTIVE=false → 전면 no-op") self.assertEqual(recs[0].activation, drv.ACTIVATION_DISABLED) self.assertTrue(os.path.isfile(live), "ACTIVE=false → 수거/이동 0") self.assertEqual(_ledger_lines(self.root), [], "ACTIVE=false → ledger 0") self.assertEqual(len(spy.calls), 0, "ACTIVE=false → fire 0") # 기본 동작(flag_reader 미주입, 실파일 부재)도 disabled 여야 함. recs_default = drv.scan_live_inbox_once(self.root) self.assertEqual(_verdicts(recs_default), [drv.VERDICT_NOOP_DISABLED]) # ── 8. 기존 task-2758 non-matching thin gate 상태 불변 ──────────────────── def test_task2758_thin_gate_unchanged(self): import dispatch.v2_13_gate as gate self.assertTrue(hasattr(gate, "classify_gate")) self.assertTrue(hasattr(gate, "GateVerdict")) # non-matching thin gate: allowlist 축 텍스트 검출기는 제거된 상태로 유지(재도입 0). self.assertFalse(hasattr(gate, "detect_allowlist_axis"), "matching 축 검출기는 task-2758 에서 제거된 상태 불변") contract = { "goal_id": "g-2760", "scope": "narrow", "blast_radius": "LOW", "allowed_capability_delta": [], "expected_files": ["dispatch/anu_pickup_driver.py"], "allowed_runtime_changes": [], "forbidden_operations": [], "rollback_floor": "LOW", "budget": {}, "exit_criteria": "tests pass", "provenance": "chair_injected", } res = gate.classify_gate(contract, "diff", [], {}) self.assertEqual(res["touched_axes"], [], "matching 제거 → touched_axes 항상 []") self.assertNotEqual(res["verdict"], gate.GateVerdict.ALLOWLIST_AXIS_CHAIR_REQUIRED) # ── 9. matching_axis DESIGN_REVIEW_REQUIRED 유지(matching 재도입 0) ──────── def test_matching_axis_not_reintroduced(self): import dispatch.v2_13_gate as gate # matching detector 부재 확인(재도입 0). self.assertFalse(hasattr(gate, "detect_allowlist_axis")) contract = { "goal_id": "g-2760", "scope": "narrow", "blast_radius": "LOW", "allowed_capability_delta": [], "expected_files": [], "allowed_runtime_changes": [], "forbidden_operations": [], "rollback_floor": "LOW", "budget": {}, "exit_criteria": "x", "provenance": "strict_template", } # diff/tool_calls 에 allowlist 축 텍스트가 있어도 matching 축 추출 0(시그니처 호환만). diff = "touch deploy/systemd/anu-pickup.path; forbidden allowlist axis matched" tool_calls = [{"name": "Edit", "path": "deploy/systemd/anu-pickup.path"}] res = gate.classify_gate(contract, diff, tool_calls, {}) self.assertEqual(res["touched_axes"], [], "matching 축 보류(DESIGN_REVIEW_REQUIRED) → diff 텍스트 무관 touched_axes []") self.assertNotEqual(res["verdict"], gate.GateVerdict.ALLOWLIST_AXIS_CHAIR_REQUIRED, "ALLOWLIST_AXIS_CHAIR_REQUIRED 는 dead trigger(matching 재도입 0)") # ───────────────────────────────────────────────────────────────────────── # task-2760+2 — move-first 핵심 불변식(HIGH 1). 완료(ledger 존재) ⟺ move 성공. # ───────────────────────────────────────────────────────────────────────── # ── 10. move 성공 시 완료 ledger 기록(완료 entry 존재 ⟺ move 성공) ───────── def test_move_success_records_complete_ledger(self): live = _write_live_result(self.root, "task-2760") recs = drv.scan_live_inbox_once( self.root, flag_reader=_enabled, owner_gate_fn=lambda *a, **k: _Gate(True), ) self.assertEqual(_verdicts(recs), [drv.VERDICT_LIVE_PROCESSED], "move 성공 → 완료(PROCESSED)") self.assertFalse(os.path.isfile(live), "move 성공 → 원본 processed 이동(inbox 비움)") moved = os.path.join(self.root, *drv.LIVE_PROCESSED_DIR_REL.split("/"), "task-2760.result.json") self.assertTrue(os.path.isfile(moved), "원본은 p0b_processed 로 이동") lines = _ledger_lines(self.root) self.assertEqual(len(lines), 1, "move 성공 시에만 완료 ledger 1건 기록") self.assertEqual(lines[0]["task_id"], "task-2760") self.assertEqual(lines[0]["outcome"], drv.VERDICT_LIVE_PROCESSED) # ── 11~13. move 실패 시: 완료 ledger 미기록 · PENDING_MOVE · inbox 잔존 ──── def test_move_failure_no_complete_ledger_and_pending(self): live = _write_live_result(self.root, "task-2760") spy = _Spy() orig = drv._move_processed try: drv._move_processed = lambda *a, **k: "processed move 실패: simulated" recs = drv.scan_live_inbox_once( self.root, flag_reader=_enabled, owner_gate_fn=lambda *a, **k: _Gate(True), callback_launch_fn=spy, ) finally: drv._move_processed = orig # 12. move 실패 → VERDICT_LIVE_PENDING_MOVE self.assertEqual(_verdicts(recs), [drv.VERDICT_LIVE_PENDING_MOVE], "move 실패 → 완료 아님(PENDING_MOVE)") # 11. move 실패 → 완료 ledger 미기록(완료 ⟺ move 성공) self.assertEqual(_ledger_lines(self.root), [], "move 실패 시 완료 ledger entry 0 (완료 ⟺ move 성공)") # 13. move 실패 → 원본 inbox 잔존(retry pending) self.assertTrue(os.path.isfile(live), "move 실패 → inbox 잔존(retry pending)") self.assertEqual(recs[0].error, "processed move 실패: simulated") # ── 14. move 실패가 stuck-complete dedupe-skip 안 됨 + 복구 시 재시도 완료 ── def test_move_failure_not_stuck_then_recovers(self): live = _write_live_result(self.root, "task-2760") orig = drv._move_processed try: drv._move_processed = lambda *a, **k: "processed move 실패: simulated" recs1 = drv.scan_live_inbox_once( self.root, flag_reader=_enabled, owner_gate_fn=lambda *a, **k: _Gate(True), ) self.assertEqual(_verdicts(recs1), [drv.VERDICT_LIVE_PENDING_MOVE]) # 다음 cycle: 여전히 move 실패 — 완료 ledger 가 없으므로 SKIP_DEDUPE 로 굳지 않고 # 다시 PENDING_MOVE(재시도). stuck-complete dedupe leak 0. recs2 = drv.scan_live_inbox_once( self.root, flag_reader=_enabled, owner_gate_fn=lambda *a, **k: _Gate(True), ) self.assertNotIn(drv.VERDICT_LIVE_SKIP_DEDUPE, _verdicts(recs2), "move 실패 파일이 완료로 dedupe-skip 되면 안 됨(영구 stuck 금지)") self.assertEqual(_verdicts(recs2), [drv.VERDICT_LIVE_PENDING_MOVE]) self.assertEqual(_ledger_lines(self.root), [], "pending 동안 완료 ledger 0") finally: drv._move_processed = orig # move 복구 후 재시도 → 정상 완료(PROCESSED) + 완료 ledger 1 + 원본 이동. recs3 = drv.scan_live_inbox_once( self.root, flag_reader=_enabled, owner_gate_fn=lambda *a, **k: _Gate(True), ) self.assertIn(drv.VERDICT_LIVE_PROCESSED, _verdicts(recs3), "move 복구 시 pending → 정상 완료(재시도 가능)") self.assertFalse(os.path.isfile(live), "복구 후 원본 processed 이동") self.assertEqual(len(_ledger_lines(self.root)), 1, "복구 후 완료 ledger 1건") # ───────────────────────────────────────────────────────────────────────── # task-2760+2 — compaction/robustness 제거 검증(범위 축소 replacement). # ledger compaction/self-healing/robustness 는 본 PR 범위 밖(DESIGN_BACKLOG). # driver 는 단순 append-only ledger 읽기/쓰기만 유지한다. # ───────────────────────────────────────────────────────────────────────── # ── 15. LIVE_LEDGER_MAX_ENTRIES 상수 없음 ───────────────────────────────── def test_no_live_ledger_max_entries(self): self.assertFalse(hasattr(drv, "LIVE_LEDGER_MAX_ENTRIES"), "compaction 상한 상수는 본 PR 에서 제거(범위 분리)") self.assertNotIn("LIVE_LEDGER_MAX_ENTRIES", getattr(drv, "__all__", []), "__all__ 에서도 노출 0") # ── 16. _rewrite_ledger 함수 없음 ───────────────────────────────────────── def test_no_rewrite_ledger(self): self.assertFalse(hasattr(drv, "_rewrite_ledger"), "atomic 재기록(compaction) 함수는 본 PR 에서 제거") # ── 17. _load_processed_ledger compaction trim 없음(헬퍼 자체 부재) ──────── def test_no_load_processed_ledger_helper_with_trim(self): self.assertFalse(hasattr(drv, "_load_processed_ledger"), "compaction trim 헬퍼는 본 PR 에서 제거(단순 인라인 읽기로 대체)") import inspect src = inspect.getsource(drv.scan_live_inbox_once) for token in ("max_entries", "order[-", "_rewrite_ledger", "kept", ".compact.tmp"): self.assertNotIn(token, src, f"compaction trim 잔재 토큰 없음: {token!r}") # ── 18. ledger robustness 신규 방어 로직 없음(단순 append-only 유지) ─────── def test_no_new_ledger_robustness(self): import inspect src = inspect.getsource(drv) # MEDIUM 4 robustness 신규 방어(glob.escape · 손상/불완전 튜플 self-healing · # compaction 재기록)가 모듈 어디에도 재도입되지 않았는지 확인. # (os.replace 는 기존 base 함수 _move_processed 의 atomic rename 으로 정당 — compaction # 전용 잔재 토큰만 검사한다.) for token in ("glob.escape", "LIVE_LEDGER_MAX_ENTRIES", "_rewrite_ledger", "_load_processed_ledger", ".compact.tmp"): self.assertNotIn(token, src, f"ledger robustness 신규 방어 잔재 없음: {token!r}") # move-first 불변식 라벨은 유지(범위 축소 후에도 핵심 불변식 보존). self.assertTrue(hasattr(drv, "VERDICT_LIVE_PENDING_MOVE"), "move-first 불변식 verdict 는 유지") self.assertIn("VERDICT_LIVE_PENDING_MOVE", drv.__all__) if __name__ == "__main__": unittest.main()