674 lines
29 KiB
Python
674 lines
29 KiB
Python
# /// script
|
|
# requires-python = ">=3.10"
|
|
# dependencies = []
|
|
# ///
|
|
"""Unit tests for the spire:forks helper.
|
|
|
|
Run from the repo root: `uv run plugins/spire/skills/forks/tests/test_forks.py`.
|
|
Imports forks.py by path. Everything generated (stub binary, state dirs, settings files)
|
|
lives under `tempfile`; nothing touches ~/.claude or the repo's own state.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
FORKS_PY = os.path.join(os.path.dirname(HERE), "forks.py")
|
|
|
|
|
|
def _load_forks():
|
|
spec = importlib.util.spec_from_file_location("forks", FORKS_PY)
|
|
mod = importlib.util.module_from_spec(spec)
|
|
sys.modules["forks"] = mod
|
|
spec.loader.exec_module(mod)
|
|
return mod
|
|
|
|
|
|
forks = _load_forks()
|
|
IS_WINDOWS = os.name == "nt"
|
|
STUB_VERSION = "2.1.999"
|
|
|
|
|
|
def make_stub_binary(d: str, version: str = STUB_VERSION) -> str:
|
|
"""A fake `claude` that answers --version. Generated at runtime (A18), never committed."""
|
|
if IS_WINDOWS:
|
|
p = os.path.join(d, "claude-stub.cmd")
|
|
with open(p, "w", newline="\r\n") as f:
|
|
f.write("@echo off\necho %s (Claude Code stub)\n" % version)
|
|
else:
|
|
p = os.path.join(d, "claude-stub")
|
|
with open(p, "w", newline="\n") as f:
|
|
f.write("#!/bin/sh\necho '%s (Claude Code stub)'\n" % version)
|
|
os.chmod(p, 0o755)
|
|
return p
|
|
|
|
|
|
def call(argv):
|
|
"""Run forks.main in-process. Returns (exit code, parsed JSON, raw stdout)."""
|
|
out, err = io.StringIO(), io.StringIO()
|
|
with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
|
|
code = forks.main(list(argv))
|
|
raw = out.getvalue()
|
|
return code, json.loads(raw), raw
|
|
|
|
|
|
def child_env():
|
|
env = dict(os.environ)
|
|
env.pop(forks.CHILD_MARKER, None)
|
|
return env
|
|
|
|
|
|
class Base(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.mkdtemp(prefix="forks-test-")
|
|
self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True)
|
|
env_patch = mock.patch.dict(os.environ)
|
|
env_patch.start()
|
|
self.addCleanup(env_patch.stop)
|
|
os.environ.pop(forks.CHILD_MARKER, None)
|
|
|
|
def assert_one_json_line(self, raw: str):
|
|
self.assertTrue(raw.endswith("\n"), repr(raw[-40:]))
|
|
self.assertEqual(raw.count("\n"), 1, raw)
|
|
obj = json.loads(raw)
|
|
self.assertIsInstance(obj, dict)
|
|
return obj
|
|
|
|
|
|
# --------------------------------------------------------------------------------------
|
|
class PathTests(Base):
|
|
def test_windows_forms(self):
|
|
cwd = "G:\\dev\\proj"
|
|
cases = {
|
|
"docs\\a.md": "G:\\dev\\proj\\docs\\a.md",
|
|
"docs/a.md": "G:\\dev\\proj\\docs\\a.md",
|
|
"G:\\x\\y.md": "G:\\x\\y.md",
|
|
"G:/x/y.md": "G:\\x\\y.md",
|
|
"g:/x/y.md": "G:\\x\\y.md",
|
|
"/g/x/y.md": "G:\\x\\y.md",
|
|
"/c/Users/me": "C:\\Users\\me",
|
|
"/c": "C:\\",
|
|
"/c/": "C:\\",
|
|
"../other/./z.md": "G:\\dev\\other\\z.md",
|
|
"\\rooted\\a": "G:\\rooted\\a",
|
|
"/tmp/x": "G:\\tmp\\x", # not a drive form: rooted on the cwd's drive
|
|
" G:/x/y.md ": "G:\\x\\y.md",
|
|
}
|
|
for src, want in cases.items():
|
|
with self.subTest(src=src):
|
|
self.assertEqual(forks.norm_path(src, cwd=cwd, windows=True), want)
|
|
|
|
def test_windows_cwd_in_gitbash_form(self):
|
|
self.assertEqual(forks.norm_path("a.md", cwd="/g/dev/proj", windows=True), "G:\\dev\\proj\\a.md")
|
|
|
|
def test_posix_forms(self):
|
|
cwd = "/home/u/proj"
|
|
self.assertEqual(forks.norm_path("a/b.md", cwd=cwd, windows=False), "/home/u/proj/a/b.md")
|
|
self.assertEqual(forks.norm_path("/abs/x.md", cwd=cwd, windows=False), "/abs/x.md")
|
|
self.assertEqual(forks.norm_path("../z", cwd=cwd, windows=False), "/home/u/z")
|
|
|
|
def test_native_relative(self):
|
|
self.assertEqual(os.path.normcase(forks.norm_path("x/y.md")),
|
|
os.path.normcase(os.path.abspath(os.path.join("x", "y.md"))))
|
|
|
|
def test_empty_refused(self):
|
|
with self.assertRaises(forks.ForksError):
|
|
forks.norm_path(" ")
|
|
|
|
@unittest.skipUnless(IS_WINDOWS, "Windows path forms")
|
|
def test_state_dir_windows_forms(self):
|
|
want = "G:\\w\\reports\\2026-10-16-08-x-run-forks"
|
|
for src in ("G:\\w\\reports\\2026-10-16-08-x-run.md", "G:/w/reports/2026-10-16-08-x-run.md",
|
|
"/g/w/reports/2026-10-16-08-x-run.md"):
|
|
with self.subTest(src=src):
|
|
self.assertEqual(forks.state_dir_for(src), want)
|
|
|
|
def test_state_dir_relative(self):
|
|
got = forks.state_dir_for(os.path.join("reports", "2026-10-16-08-x-run.md"))
|
|
self.assertEqual(os.path.normcase(got),
|
|
os.path.normcase(os.path.abspath(os.path.join("reports", "2026-10-16-08-x-run-forks"))))
|
|
|
|
def test_state_dir_needs_md(self):
|
|
with self.assertRaises(forks.ForksError) as cm:
|
|
forks.state_dir_for(os.path.join(self.tmp, "run.txt"))
|
|
self.assertEqual(cm.exception.code, 2)
|
|
|
|
|
|
# --------------------------------------------------------------------------------------
|
|
class ScrubTests(Base):
|
|
def test_scrubbed_env(self):
|
|
venv = os.path.join(self.tmp, "venv")
|
|
keep_dir = os.path.join(self.tmp, "bin")
|
|
host = {
|
|
"CLAUDE_CODE_MESSAGING_SOCKET": "s", "CLAUDE_CODE_SESSION_ID": "x", "CLAUDE_EFFORT": "high",
|
|
"ANTHROPIC_API_KEY": "k", "ANTHROPIC_DEFAULT_OPUS_MODEL": "m", "UV_CACHE_DIR": "c",
|
|
"VIRTUAL_ENV": venv, "CLAUDECODE": "1", "CLAUDE_CODE_CHILD_SESSION": "1", "AI_AGENT": "a",
|
|
"CLAUDE_PID": "1", "ANTHROPIC_BASE_URL": "u", "HOME": "h",
|
|
"CLAUDE_CODE_PROMPT_CACHE_TTL": "5m", "CLAUDE_CODE_EXECPATH": "old",
|
|
"PATH": os.pathsep.join([os.path.join(venv, "Scripts"), keep_dir]),
|
|
}
|
|
env = forks.scrubbed_env("BIN", environ=host)
|
|
for k in ("CLAUDE_CODE_MESSAGING_SOCKET", "CLAUDE_CODE_SESSION_ID", "CLAUDE_EFFORT",
|
|
"ANTHROPIC_API_KEY", "ANTHROPIC_DEFAULT_OPUS_MODEL", "UV_CACHE_DIR", "VIRTUAL_ENV"):
|
|
self.assertNotIn(k, env)
|
|
for k in ("CLAUDECODE", "CLAUDE_CODE_CHILD_SESSION", "AI_AGENT", "CLAUDE_PID",
|
|
"ANTHROPIC_BASE_URL", "HOME"):
|
|
self.assertEqual(env[k], host[k])
|
|
self.assertEqual(env["CLAUDE_CODE_PROMPT_CACHE_TTL"], "1h")
|
|
self.assertEqual(env["DISABLE_AUTOUPDATER"], "1")
|
|
self.assertEqual(env["SPIRE_FORKS_CHILD"], "1")
|
|
self.assertEqual(env["CLAUDE_CODE_EXECPATH"], "BIN")
|
|
self.assertEqual(env["PATH"], keep_dir)
|
|
|
|
def test_unknown(self):
|
|
host = {"CLAUDECODE": "1", "CLAUDE_EFFORT": "x", "CLAUDE_CODE_NEW_THING": "1",
|
|
"ANTHROPIC_BASE_URL": "u", "OTHER": "1", "CLAUDE_CODE_EXECPATH": "p"}
|
|
self.assertEqual(forks.scrub_unknown(host), ["ANTHROPIC_BASE_URL", "CLAUDE_CODE_NEW_THING"])
|
|
|
|
|
|
# --------------------------------------------------------------------------------------
|
|
class LockTests(Base):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.state = os.path.join(self.tmp, "st")
|
|
os.makedirs(self.state)
|
|
|
|
def test_contention_times_out(self):
|
|
with forks.RegistryLock(self.state):
|
|
t0 = time.monotonic()
|
|
with self.assertRaises(forks.ForksError) as cm:
|
|
forks.RegistryLock(self.state, timeout=0.3).acquire()
|
|
self.assertGreaterEqual(time.monotonic() - t0, 0.25)
|
|
self.assertEqual(cm.exception.payload["error"], "registry locked")
|
|
with forks.RegistryLock(self.state, timeout=0.3):
|
|
pass
|
|
self.assertFalse(os.path.exists(forks.lock_path(self.state)))
|
|
|
|
def test_waiter_gets_lock_after_release(self):
|
|
l1 = forks.RegistryLock(self.state).acquire()
|
|
threading.Timer(0.2, l1.release).start()
|
|
t0 = time.monotonic()
|
|
with forks.RegistryLock(self.state, timeout=5) as l2:
|
|
self.assertGreaterEqual(time.monotonic() - t0, 0.15)
|
|
self.assertFalse(l2.broke_stale)
|
|
|
|
def test_contention_across_processes(self):
|
|
code = ("import importlib.util,sys;"
|
|
"s=importlib.util.spec_from_file_location('forks',sys.argv[1]);m=importlib.util.module_from_spec(s);"
|
|
"s.loader.exec_module(m)\n"
|
|
"try:\n m.RegistryLock(sys.argv[2],timeout=0.3).acquire(); sys.exit(0)\n"
|
|
"except m.ForksError: sys.exit(7)\n")
|
|
with forks.RegistryLock(self.state):
|
|
r = subprocess.run([sys.executable, "-c", code, FORKS_PY, self.state], capture_output=True)
|
|
self.assertEqual(r.returncode, 7, r.stderr)
|
|
|
|
def test_stale_lock_broken(self):
|
|
with open(forks.lock_path(self.state), "w") as f:
|
|
json.dump({"pid": 1, "at": time.time() - (forks.LOCK_STALE_S + 1), "token": "old"}, f)
|
|
with forks.RegistryLock(self.state, timeout=2) as lk:
|
|
self.assertTrue(lk.broke_stale)
|
|
with open(forks.lock_path(self.state)) as f:
|
|
self.assertEqual(json.load(f)["token"], lk.token)
|
|
|
|
def test_garbage_stale_lock_broken_by_mtime(self):
|
|
p = forks.lock_path(self.state)
|
|
with open(p, "w") as f:
|
|
f.write("")
|
|
old = time.time() - (forks.LOCK_STALE_S + 5)
|
|
os.utime(p, (old, old))
|
|
with forks.RegistryLock(self.state, timeout=2) as lk:
|
|
self.assertTrue(lk.broke_stale)
|
|
|
|
def test_fresh_lock_not_broken(self):
|
|
with open(forks.lock_path(self.state), "w") as f:
|
|
json.dump({"pid": 1, "at": time.time() - 5, "token": "other"}, f)
|
|
with self.assertRaises(forks.ForksError):
|
|
forks.RegistryLock(self.state, timeout=0.3).acquire()
|
|
|
|
def test_stale_under_injected_clock_and_safe_release(self):
|
|
l1 = forks.RegistryLock(self.state).acquire()
|
|
real = time.time()
|
|
with mock.patch.object(forks, "_now", lambda: real + forks.LOCK_STALE_S + 1):
|
|
l2 = forks.RegistryLock(self.state, timeout=2).acquire()
|
|
self.assertTrue(l2.broke_stale)
|
|
l1.release() # must not remove l2's lock
|
|
self.assertTrue(os.path.exists(forks.lock_path(self.state)))
|
|
l2.release()
|
|
self.assertFalse(os.path.exists(forks.lock_path(self.state)))
|
|
|
|
|
|
# --------------------------------------------------------------------------------------
|
|
class AtomicTests(Base):
|
|
def test_replace_while_reader_holds_target(self):
|
|
target = os.path.join(self.tmp, "registry.json")
|
|
forks.write_json_atomic(target, {"v": 1})
|
|
opened, done = threading.Event(), threading.Event()
|
|
|
|
def reader():
|
|
with open(target, "rb") as f:
|
|
opened.set()
|
|
time.sleep(0.4)
|
|
f.read()
|
|
done.set()
|
|
th = threading.Thread(target=reader)
|
|
th.start()
|
|
opened.wait(2)
|
|
attempts = forks.write_json_atomic(target, {"v": 2})
|
|
th.join()
|
|
self.assertEqual(forks.read_json(target), {"v": 2})
|
|
if IS_WINDOWS:
|
|
self.assertGreater(attempts, 1, "replace should have hit a sharing violation and retried")
|
|
self.assertEqual([n for n in os.listdir(self.tmp) if n.endswith(".tmp")], [])
|
|
|
|
def test_read_json_missing_is_none(self):
|
|
self.assertIsNone(forks.read_json(os.path.join(self.tmp, "nope.json")))
|
|
|
|
def test_registry_roundtrip_and_schema(self):
|
|
st = os.path.join(self.tmp, "st")
|
|
forks.ensure_state(st)
|
|
with forks.RegistryLock(st):
|
|
forks.write_registry(st, forks.new_registry())
|
|
reg = forks.read_registry(st)
|
|
self.assertEqual(reg["schema"], 1)
|
|
with open(forks.registry_path(st), "rb") as f:
|
|
self.assertNotIn(b"\r\n", f.read())
|
|
forks.write_json_atomic(forks.registry_path(st), {"schema": 99})
|
|
with self.assertRaises(forks.ForksError):
|
|
forks.read_registry(st)
|
|
|
|
def test_ledger_append(self):
|
|
st = os.path.join(self.tmp, "st")
|
|
forks.ensure_state(st)
|
|
with forks.RegistryLock(st):
|
|
forks.ledger_append(st, {"job": "j0001"})
|
|
forks.ledger_append(st, {"job": "j0002"})
|
|
with open(forks.ledger_path(st), "rb") as f:
|
|
lines = f.read().decode().splitlines()
|
|
self.assertEqual([json.loads(x)["job"] for x in lines], ["j0001", "j0002"])
|
|
|
|
|
|
# --------------------------------------------------------------------------------------
|
|
class PidTests(Base):
|
|
def test_own_pid(self):
|
|
rec = forks.pid_record(os.getpid())
|
|
self.assertTrue(forks.record_alive(rec))
|
|
if rec["ctime"] is not None:
|
|
self.assertFalse(forks.pid_alive(os.getpid(), rec["ctime"] + 1))
|
|
|
|
def test_dead_pid(self):
|
|
p = subprocess.Popen([sys.executable, "-c", "pass"])
|
|
rec = forks.pid_record(p.pid)
|
|
p.wait()
|
|
self.assertFalse(forks.record_alive(rec))
|
|
self.assertFalse(forks.pid_alive(0))
|
|
|
|
def test_supervisor_interpreter(self):
|
|
self.assertTrue(os.path.isfile(forks.supervisor_interpreter()))
|
|
|
|
|
|
# --------------------------------------------------------------------------------------
|
|
class InitTests(Base):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.cwd = os.path.join(self.tmp, "proj")
|
|
os.makedirs(os.path.join(self.cwd, ".claude", "agents"))
|
|
os.makedirs(os.path.join(self.tmp, "reports"))
|
|
self.run_log = os.path.join(self.tmp, "reports", "2026-10-16-08-t-run.md")
|
|
self.state = forks.state_dir_for(self.run_log)
|
|
self.user_settings = os.path.join(self.tmp, "home", ".claude", "settings.json")
|
|
p = mock.patch.object(forks, "USER_SETTINGS_PATH", self.user_settings)
|
|
p.start()
|
|
self.addCleanup(p.stop)
|
|
self.stub = make_stub_binary(self.tmp)
|
|
|
|
def init(self, *extra, model="claude-opus-5-5[1m]"):
|
|
argv = ["init", "--run", self.run_log, "--model", model, "--cwd", self.cwd,
|
|
"--binary", self.stub, *extra]
|
|
return call(argv)
|
|
|
|
def write_settings(self, path, models):
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
with open(path, "w") as f:
|
|
json.dump({"availableModels": models}, f)
|
|
|
|
def add_fake_base(self):
|
|
with forks.RegistryLock(self.state):
|
|
reg = forks.read_registry(self.state)
|
|
reg["bases"]["B1"] = {"status": "active", "session": "00000000-0000-0000-0000-000000000000"}
|
|
forks.write_registry(self.state, reg)
|
|
|
|
def reg_bytes(self):
|
|
with open(forks.registry_path(self.state), "rb") as f:
|
|
return f.read()
|
|
|
|
# --- basic ---
|
|
def test_init_creates_state_and_config(self):
|
|
code, out, raw = self.init()
|
|
self.assertEqual(code, 0, out)
|
|
self.assert_one_json_line(raw)
|
|
self.assertTrue(out["created"])
|
|
st = self.state
|
|
for d in forks.STATE_SUBDIRS:
|
|
self.assertTrue(os.path.isdir(os.path.join(st, d)), d)
|
|
with open(os.path.join(st, ".gitignore"), "rb") as f:
|
|
self.assertEqual(f.read(), b"raw/\n*.lock\n")
|
|
self.assertEqual(forks.read_json(forks.mcp_empty_path(st)), {"mcpServers": {}})
|
|
self.assertFalse(os.path.exists(forks.lock_path(st)))
|
|
reg = forks.read_registry(st)
|
|
self.assertEqual(reg["schema"], 1)
|
|
self.assertEqual(reg["bases"], {})
|
|
self.assertEqual(reg["jobs"], {})
|
|
cfg = reg["config"]
|
|
la = cfg["launch"]
|
|
self.assertEqual(la["model"], "claude-opus-5-5[1m]")
|
|
self.assertEqual(la["effort"], "high")
|
|
self.assertEqual(la["permission_mode"], "dontAsk")
|
|
self.assertEqual(la["tools"], forks.DEFAULT_TOOLS.split(","))
|
|
self.assertEqual(la["allowed_tools"], la["tools"])
|
|
self.assertEqual(la["disallowed_tools"], forks.DEFAULT_DISALLOWED)
|
|
self.assertEqual(la["mcp"], {"strict": True, "config": forks.mcp_empty_path(st)})
|
|
self.assertEqual(la["cwd"], forks.norm_path(self.cwd))
|
|
self.assertIsNone(la["agent"])
|
|
self.assertEqual(la["env"]["pinned"], {"CLAUDE_CODE_PROMPT_CACHE_TTL": "1h", "DISABLE_AUTOUPDATER": "1"})
|
|
b = la["binary"]
|
|
self.assertEqual(b["path"], forks.norm_path(self.stub))
|
|
self.assertTrue(os.path.isabs(b["path"]))
|
|
self.assertEqual(b["version"], STUB_VERSION)
|
|
self.assertEqual(b["size"], os.path.getsize(self.stub))
|
|
pre = la["append_system_prompt"]
|
|
self.assertEqual(pre["path"], forks.preamble_path())
|
|
self.assertEqual(pre["sha256"], forks._sha256_file(forks.preamble_path()))
|
|
self.assertEqual(cfg["fingerprint"], forks.fingerprint_hash(la))
|
|
self.assertEqual(len(cfg["fingerprint"]), 64)
|
|
self.assertEqual(cfg["parallel_cap"], 2)
|
|
self.assertEqual(cfg["orphan_after"], 1200)
|
|
self.assertEqual(cfg["pings"], "on")
|
|
self.assertEqual(cfg["ttl_s"], 3600)
|
|
self.assertEqual(cfg["max_minutes"], 90)
|
|
self.assertEqual(cfg["stall_after_s"], 900)
|
|
self.assertEqual(cfg["thresholds"], forks.THRESHOLD_DEFAULTS)
|
|
self.assertEqual(out["config"], cfg)
|
|
|
|
def test_fingerprint_canonical(self):
|
|
a = {"b": [1, 2], "a": {"y": 1, "x": None}}
|
|
b = {"a": {"x": None, "y": 1}, "b": [1, 2]}
|
|
self.assertEqual(forks.fingerprint_hash(a), forks.fingerprint_hash(b))
|
|
self.assertEqual(forks.canonical_json(a), '{"a":{"x":null,"y":1},"b":[1,2]}')
|
|
|
|
def test_list_flags(self):
|
|
code, out, _ = self.init("--tools", "Read,Bash", "--allowed-tools", "Read", "Bash(git status:*)",
|
|
"--disallowed-tools", "Bash(rm:*),Bash(a,b)")
|
|
self.assertEqual(code, 0, out)
|
|
la = out["config"]["launch"]
|
|
self.assertEqual(la["tools"], ["Read", "Bash"])
|
|
self.assertEqual(la["allowed_tools"], ["Read", "Bash(git status:*)"])
|
|
self.assertEqual(la["disallowed_tools"], ["Bash(rm:*)", "Bash(a,b)"])
|
|
|
|
# --- idempotency ---
|
|
def test_idempotent(self):
|
|
self.assertEqual(self.init()[0], 0)
|
|
before = self.reg_bytes()
|
|
code, out, _ = self.init()
|
|
self.assertEqual(code, 0, out)
|
|
self.assertFalse(out["created"])
|
|
self.assertFalse(out["changed"])
|
|
self.assertEqual(self.reg_bytes(), before)
|
|
|
|
def test_changed_values_before_base(self):
|
|
_, first, _ = self.init()
|
|
code, out, _ = self.init("--effort", "medium", "--orphan-after", "1500")
|
|
self.assertEqual(code, 0, out)
|
|
self.assertTrue(out["changed"])
|
|
self.assertIn("launch.effort", out["changed_fields"])
|
|
self.assertIn("orphan_after", out["changed_fields"])
|
|
self.assertIn("fingerprint", out["changed_fields"])
|
|
cfg = forks.read_registry(self.state)["config"]
|
|
self.assertEqual(cfg["launch"]["effort"], "medium")
|
|
self.assertEqual(cfg["orphan_after"], 1500)
|
|
self.assertNotEqual(cfg["fingerprint"], first["config"]["fingerprint"])
|
|
|
|
def test_changed_values_after_base_refused(self):
|
|
self.assertEqual(self.init()[0], 0)
|
|
self.add_fake_base()
|
|
before = self.reg_bytes()
|
|
code, out, raw = self.init("--effort", "medium")
|
|
self.assertEqual(code, 2)
|
|
self.assert_one_json_line(raw)
|
|
self.assertIn("base exists", out["error"])
|
|
self.assertEqual(out["bases"], ["B1"])
|
|
self.assertIn("launch.effort", out["fields"])
|
|
self.assertEqual(self.reg_bytes(), before)
|
|
self.assertFalse(os.path.exists(forks.lock_path(self.state)))
|
|
# identical values stay fine with a base
|
|
code, out, _ = self.init()
|
|
self.assertEqual(code, 0, out)
|
|
self.assertFalse(out["changed"])
|
|
|
|
# --- refusals ---
|
|
def assert_refused_no_state(self, *extra, model="claude-opus-5-5[1m]"):
|
|
code, out, raw = self.init(*extra, model=model)
|
|
self.assertEqual(code, 2, out)
|
|
self.assert_one_json_line(raw)
|
|
self.assertIn("error", out)
|
|
self.assertFalse(os.path.exists(self.state), "a refused init must not write")
|
|
return out
|
|
|
|
def test_bad_model_ids(self):
|
|
for m in ("opus", "sonnet", "claude-opus-5-5[200k]", "Claude-Opus-5-5", "claude-opus 5"):
|
|
with self.subTest(model=m):
|
|
self.assert_refused_no_state(model=m)
|
|
|
|
def test_plan_mode_refused(self):
|
|
out = self.assert_refused_no_state("--permission-mode", "plan")
|
|
self.assertEqual(out["error"], "permission mode refused")
|
|
|
|
def test_unknown_mode_and_effort_refused(self):
|
|
self.assert_refused_no_state("--permission-mode", "yolo")
|
|
self.assert_refused_no_state("--effort", "turbo")
|
|
|
|
def test_missing_binary_refused(self):
|
|
out = self.assert_refused_no_state("--binary", os.path.join(self.tmp, "nope.exe"))
|
|
self.assertEqual(out["error"], "claude binary not found")
|
|
|
|
def test_agent(self):
|
|
self.assert_refused_no_state("--agent", "ghost")
|
|
self.assert_refused_no_state("--agent", "../x")
|
|
ap = os.path.join(self.cwd, ".claude", "agents", "impl.md")
|
|
with open(ap, "w") as f:
|
|
f.write("---\nname: impl\n---\nbody\n")
|
|
code, out, _ = self.init("--agent", "impl")
|
|
self.assertEqual(code, 0, out)
|
|
ag = out["config"]["launch"]["agent"]
|
|
self.assertEqual(ag["name"], "impl")
|
|
self.assertEqual(ag["sha256"], forks._sha256_file(ap))
|
|
|
|
# --- C8 admission ---
|
|
def proj(self):
|
|
return os.path.join(self.cwd, ".claude", "settings.json")
|
|
|
|
def local(self):
|
|
return os.path.join(self.cwd, ".claude", "settings.local.json")
|
|
|
|
def test_admission_no_list_admits(self):
|
|
self.assertTrue(forks.model_admission("claude-whatever-9", self.cwd)["admitted"])
|
|
|
|
def test_admission_exact(self):
|
|
self.write_settings(self.user_settings, ["claude-opus-5-5[1m]"])
|
|
self.assertTrue(forks.model_admission("claude-opus-5-5[1m]", self.cwd)["admitted"])
|
|
# the bare id is not the [1m] entry and `opus` is not listed
|
|
self.assertFalse(forks.model_admission("claude-opus-5-5", self.cwd)["admitted"])
|
|
|
|
def test_admission_strips_1m(self):
|
|
self.write_settings(self.proj(), ["claude-opus-5-5"])
|
|
adm = forks.model_admission("claude-opus-5-5[1m]", self.cwd)
|
|
self.assertTrue(adm["admitted"])
|
|
self.assertIn("without [1m]", adm["via"])
|
|
|
|
def test_admission_family_alias(self):
|
|
self.write_settings(self.local(), ["sonnet"])
|
|
adm = forks.model_admission("claude-sonnet-5-5", self.cwd)
|
|
self.assertTrue(adm["admitted"])
|
|
self.assertIn("family alias", adm["via"])
|
|
|
|
def test_admission_any_list_admits(self):
|
|
self.write_settings(self.user_settings, ["claude-opus-5-5"])
|
|
self.write_settings(self.local(), ["haiku"])
|
|
self.assertTrue(forks.model_admission("claude-haiku-5-5", self.cwd)["admitted"])
|
|
|
|
def test_admission_unlisted_refused_by_init(self):
|
|
self.write_settings(self.user_settings, ["claude-opus-5-5"])
|
|
self.write_settings(self.local(), ["opus"])
|
|
with open(self.local(), "rb") as f:
|
|
local_before = f.read()
|
|
out = self.assert_refused_no_state(model="claude-sonnet-5-5")
|
|
self.assertEqual(out["error"], "model not admitted by availableModels")
|
|
self.assertEqual(out["model"], "claude-sonnet-5-5")
|
|
self.assertEqual(sorted(out["files"]), sorted([self.user_settings, self.local()]))
|
|
self.assertIn("must add", out["message"])
|
|
self.assertIn(self.local(), out["message"])
|
|
with open(self.local(), "rb") as f:
|
|
self.assertEqual(f.read(), local_before, "settings must never be written")
|
|
self.assertFalse(os.path.exists(self.proj()))
|
|
|
|
def test_admission_admitted_by_init(self):
|
|
self.write_settings(self.proj(), ["opus", "sonnet"])
|
|
code, out, _ = self.init(model="claude-sonnet-5-5")
|
|
self.assertEqual(code, 0, out)
|
|
self.assertIn("family alias", out["config"]["model_admission"])
|
|
|
|
|
|
# --------------------------------------------------------------------------------------
|
|
class CliTests(Base):
|
|
def test_not_implemented(self):
|
|
run_log = os.path.join(self.tmp, "r-run.md")
|
|
cases = {
|
|
("train", "--base", "B1", "--brief", "b.md"): {"block": "1.2"},
|
|
("fork", "--base", "B1", "--block", "1.1", "--brief", "b.md"): {"block": "1.2"},
|
|
("continue", "--job", "j0001", "--brief", "b.md"): {"block": "1.2"},
|
|
("wait", "--all"): {"block": "1.2"},
|
|
("list",): {"block": "1.2"},
|
|
("expire", "--all"): {"block": "1.2"},
|
|
("kill", "--job", "j0001"): {"block": "1.2"},
|
|
("_supervise", "--job", "j0001"): {"block": "1.2"},
|
|
("rebase",): {"phase": 2},
|
|
("usage", "--phase", "1"): {"phase": 2},
|
|
("tick",): {"phase": 2},
|
|
}
|
|
for argv, want in cases.items():
|
|
with self.subTest(cmd=argv[0]):
|
|
code, out, raw = call([argv[0], "--run", run_log, *argv[1:]])
|
|
self.assertEqual(code, 2)
|
|
self.assert_one_json_line(raw)
|
|
self.assertEqual(out["error"], "not implemented")
|
|
for k, v in want.items():
|
|
self.assertEqual(out[k], v)
|
|
self.assertFalse(os.path.exists(run_log[:-3] + "-forks"))
|
|
|
|
def test_unknown_command_and_missing_run(self):
|
|
for argv in (["bogus"], [], ["list"], ["init", "--run", "x.md"],
|
|
["list", "--run", "x.md", "--bogus"]):
|
|
with self.subTest(argv=argv):
|
|
code, out, raw = call(argv)
|
|
self.assertEqual(code, 2)
|
|
self.assert_one_json_line(raw)
|
|
self.assertEqual(out["error"], "usage")
|
|
|
|
def test_help_is_json(self):
|
|
code, out, raw = call(["--help"])
|
|
self.assertEqual(code, 0)
|
|
self.assert_one_json_line(raw)
|
|
self.assertIn("where", out["commands"])
|
|
self.assertNotIn("_supervise", out["commands"])
|
|
code, out, _ = call(["init", "--help"])
|
|
self.assertEqual(code, 0)
|
|
self.assertIn("--model", out["help"])
|
|
|
|
def test_child_refuses_launching(self):
|
|
os.environ[forks.CHILD_MARKER] = "1"
|
|
run_log = os.path.join(self.tmp, "r-run.md")
|
|
code, out, _ = call(["fork", "--run", run_log, "--base", "B1", "--block", "1", "--brief", "b"])
|
|
self.assertEqual(code, 2)
|
|
self.assertIn("inside a fork", out["error"])
|
|
code, out, _ = call(["list", "--run", run_log])
|
|
self.assertEqual(out["error"], "not implemented") # not refused as a launch
|
|
stub = make_stub_binary(self.tmp)
|
|
self.assertEqual(call(["where", "--binary", stub])[0], 0)
|
|
|
|
def test_internal_error_keeps_invariant(self):
|
|
def boom(args):
|
|
raise RuntimeError("kaboom")
|
|
with mock.patch.dict(forks.HANDLERS, {"list": boom}):
|
|
code, out, raw = call(["list", "--run", os.path.join(self.tmp, "r-run.md")])
|
|
self.assertEqual(code, 2)
|
|
self.assert_one_json_line(raw)
|
|
self.assertEqual(out["error"], "internal error")
|
|
|
|
def test_where_in_process_missing_binary(self):
|
|
code, out, raw = call(["where", "--binary", os.path.join(self.tmp, "nope.exe")])
|
|
self.assertEqual(code, 2)
|
|
self.assert_one_json_line(raw)
|
|
self.assertFalse(out["ok"])
|
|
self.assertIn("reason", out)
|
|
self.assertIn("error", out)
|
|
|
|
|
|
# --------------------------------------------------------------------------------------
|
|
@unittest.skipUnless(shutil.which("uv"), "uv not on PATH")
|
|
class SubprocessTests(Base):
|
|
def run_uv(self, *argv):
|
|
return subprocess.run(["uv", "run", FORKS_PY, *argv], cwd=self.tmp, env=child_env(),
|
|
capture_output=True, timeout=180)
|
|
|
|
def test_where_via_uv_stub(self):
|
|
stub = make_stub_binary(self.tmp)
|
|
r = self.run_uv("where", "--binary", stub)
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
|
out = self.assert_one_json_line(r.stdout.decode("utf-8"))
|
|
self.assertTrue(out["ok"])
|
|
self.assertEqual(out["spawn"], "cim-hidden" if IS_WINDOWS else "setsid")
|
|
self.assertEqual(out["binary"]["version"], STUB_VERSION)
|
|
self.assertEqual(out["binary"]["how"], "--binary")
|
|
self.assertEqual(os.path.normcase(out["script"]), os.path.normcase(os.path.abspath(FORKS_PY)))
|
|
for k in ("python", "uv", "git", "scrub"):
|
|
self.assertIn(k, out)
|
|
for k in ("drop", "keep", "pinned", "unknown"):
|
|
self.assertIn(k, out["scrub"])
|
|
self.assertIn("base_executable", out["python"])
|
|
self.assertIsNotNone(out["git"]["path"])
|
|
|
|
def test_where_via_uv_host_binary(self):
|
|
if not (os.environ.get("CLAUDE_CODE_EXECPATH") or shutil.which("claude")):
|
|
self.skipTest("no claude binary on this host")
|
|
r = self.run_uv("where")
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
|
self.assertTrue(self.assert_one_json_line(r.stdout.decode("utf-8"))["ok"])
|
|
|
|
def test_error_case_stdout_invariant(self):
|
|
run_log = os.path.join(self.tmp, "x-run.md")
|
|
r = self.run_uv("init", "--run", run_log, "--model", "opus", "--cwd", self.tmp)
|
|
self.assertEqual(r.returncode, 2, r.stderr)
|
|
out = self.assert_one_json_line(r.stdout.decode("utf-8"))
|
|
self.assertIn("error", out)
|
|
self.assertFalse(os.path.exists(os.path.join(self.tmp, "x-run-forks")))
|
|
|
|
def test_unknown_command_exit_2(self):
|
|
r = self.run_uv("frobnicate")
|
|
self.assertEqual(r.returncode, 2, r.stderr)
|
|
self.assertEqual(self.assert_one_json_line(r.stdout.decode("utf-8"))["error"], "usage")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|