"""런 원장을 쓰는 도구. `bin/task.py` 가 이 계약의 초안이다 — flock 을 잡고 임시 파일에 완성한 뒤 원자적으로 바꾸고, `startedAt` 부터 지금까지를 누적에 더한다. 여기서는 그것이 `runs/` 쪽에서도 성립하는지 본다. """ import importlib.util import json import os import subprocess import tempfile import time import unittest ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) TOOL = os.path.join(ROOT, "scripts", "run-ledger.py") _spec = importlib.util.spec_from_file_location("run_ledger", TOOL) rl = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(rl) def _cli(*args, **kw): return subprocess.run(["python3", TOOL, *args], cwd=ROOT, capture_output=True, text=True, **kw) class LedgerTest(unittest.TestCase): def setUp(self): self.dir = tempfile.mkdtemp() self.led = os.path.join(self.dir, "runs", "demo", "2026-01-01-0000", "run.json") p = _cli("open", self.led, "--project", "demo", "--record", "docs/demo/x.md") self.assertEqual(0, p.returncode, p.stderr) def _read(self): return json.load(open(self.led, encoding="utf-8")) def _stage(self, sid): return next(s for s in self._read()["stages"] if s["id"] == sid) # ── 상태를 지킨다 ──────────────────────────────────────────────── def test_opening_twice_is_refused(self): """이미 있는 런을 덮어쓰지 않는다. 이어서 하려면 status 로 본다.""" p = _cli("open", self.led, "--project", "demo", "--record", "x") self.assertEqual(2, p.returncode) def test_two_stages_cannot_be_open_at_once(self): """단계를 겹쳐 열지 않는다. 병렬을 꺼도 같은 상태에서 이어갈 수 있어야 한다.""" _cli("begin", self.led, "--stage", "S3") p = _cli("begin", self.led, "--stage", "S5") self.assertEqual(2, p.returncode) self.assertIn("RUNNING", p.stderr) def test_an_unskippable_stage_cannot_be_skipped(self): _cli("begin", self.led, "--stage", "S3") p = _cli("end", self.led, "--stage", "S3", "--status", "SKIPPED", "--why", "x") self.assertEqual(2, p.returncode) def test_a_skipped_stage_needs_a_reason(self): """적지 않고 빠뜨린 것과 판단해서 건너뛴 것을 구분해야 한다.""" _cli("begin", self.led, "--stage", "S4") self.assertEqual(2, _cli("end", self.led, "--stage", "S4", "--status", "SKIPPED").returncode) self.assertEqual(0, _cli("end", self.led, "--stage", "S4", "--status", "SKIPPED", "--why", "그림이 필요 없다").returncode) def test_a_done_stage_needs_a_skill_receipt(self): """안 연 스킬의 영수증을 적으면 그것이 지어낸 것이다. 빈 채로 DONE 을 못 찍는다.""" _cli("begin", self.led, "--stage", "S3") self.assertEqual(2, _cli("end", self.led, "--stage", "S3", "--status", "DONE").returncode) self.assertEqual(0, _cli("end", self.led, "--stage", "S3", "--status", "DONE", "--echo", "본문이 있는 종류는 Case 와 Concept 둘뿐이다.").returncode) def test_a_gate_must_carry_an_exit_code(self): """종료 코드 없는 관문을 못 적는다. 돌리지 않고 적는 경로를 막는다.""" p = _cli("gate", self.led, "--stage", "S3", "--cmd", "x") self.assertNotEqual(0, p.returncode) # ── 끊겨도 이어진다 ────────────────────────────────────────────── def test_the_file_is_never_half_written(self): """쓰는 도중에 죽여도 읽는 쪽은 이전 판이나 다음 판 중 하나를 본다.""" for i in range(40): pr = subprocess.Popen(["python3", TOOL, "gate", self.led, "--stage", "S6", "--cmd", f"x{i}", "--exit", "0"], cwd=ROOT, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) time.sleep(0.001 * (i % 7)) pr.kill() pr.wait() json.load(open(self.led, encoding="utf-8")) # 깨졌으면 여기서 터진다 def test_concurrent_writes_do_not_lose_updates(self): """락이 없으면 마지막에 쓴 것만 남는다.""" ps = [subprocess.Popen(["python3", TOOL, "gate", self.led, "--stage", "S6", "--cmd", f"c{i}", "--exit", "0"], cwd=ROOT, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) for i in range(20)] for p in ps: p.wait() self.assertEqual(20, len(self._stage("S6")["gates"])) def test_an_interrupted_stage_is_closed_and_continued(self): """앞 세션이 열어 둔 채 끊기면, 그 구간을 닫고 잇는다. 닫은 구간에는 세션이 없던 시간이 섞이므로 누적에만 더하지 않고 따로 적는다 — 「이 단계가 오래 걸렸다」와 「중간에 끊겼다」는 다른 말이다. """ _cli("begin", self.led, "--stage", "S5") d = self._read() import datetime s5 = next(s for s in d["stages"] if s["id"] == "S5") s5["startedAt"] = (datetime.datetime.now().astimezone() - datetime.timedelta(seconds=90)).isoformat(timespec="seconds") json.dump(d, open(self.led, "w", encoding="utf-8"), ensure_ascii=False, indent=2) p = _cli("begin", self.led, "--stage", "S5") self.assertEqual(0, p.returncode, p.stderr) after = self._stage("S5") self.assertGreaterEqual(after["elapsedSeconds"], 89) self.assertEqual(1, len(after["interruptions"])) self.assertIn("세션이 없던 시간", after["interruptions"][0]["note"]) def test_status_says_where_it_stopped(self): """새 세션이 원장만 읽고 어디서 이어야 하는지 알 수 있어야 한다.""" _cli("begin", self.led, "--stage", "S3") p = _cli("status", self.led) self.assertEqual(1, p.returncode, "끊긴 자리가 있으면 0 이 아니다") self.assertIn("S3", p.stdout) self.assertIn("끊긴 자리", p.stdout) # ── 단계 밖의 시간 ─────────────────────────────────────────────── def test_a_rider_is_recorded_outside_the_stages(self): """런 도중에 끼어든 일은 단계가 아니라서 어느 칸에도 안 남았다.""" _cli("rider", self.led, "--id", "R11", "--why", "마스킹 되돌이", "--seconds", "1070") riders = self._read()["riders"] self.assertEqual(["R11"], [r["id"] for r in riders]) self.assertEqual(1070, riders[0]["seconds"]) self.assertIn("라이더", _cli("status", self.led).stdout) def test_a_rider_records_which_stage_it_interrupted(self): _cli("begin", self.led, "--stage", "S3") _cli("rider", self.led, "--id", "R12", "--why", "x", "--seconds", "60") self.assertEqual("S3", self._read()["riders"][0]["duringStage"]) # ── 두 세션의 기록이 섞이지 않는다 ──────────────────────────────── def test_a_superseded_session_cannot_write(self): """파일이 안 깨지는 것과 두 세션의 기록이 안 섞이는 것은 다른 일이다. `plan/02` B-005 — 「이전 작업자의 종료를 확인하고 재배정한 시도 번호와 맞는 결과만 받는다.」 `bin/task.py` 의 attempt 와 같은 자리다. """ _cli("begin", self.led, "--stage", "S3", "--session", "sess-1") _cli("begin", self.led, "--stage", "S3", "--session", "sess-2") for args in (("gate", self.led, "--stage", "S3", "--cmd", "stale", "--exit", "0"), ("end", self.led, "--stage", "S3", "--status", "DONE", "--echo", "x")): with self.subTest(cmd=args[0]): p = _cli(*args, "--session", "sess-1") self.assertEqual(2, p.returncode) self.assertIn("sess-2", p.stderr) def test_a_stale_generation_is_refused(self): _cli("begin", self.led, "--stage", "S3", "--session", "sess-1") _cli("begin", self.led, "--stage", "S3", "--session", "sess-2") p = _cli("gate", self.led, "--stage", "S3", "--cmd", "x", "--exit", "0", "--session", "sess-2", "--generation", "1") self.assertEqual(2, p.returncode) def test_an_anonymous_write_to_an_owned_stage_is_refused(self): """누가 썼는지 안 밝히면 나중에 원장을 읽을 수 없다.""" _cli("begin", self.led, "--stage", "S3", "--session", "sess-1") p = _cli("gate", self.led, "--stage", "S3", "--cmd", "x", "--exit", "0") self.assertEqual(2, p.returncode) def test_the_current_owner_writes_and_is_recorded(self): """대조군. 지금 주인의 쓰기는 들어가고, 누가 어느 세대에 썼는지 남는다.""" _cli("begin", self.led, "--stage", "S3", "--session", "sess-1") p = _cli("gate", self.led, "--stage", "S3", "--cmd", "ok", "--exit", "0", "--session", "sess-1", "--generation", "1") self.assertEqual(0, p.returncode, p.stderr) gate = self._stage("S3")["gates"][0] self.assertEqual(("sess-1", 1), (gate["session"], gate["generation"])) def test_an_unowned_stage_still_accepts_writes(self): """주인이 없으면 (세션을 안 쓰는 단일 세션 사용) 그대로 쓴다. 기존 쓰임을 안 깬다.""" _cli("begin", self.led, "--stage", "S3") self.assertEqual(0, _cli("gate", self.led, "--stage", "S3", "--cmd", "x", "--exit", "0").returncode) # ── 옛 원장을 깨지 않는다 ───────────────────────────────────────── def test_the_added_fields_do_not_break_the_verifier(self): """이 도구가 더한 칸이 있어도 검사기가 그대로 읽어야 한다.""" real = os.path.join(ROOT, "runs/document-haness/2026-09-10-1033/run.json") if not os.path.exists(real): self.skipTest("견줄 실제 원장이 없다") d = json.load(open(real, encoding="utf-8")) d.update({"revision": 1, "riders": [], "sessions": [], "updatedAt": "x"}) for st in d["stages"]: st.update({"startedAt": None, "finishedAt": None, "elapsedSeconds": 0}) out = os.path.join(self.dir, "runs", "document-haness", "2026-09-10-1033") os.makedirs(out, exist_ok=True) copy = os.path.join(out, "run.json") json.dump(d, open(copy, "w", encoding="utf-8"), ensure_ascii=False, indent=2) p = subprocess.run(["python3", os.path.join(ROOT, "scripts", "verify-pipeline-run.py"), copy], cwd=ROOT, capture_output=True, text=True) self.assertEqual(0, p.returncode, p.stdout + p.stderr) if __name__ == "__main__": unittest.main()