"""런 원장을 쓰는 도구. `bin/task.py` 가 이 계약의 초안이다 — flock 을 잡고 임시 파일에 완성한 뒤 원자적으로 바꾸고, `startedAt` 부터 지금까지를 누적에 더한다. 여기서는 그것이 `runs/` 쪽에서도 성립하는지 본다. """ import importlib.util import glob import json import os import subprocess import sys 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) VERIFIER = os.path.join(ROOT, "scripts", "verify-pipeline-run.py") _vspec = importlib.util.spec_from_file_location("verify_pipeline_run", VERIFIER) vpr = importlib.util.module_from_spec(_vspec) _vspec.loader.exec_module(vpr) # 지금 `writing-tech-log-records/SKILL.md` 에 있는 문장. 영수증으로 적으면 통과해야 한다. CURRENT_SENTENCE = "본문이 있는 종류는 Case·Concept·Setup 셋이다." 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", CURRENT_SENTENCE).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): """이 도구가 더한 칸이 있어도 검사기가 그대로 읽어야 한다.""" # 프로젝트 이름을 적지 않는다 — verify-pipeline.py 의 FORBIDDEN_LITERAL 가드가 # scripts/ 안에서 저장소 체크아웃 이름을 금지한다. 실제 원장 하나를 고른다. # # **아무거나 고르면 안 된다.** 이 시험이 묻는 것은 「더한 칸이 검사기를 깨뜨리는가」 # 이므로 밑바탕은 **원래 통과하는 원장**이어야 한다. 파이프라인을 돌리는 중에는 # 단계가 RUNNING·PENDING 인 원장이 `runs/` 에 있고, 그걸 고르면 더한 칸과 아무 # 상관없이 「끝나지 않은 단계가 있다」로 실패한다 — 실제로 그렇게 깨졌다. # 단계가 전부 닫힌 것만 고른다. found = [] for cand in sorted(glob.glob(os.path.join(ROOT, "runs", "*", "*", "run.json"))): try: stages = json.load(open(cand, encoding="utf-8")).get("stages") or [] except (OSError, ValueError): continue if stages and all(s.get("status") in ("DONE", "SKIPPED") for s in stages): found.append(cand) if not found: self.skipTest("단계가 전부 닫힌 실제 원장이 없다") real = found[-1] 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", os.path.basename(os.path.dirname(os.path.dirname(real))), os.path.basename(os.path.dirname(real))) 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) def _git(*args): """시험이 저장소에 직접 묻는다. 검사기와 **다른 방법**으로 물어야 대조가 된다.""" p = subprocess.run(["git", "-C", ROOT, *args], capture_output=True, encoding="utf-8", errors="replace") return p.stdout if p.returncode == 0 else None def _commit_that_still_had(skill, sentence): """그 문장을 아직 담고 있던 가장 최근 커밋. 없으면 None. 검사기는 `ls-tree` + `cat-file` 로 본문을 모아 부분 문자열을 찾는다. 여기서는 `git grep` 으로 묻는다 — 같은 코드로 확인하면 시험이 아무것도 안 보는 것이 된다. """ rel = f".agents/skills/{skill}" out = _git("log", "--max-count=200", "--format=%H", "--", rel) or "" for commit in out.split(): got = subprocess.run(["git", "-C", ROOT, "grep", "-F", "-q", sentence, commit, "--", rel], capture_output=True) if got.returncode == 0: return commit return None def _a_sentence_from(skill): """그 스킬의 SKILL.md 에서 지금 실재하는 한 줄. 문구를 시험에 박아 두지 않는다.""" path = os.path.join(ROOT, ".agents", "skills", skill, "SKILL.md") for line in open(path, encoding="utf-8"): line = line.strip() if len(line) >= 30 and not line.startswith(("#", "|", "-", ">", "`")): return line raise AssertionError(f"{skill}/SKILL.md 에서 쓸 만한 줄을 못 찾았다") @unittest.skipUnless(_git("rev-parse", "--git-dir"), "저장소가 아니라 과거를 볼 수 없다") class EchoAgainstSkillHistory(unittest.TestCase): """영수증이 지금 스킬에 없을 때, 위조와 「그 뒤에 스킬이 고쳐졌다」를 가르는지 본다. 스킬은 고쳐진다. 2026-09-12 에 `writing-tech-log-records` 의 「본문이 있는 종류」 문장을 고쳤고, 그 문장을 인용한 과거 원장 4건이 한꺼번에 error 가 됐다. **그 영수증은 사실이다** — 그때 그 문장이 거기 있었다. 원장을 고쳐 쓰는 것은 위조이고 틀린 문장을 스킬에 되살리는 것은 검사기에 답하는 것이라, 둘 다 하지 않고 검사기가 가른다. """ SKILL = "writing-tech-log-records" # 2026-09-12 에 물러난 문장. Studio 의 여섯 번째 종류 SETUP 이 빠져 있던 것을 메우며 # 바뀌었다. **이것을 현재 SKILL.md 에 되살리지 않는다** — 과거 커밋에만 있어야 한다 RETIRED = "본문이 있는 종류는 Case 와 Concept 둘뿐이다." FABRICATED = "이 문장은 그 스킬의 어느 판에도 없다 한 글자도 없다 정말로" def setUp(self): self.dir = tempfile.mkdtemp() def _ledger(self, echo, revision="", carry_field=True): """S3 의 영수증만 갈아 끼운, 그 밖에는 흠이 없는 원장.""" run = json.load(open(vpr.TEMPLATE, encoding="utf-8")) run.update({"runId": "2026-01-01-0000", "project": "demo", "record": "CLAUDE.md", "startedAt": "2026-01-01T00:00:00+09:00"}) for st in run["stages"]: spec = vpr.STAGES[st["id"]] if not carry_field: st.pop(vpr.REVISION_FIELD, None) elif st["id"] == "S3": st[vpr.REVISION_FIELD] = revision or None if spec["skippable"]: st.update({"status": "SKIPPED", "skipReason": "이 시험은 영수증만 본다"}) continue st.update({ "status": "DONE", "skillEcho": echo if st["id"] == "S3" else _a_sentence_from(st["skill"]), "gates": [{"cmd": tok, "exit": 0} for tok in spec["gates"]], }) path = os.path.join(self.dir, f"run-{len(os.listdir(self.dir))}.json") json.dump(run, open(path, "w", encoding="utf-8"), ensure_ascii=False, indent=2) return path def _run(self, path, *args): p = subprocess.run([sys.executable, VERIFIER, path, *args], cwd=ROOT, capture_output=True, text=True) return p.returncode, p.stdout + p.stderr def test_현재_스킬에_있는_영수증은_통과한다(self): """대조군. 이 자리가 통과하지 않으면 나머지 둘은 아무것도 말하지 않는다.""" code, out = self._run(self._ledger(CURRENT_SENTENCE)) self.assertEqual(0, code, out) self.assertIn("대조 못 한 영수증 0", out) self.assertNotIn("그 뒤에 스킬이 고쳐져", out) def test_과거_판에만_있는_영수증은_error_가_아니라_warn_이다(self): code, out = self._run(self._ledger(self.RETIRED)) self.assertEqual(0, code, out) self.assertIn("그 뒤에 스킬이 고쳐져 영수증을 대조할 수 없다", out) self.assertNotIn("스킬 영수증이 그 스킬의 문장이 아니다", out) self.assertIn("대조 못 한 영수증 1", out) # 어느 커밋에 있었는지 함께 적는다. 「과거 어딘가」로는 다시 찾아갈 수 없다 commit = _commit_that_still_had(self.SKILL, self.RETIRED) self.assertIsNotNone(commit, "그 문장을 담은 커밋이 이력에 없다") self.assertIn(commit[:12], out) def test_warn_은_통과가_아니다(self): """초록으로 보이면 안 된다. `--strict` 에서는 이것이 실패다.""" code, out = self._run(self._ledger(self.RETIRED), "--strict") self.assertEqual(1, code, out) def test_어느_판에도_없는_영수증은_error_다(self): code, out = self._run(self._ledger(self.FABRICATED)) self.assertEqual(1, code, out) self.assertIn("스킬 영수증이 그 스킬의 문장이 아니다", out) self.assertNotIn("그 뒤에 스킬이 고쳐져", out) def test_원장이_적은_리비전이_있으면_그_커밋을_본다(self): """새 원장은 대조를 싸게 만든다 — 이력을 훑지 않고 적힌 커밋만 본다.""" commit = _commit_that_still_had(self.SKILL, self.RETIRED) self.assertIsNotNone(commit) code, out = self._run(self._ledger(self.RETIRED, revision=commit)) self.assertEqual(0, code, out) self.assertIn("원장이 적은 리비전", out) def test_그_칸이_없는_옛_원장도_같은_답을_낸다(self): """`skillRevision` 을 모르는 원장은 이력 훑기로 떨어진다. 칸이 없다고 잡지 않는다.""" path = self._ledger(self.RETIRED, carry_field=False) self.assertNotIn(vpr.REVISION_FIELD, json.load(open(path, encoding="utf-8"))["stages"][2]) code, out = self._run(path) self.assertEqual(0, code, out) self.assertIn("그 뒤에 스킬이 고쳐져 영수증을 대조할 수 없다", out) def test_git_이_없으면_못_봤다고_한다(self): """과거를 볼 수 없는 것은 「없다」가 아니다. error 로 올리지 않는다.""" empty = os.path.join(self.dir, "bin") os.makedirs(empty, exist_ok=True) env = dict(os.environ, PATH=empty) # PATH 를 비우면 python3 도 같이 사라진다. 해석기는 절대 경로로 부른다 p = subprocess.run([sys.executable, VERIFIER, self._ledger(self.RETIRED)], cwd=ROOT, capture_output=True, text=True, env=env) out = p.stdout + p.stderr self.assertEqual(0, p.returncode, out) self.assertIn("스킬의 과거 본문을 못 봐서 영수증을 대조하지 못했다", out) self.assertIn("대조 못 한 영수증 1", out) class RunByNamesAManagedAgent(unittest.TestCase): """단계를 **누가** 돌렸는지가 원장에 남는가. `skillEcho` 는 「스킬을 열었다」를 증명하지만 누가 열었는지는 증명하지 않는다 — 매번 새로 띄운 일반 에이전트도 SKILL.md 를 읽고 한 줄을 옮겨 적을 수 있다. 그동안 `runBy` 는 `"subagent"` 라는 상수였고, 그래서 원장 5건이 전부 통과하는 동안에도 어느 에이전트가 돌았는지는 아무 데도 없었다. """ def setUp(self): self.dir = tempfile.mkdtemp() def _ledger(self, schema=2, run_by=None): """runBy 만 갈아 끼운, 그 밖에는 흠이 없는 원장.""" run = json.load(open(vpr.TEMPLATE, encoding="utf-8")) run.update({"runId": "2026-01-01-0000", "project": "demo", "record": "CLAUDE.md", "startedAt": "2026-01-01T00:00:00+09:00", "finishedAt": "2026-01-01T01:00:00+09:00", "schemaVersion": schema}) for st in run["stages"]: spec = vpr.STAGES[st["id"]] if run_by is not None: st["runBy"] = run_by if spec["skippable"]: st.update({"status": "SKIPPED", "skipReason": "이 시험은 runBy 만 본다"}) continue st.update({"status": "DONE", "skillEcho": _a_sentence_from(st["skill"]), "gates": [{"cmd": tok, "exit": 0} for tok in spec["gates"]]}) path = os.path.join(self.dir, f"run-{len(os.listdir(self.dir))}.json") json.dump(run, open(path, "w", encoding="utf-8"), ensure_ascii=False, indent=2) return path def _run(self, path, *args): p = subprocess.run([sys.executable, VERIFIER, path, *args], cwd=ROOT, capture_output=True, text=True) return p.returncode, p.stdout + p.stderr # ── 계약이 스스로 맞는가 ───────────────────────────────────────── def test_모든_단계에_에이전트가_배정돼_있다(self): """빠진 단계가 있으면 그 단계만 조용히 일반 에이전트로 돌아간다.""" for sid in vpr.ORDER: with self.subTest(stage=sid): self.assertTrue(vpr.STAGES[sid].get("agent"), f"{sid} 에 agent 가 없다") def test_배정된_에이전트가_실재한다(self): """`.claude/agents/<이름>.md` 가 없으면 그 이름은 약속일 뿐이다.""" for sid in vpr.ORDER: agent = vpr.STAGES[sid]["agent"] with self.subTest(stage=sid, agent=agent): self.assertTrue( os.path.exists(os.path.join(ROOT, ".claude", "agents", f"{agent}.md")), f"{sid} 이 가리키는 .claude/agents/{agent}.md 가 없다") def test_틀과_검사기가_같은_에이전트를_말한다(self): """틀에만 적어 두면 STAGES 와 갈린다. 갈린 채로는 둘 다 「계약」이라고 말한다.""" run = json.load(open(vpr.TEMPLATE, encoding="utf-8")) for st in run["stages"]: with self.subTest(stage=st["id"]): self.assertEqual(vpr.STAGES[st["id"]]["agent"], st["runBy"]) # ── 판정 ──────────────────────────────────────────────────────── def test_계약대로_적은_원장은_통과한다(self): """대조군. 이 자리가 통과하지 않으면 나머지는 아무것도 말하지 않는다.""" code, out = self._run(self._ledger()) self.assertEqual(0, code, out) self.assertIn("누가 돌렸는지 모르는 단계 0", out) def test_옛_판의_원장은_error_가_아니라_warn_이다(self): """`schemaVersion` 1 에는 그 칸이 없었다. 위조가 아니라 그때의 계약이다.""" code, out = self._run(self._ledger(schema=1, run_by=vpr.LEGACY_RUNBY)) self.assertEqual(0, code, out) self.assertIn("옛 판의 원장이라 누가 돌렸는지 적혀 있지 않다", out) self.assertNotIn("단계를 맡은 에이전트가 계약과 다르다", out) def test_옛_판이어도_초록으로_보이지_않는다(self): """warn 은 통과가 아니다. 요약 줄이 그 수를 따로 센다.""" code, out = self._run(self._ledger(schema=1, run_by=vpr.LEGACY_RUNBY)) self.assertIn("누가 돌렸는지 모르는 단계 7", out) code, _ = self._run(self._ledger(schema=1, run_by=vpr.LEGACY_RUNBY), "--strict") self.assertEqual(1, code) def test_새_판에서_상수를_적으면_error_다(self): """유예는 옛 원장의 것이다. 지금 판으로 열고 상수를 적는 것은 다른 일이다.""" code, out = self._run(self._ledger(schema=2, run_by=vpr.LEGACY_RUNBY)) self.assertEqual(1, code, out) self.assertIn("단계를 맡은 에이전트가 계약과 다르다", out) def test_다른_관리_에이전트를_적어도_error_다(self): """실재하는 이름이라고 맞는 것은 아니다. 단계마다 맡은 역할이 다르다.""" code, out = self._run(self._ledger(schema=2, run_by="fact-reviewer")) self.assertEqual(1, code, out) self.assertIn("단계를 맡은 에이전트가 계약과 다르다", out) # ── 원장을 여는 도구가 계약값을 지우지 않는가 ─────────────────── def test_런을_열면_계약이_적힌다(self): led = os.path.join(self.dir, "opened", "run.json") p = _cli("open", led, "--project", "demo", "--record", "docs/demo/x.md") self.assertEqual(0, p.returncode, p.stderr) run = json.load(open(led, encoding="utf-8")) self.assertEqual(vpr.AGENT_RUNBY_SCHEMA, run["schemaVersion"]) for st in run["stages"]: self.assertEqual(vpr.STAGES[st["id"]]["agent"], st["runBy"]) def test_단계를_열어도_계약값이_남는다(self): """`begin` 의 기본값이 계약값을 덮어쓰면 원장은 다시 누가 돌렸는지 잃는다.""" led = os.path.join(self.dir, "begun", "run.json") _cli("open", led, "--project", "demo", "--record", "docs/demo/x.md") p = _cli("begin", led, "--stage", "S3") self.assertEqual(0, p.returncode, p.stderr) st = next(s for s in json.load(open(led, encoding="utf-8"))["stages"] if s["id"] == "S3") self.assertEqual(vpr.STAGES["S3"]["agent"], st["runBy"]) def test_사람이_지목하면_그것을_적는다(self): """계약과 다르면 검사기가 잡는다. 도구가 값을 막지는 않는다 — 거짓말은 원장에 남아야 한다.""" led = os.path.join(self.dir, "named", "run.json") _cli("open", led, "--project", "demo", "--record", "docs/demo/x.md") _cli("begin", led, "--stage", "S3", "--runby", "fact-reviewer") st = next(s for s in json.load(open(led, encoding="utf-8"))["stages"] if s["id"] == "S3") self.assertEqual("fact-reviewer", st["runBy"]) if __name__ == "__main__": unittest.main()