# -*- coding: utf-8 -*- """ regression: scripts/task2553_closeout_collect.py (task-2553+50 Track 1) 검증 축 ------- 1. Layer A NO-CRON: collector 스크립트 AST 에 subprocess/os.system/Popen/ cokacdir/cron-dispatch import·call 0. 2. read-only consume: collect() 전후 frozen anchor 6모듈 + consumed upstream 산출물 byte-0 무변(SHA256 EQUAL). 3. write surface = §4 allowlist 한정 — 비-allowlist write 거부. 4. 9 필수구분(회장 §2 1~9) 전수 산출 + 정형 schema. 5. hold_for_chair=false + §6 트리거 전수 non-operative. 6. git invariant·frozen byte-0 캡처 정확. 7. pending→완료 과장 0 (remaining-backlog 에 OPEN 항목 명시 존재). 8. 산출 파일 생성(temp out-root 격리) — 실 repo write 0(테스트 자체). """ import ast import hashlib import json import os import sys import unittest REPO = "/home/jay/workspace" SCRIPT_REL = "scripts/task2553_closeout_collect.py" SCRIPT = os.path.join(REPO, SCRIPT_REL) sys.path.insert(0, os.path.join(REPO, "scripts")) import task2553_closeout_collect as C # noqa: E402 def _sha(path): h = hashlib.sha256() with open(path, "rb") as fh: for ch in iter(lambda: fh.read(65536), b""): h.update(ch) return h.hexdigest() class LayerANoCronTest(unittest.TestCase): def test_01_no_subprocess_cokacdir_cron_dispatch(self): src = open(SCRIPT, "r", encoding="utf-8").read() tree = ast.parse(src) banned_import = {"subprocess", "pty"} for node in ast.walk(tree): if isinstance(node, ast.Import): for n in node.names: self.assertNotIn(n.name.split(".")[0], banned_import, f"banned import {n.name}") if isinstance(node, ast.ImportFrom): root = (node.module or "").split(".")[0] self.assertNotIn(root, banned_import, f"banned from-import {node.module}") if isinstance(node, ast.Call): f = node.func if isinstance(f, ast.Attribute): self.assertNotIn(f.attr, {"system", "Popen", "popen", "spawn", "spawnv", "fork", "exec"}, f"banned call attr {f.attr}") low = src.lower() # 실행되는 cokacdir/cron-direct/dispatch 문자열 0 for tok in ("/usr/local/bin/cokacdir", "--sendfile", "--cron", "subprocess.run", "os.system("): self.assertNotIn(tok, src, f"banned token present: {tok}") self.assertNotIn("cokacdir --", low) class ReadOnlyConsumeTest(unittest.TestCase): def test_02_frozen_and_consumed_byte0_unchanged_after_collect(self): targets = list(C.FROZEN_BYTE0.keys()) + [ r for r in C.CONSUMED_ARTIFACTS if os.path.isfile(os.path.join(REPO, r)) ] before = {r: _sha(os.path.join(REPO, r)) for r in targets} produced = C.collect(REPO) after = {r: _sha(os.path.join(REPO, r)) for r in targets} self.assertEqual(before, after, "read-only 위반 — 소스 산출물 변경됨") self.assertIsInstance(produced, dict) def test_03_frozen_byte0_all_equal_expected(self): fz = C.verify_frozen_byte0(REPO) self.assertEqual(len(fz), 6) for rel, v in fz.items(): self.assertTrue(v["present"], f"{rel} missing") self.assertTrue(v["equal"], f"{rel} byte-0 drift {v}") class AllowlistWriteSurfaceTest(unittest.TestCase): def test_04_reject_non_allowlisted_write(self): with self.assertRaises(RuntimeError): C._assert_allowlisted("memory/events/NOT-ALLOWED.json") # allowlist 항목은 통과 for rel in C.ALLOWLIST: C._assert_allowlisted(rel) def test_05_write_outputs_only_allowlisted(self): import tempfile produced = C.collect(REPO) with tempfile.TemporaryDirectory() as td: written = C.write_outputs(td, produced) for rel in written: self.assertIn(rel, C.ALLOWLIST) self.assertTrue(os.path.isfile(os.path.join(td, rel))) # 실 repo 의 closeout 산출물은 본 테스트가 건드리지 않았는지(temp 격리) self.assertGreaterEqual(len(written), 8) def test_06_track2_3_disjoint_no_overlap_tokens(self): # allowlist 의 closeout/+50 경로는 Track2/3 산출물 네임스페이스와 분리 for rel in C.ALLOWLIST: self.assertNotIn("track2", rel.lower()) self.assertNotIn("track3", rel.lower()) class NineDistinctionsTest(unittest.TestCase): @classmethod def setUpClass(cls): cls.out = C.collect(REPO) def test_07_nine_categories_present(self): result = self.out["memory/events/task-2553.final-closeout.result.json"] nd = result["nine_distinctions"] for n in range(1, 10): key = [k for k in nd if k.startswith(f"cat{n}_")] self.assertEqual(len(key), 1, f"cat{n} 누락/중복: {key}") self.assertIn("title", nd[key[0]]) def test_08_cat7_selfchain_vs_independent_anu(self): nd = self.out["memory/events/task-2553.final-closeout.result.json"]["nine_distinctions"] c7 = nd["cat7_selfchain_vs_independent_anu"] self.assertIn("QUARANTINED", json.dumps(c7, ensure_ascii=False)) self.assertIn("AUTHORITATIVE_VERDICT_PENDING", json.dumps(c7, ensure_ascii=False)) self.assertEqual(c7["independent_anu"]["definition"].count("c119085addb0f8b7"), 1) def test_09_cat8_cancel_on_success_tiers(self): nd = self.out["memory/events/task-2553.final-closeout.result.json"]["nine_distinctions"] c8 = nd["cat8_cancel_on_success_tiers"] self.assertEqual(c8["live"]["status"], "NOT_PERFORMED") self.assertEqual(c8["mock"]["ref"], "+41") self.assertEqual(c8["isolated_e2e"]["ref"], "+48") def test_10_cat9_policy_profile_engine_seam(self): nd = self.out["memory/events/task-2553.final-closeout.result.json"]["nine_distinctions"] c9 = nd["cat9_policy_profile_engine_seam_scope"] ppe = c9["completed_scope"] self.assertTrue(ppe["engine_byte0_frozen"]) # 병렬 Track3 가 profile 을 landed 했을 수 있음 — closeout 은 관측된 현실을 # 정확히 기록해야 한다(강제 부재 아님). 실 디렉터리 스캔과 일치 + owner=Track3. actual_present = os.path.isfile( os.path.join(REPO, "memory/policy_profiles/task_2553_final_closeout.json")) self.assertEqual(ppe["task_2553_final_closeout_profile_present"], actual_present, "관측값이 실 디렉터리와 불일치(read-only 정확성 위반)") self.assertIn("Track3", ppe["task_2553_final_closeout_profile_owner"]) self.assertIn("read-only", ppe["track1_disjoint_note"]) class CloseoutIntegrityTest(unittest.TestCase): @classmethod def setUpClass(cls): cls.out = C.collect(REPO) def test_11_hold_for_chair_false(self): dec = self.out["memory/events/task-2553.final-closeout.decision.json"] res = self.out["memory/events/task-2553.final-closeout.result.json"] self.assertFalse(dec["hold_for_chair"]) self.assertFalse(res["hold_for_chair"]) for v in dec["hold_assessment"]["triggers"].values(): self.assertFalse(v, f"HOLD 트리거 적중: {v}") def test_12_git_invariant_branch_match(self): inv = self.out["memory/events/task-2553.final-closeout.result.json"]["invariants"] self.assertEqual(inv["git_branch"], C.EXPECTED_GIT_BRANCH) self.assertTrue(inv["git_branch_matches_expected"]) self.assertTrue(inv["frozen_byte0_all_equal"]) def test_13_no_overclaim_pending_open_present(self): bl = self.out["memory/events/task-2553.remaining-backlog_260518.json"] opens = [b for b in bl["backlog"] if str(b.get("status", "")).startswith("OPEN")] self.assertTrue(opens, "pending 항목 OPEN 명시 누락 — 과장 방지 실패") ids = {b["id"] for b in bl["backlog"]} self.assertIn("BL-1", ids) # +44_46 tri-state MANDATORY follow-up def test_14_pilot_readiness_no_autostart(self): pr = self.out["memory/events/task-2553.operational-pilot-readiness_260518.json"] self.assertIn("별도 회장 GO", pr["readiness_gate"]) self.assertGreaterEqual(len(pr["candidates"]), 3) cur = pr["engine_auto_resolve_registration"]["current"] self.assertTrue(cur.startswith("PRESENT") or cur.startswith("ABSENT"), f"engine_auto_resolve current 비정형: {cur}") self.assertEqual(pr["engine_auto_resolve_registration"]["owner"], "Track3 (mapping 신설)") def test_15_callback_independent_anu_key_only(self): dec = self.out["memory/events/task-2553.final-closeout.decision.json"] ca = dec["callback_a"] self.assertIn("c119085addb0f8b7", ca["rule"]) self.assertIn("109fa85250c6d46b", ca["rule"]) # executor self key 금지 명시 self.assertEqual(ca["collector_role"], "ANU") res = self.out["memory/events/task-2553+50.result.json"] self.assertIn("c119085addb0f8b7", res["callback_a_fired_with"]) def test_16_consolidated_md_and_plus50_records(self): md = self.out["memory/reports/task-2553.final-closeout-consolidated-summary_260518.md"] self.assertIn("FINAL CLOSEOUT", md) for n in range(1, 10): self.assertIn(f"§2.{n}", md) self.assertIn("memory/events/task-2553+50.decision.json", self.out) self.assertIn("memory/events/task-2553+50.result.json", self.out) self.assertIn("memory/reports/task-2553+50.md", self.out) def test_17_durable_registry_completed_tasks(self): reg = self.out["memory/events/task-2553.final-closeout.result.json"]["durable_registry"] self.assertGreaterEqual(reg["completed_records"], 1) # +47,+48,+44_46 등 COMPLETED durable-success 가 ledger 에서 관측 self.assertTrue(any("task-2553+47" in t for t in reg["completed_durable_success_tasks"])) class GitRefInvariantTest(unittest.TestCase): def test_18_repo_head_branch_equal_before_after_collect(self): before = C.read_git_invariant(REPO) C.collect(REPO) after = C.read_git_invariant(REPO) self.assertEqual(before, after) self.assertEqual(after["branch"], C.EXPECTED_GIT_BRANCH) if __name__ == "__main__": unittest.main(verbosity=2)