✅ Gremlins MVP: unit and CLI tests with a fake claude
This commit is contained in:
@@ -0,0 +1,64 @@
|
||||
"""Fake `claude -p` for the CLI tests. $FAKE_CLAUDE_SCENARIO: ok (default), miss, api_error,
|
||||
noresult, two_models, no_cost, sleep. $FAKE_CLAUDE_LOG: one JSON line per call (argv, pid)."""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import uuid
|
||||
|
||||
TRAIN_MU = {"inputTokens": 10, "outputTokens": 20, "cacheReadInputTokens": 0,
|
||||
"cacheCreationInputTokens": 1000}
|
||||
TRAIN_COST, CALL_COST = 0.05, 0.02
|
||||
|
||||
|
||||
def opt(name):
|
||||
return sys.argv[sys.argv.index(name) + 1] if name in sys.argv else None
|
||||
|
||||
|
||||
def emit(ev):
|
||||
print(json.dumps(ev), flush=True)
|
||||
|
||||
|
||||
def main():
|
||||
prompt = sys.stdin.buffer.read().decode("utf-8")
|
||||
scen = os.environ.get("FAKE_CLAUDE_SCENARIO", "ok")
|
||||
if os.environ.get("FAKE_CLAUDE_LOG"):
|
||||
with open(os.environ["FAKE_CLAUDE_LOG"], "a", encoding="utf-8") as f:
|
||||
f.write(json.dumps({"argv": sys.argv, "pid": os.getpid(), "prompt_chars": len(prompt)}) + "\n")
|
||||
resume, model = opt("--resume"), opt("--model")
|
||||
# a plain resume (no fork, no --session-id) gets a new id, as `-p` may give one
|
||||
sid = opt("--session-id") or ("resumed-" + uuid.uuid4().hex if resume else str(uuid.uuid4()))
|
||||
read = 1000 if resume and scen != "miss" else 0
|
||||
created = 100 if resume else 1000
|
||||
usage = {"input_tokens": 10, "output_tokens": 20, "cache_creation_input_tokens": created,
|
||||
"cache_read_input_tokens": read,
|
||||
"cache_creation": {"ephemeral_1h_input_tokens": created, "ephemeral_5m_input_tokens": 0}}
|
||||
emit({"type": "system", "subtype": "init", "session_id": sid, "model": model})
|
||||
emit({"type": "assistant", "session_id": sid, "message": {"id": "m1", "usage": usage}})
|
||||
if scen == "sleep":
|
||||
time.sleep(60)
|
||||
if scen == "noresult":
|
||||
sys.exit(1)
|
||||
mu = {"inputTokens": 10, "outputTokens": 20, "cacheReadInputTokens": read,
|
||||
"cacheCreationInputTokens": created}
|
||||
cost = CALL_COST if resume else TRAIN_COST
|
||||
if resume: # the session's totals include the base's training call
|
||||
mu = {k: v + TRAIN_MU[k] for k, v in mu.items()}
|
||||
cost += TRAIN_COST
|
||||
res = {"type": "result", "subtype": "success", "is_error": False, "session_id": sid,
|
||||
"result": "done", "num_turns": 1, "usage": usage, "modelUsage": {model: mu},
|
||||
"total_cost_usd": cost}
|
||||
if scen == "two_models":
|
||||
res["modelUsage"]["claude-haiku-5-5"] = {"inputTokens": 5, "outputTokens": 1,
|
||||
"cacheReadInputTokens": 0, "cacheCreationInputTokens": 0}
|
||||
if scen == "no_cost":
|
||||
del res["total_cost_usd"]
|
||||
if scen == "api_error":
|
||||
res.update(subtype="error_during_execution", is_error=True, api_error_status=429,
|
||||
result="API Error: 429 rate limit")
|
||||
emit(res)
|
||||
sys.exit(1 if scen == "api_error" else 0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,57 @@
|
||||
import os
|
||||
import signal
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from grm import call # noqa: E402
|
||||
|
||||
CHILD = ("import json, sys, time\n"
|
||||
"p = sys.stdin.buffer.read().decode('utf-8')\n"
|
||||
"print(json.dumps({'type': 'system', 'subtype': 'init', 'session_id': p}), flush=True)\n"
|
||||
"if p == 'sleep': time.sleep(30)\n"
|
||||
"print(json.dumps({'type': 'result', 'session_id': p}), flush=True)\n")
|
||||
|
||||
|
||||
class RunClaudeTest(unittest.TestCase):
|
||||
def _run(self, d, prompt):
|
||||
pids = []
|
||||
raw = Path(d) / "j.jsonl"
|
||||
out = call.run_claude([sys.executable, "-c", CHILD], prompt, dict(os.environ), d,
|
||||
raw, pids.append)
|
||||
return out, raw, pids
|
||||
|
||||
def test_tee_and_parse(self):
|
||||
"""B1: prompt goes in on stdin; the stream is parsed and teed to the raw file."""
|
||||
with tempfile.TemporaryDirectory() as d:
|
||||
(ss, code, stopped), raw, pids = self._run(d, "héllo")
|
||||
self.assertEqual((code, stopped, len(pids)), (0, False, 1))
|
||||
self.assertEqual(ss.session_id, "héllo")
|
||||
self.assertIsNotNone(ss.result)
|
||||
self.assertEqual(len(raw.read_bytes().splitlines()), 2)
|
||||
|
||||
def test_sigterm_stops_child(self):
|
||||
"""B4: SIGTERM stops the child; stopped is set and the partial raw file is kept."""
|
||||
with tempfile.TemporaryDirectory() as d:
|
||||
timer = threading.Timer(1.5, signal.raise_signal, (signal.SIGTERM,))
|
||||
t0 = time.time()
|
||||
timer.start()
|
||||
try:
|
||||
(ss, code, stopped), raw, pids = self._run(d, "sleep")
|
||||
finally:
|
||||
timer.cancel()
|
||||
self.assertTrue(stopped)
|
||||
self.assertIsNotNone(code) # the child has exited
|
||||
self.assertLess(time.time() - t0, 20)
|
||||
self.assertIsNone(ss.result)
|
||||
self.assertEqual(ss.session_id, "sleep")
|
||||
self.assertIn(b'"init"', raw.read_bytes())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,211 @@
|
||||
import contextlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import signal
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
SKILL = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(SKILL))
|
||||
|
||||
from grm import cli # noqa: E402
|
||||
|
||||
FAKE = str(SKILL / "tests" / "fake_claude.py")
|
||||
MODEL = "claude-opus-5-5"
|
||||
|
||||
|
||||
class CliTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = Path(tempfile.mkdtemp())
|
||||
self.addCleanup(shutil.rmtree, self.tmp, True)
|
||||
repo = self.tmp / "repo"
|
||||
repo.mkdir()
|
||||
for a in (["init", "-q"], ["-c", "user.email=t@t", "-c", "user.name=t", "commit", "-q",
|
||||
"--allow-empty", "-m", "x"]):
|
||||
subprocess.run(["git", *a], cwd=repo, check=True, capture_output=True)
|
||||
self.run_md = str(self.tmp / "run.md")
|
||||
self.state = self.tmp / "run-gremlins"
|
||||
self.log = self.tmp / "fake.jsonl"
|
||||
self.brief = self.tmp / "brief.md"
|
||||
self.brief.write_text("do the thing\n", encoding="utf-8")
|
||||
env = mock.patch.dict(os.environ, {"FAKE_CLAUDE_LOG": str(self.log), "FAKE_CLAUDE_SCENARIO": "ok"})
|
||||
env.start()
|
||||
self.addCleanup(env.stop)
|
||||
code, out = self.g("init", "--run", self.run_md, "--model", MODEL, "--cwd", str(repo), "--binary", sys.executable)
|
||||
self.assertEqual(code, 0, out)
|
||||
run = json.loads((self.state / "run.json").read_text(encoding="utf-8"))
|
||||
run["argv"] = [sys.executable, FAKE] + run["argv"][1:]
|
||||
(self.state / "run.json").write_text(json.dumps(run), encoding="utf-8")
|
||||
self.stored = run["argv"]
|
||||
|
||||
def g(self, *args, scenario=None):
|
||||
if scenario:
|
||||
os.environ["FAKE_CLAUDE_SCENARIO"] = scenario
|
||||
buf = io.StringIO()
|
||||
with contextlib.redirect_stdout(buf):
|
||||
code = cli.main(list(args))
|
||||
os.environ["FAKE_CLAUDE_SCENARIO"] = "ok"
|
||||
return code, json.loads(buf.getvalue().splitlines()[-1])
|
||||
|
||||
def train(self, scenario=None):
|
||||
return self.g("train", "--run", self.run_md, "--base", "B1", "--brief", str(self.brief),
|
||||
scenario=scenario)
|
||||
|
||||
def spawn(self, job, base="B1", scenario=None):
|
||||
return self.g("spawn", "--run", self.run_md, "--base", base, "--job", job,
|
||||
"--brief", str(self.brief), scenario=scenario)
|
||||
|
||||
def calls(self):
|
||||
if not self.log.exists():
|
||||
return []
|
||||
return [json.loads(ln) for ln in self.log.read_text(encoding="utf-8").splitlines()]
|
||||
|
||||
def ledger(self):
|
||||
return [json.loads(ln) for ln in (self.state / "ledger.jsonl").read_text(encoding="utf-8").splitlines()]
|
||||
|
||||
def base_rec(self):
|
||||
p = self.state / "bases" / "B1.json"
|
||||
return json.loads(p.read_text(encoding="utf-8")) if p.exists() else None
|
||||
|
||||
def test_train_spawn_happy(self):
|
||||
"""B1, B2, B5: base keeps the --session-id; fork argv = stored argv + resume/fork; ledger."""
|
||||
code, out = self.train()
|
||||
self.assertEqual((code, out["status"]), (0, "ok"))
|
||||
sid = self.calls()[0]["argv"][-1]
|
||||
# the fake logs its sys.argv: the stored argv without the interpreter
|
||||
self.assertEqual(self.calls()[0]["argv"], self.stored[1:] + ["--session-id", sid])
|
||||
self.assertEqual(self.base_rec()["session_id"], sid)
|
||||
code, out = self.spawn("1.1")
|
||||
self.assertEqual((code, out["status"], out["cache"]), (0, "ok", "hit"))
|
||||
argv = self.calls()[1]["argv"]
|
||||
n = len(self.stored) - 1
|
||||
self.assertEqual(argv[:n], self.stored[1:])
|
||||
self.assertEqual(argv[n:n + 3], ["--resume", sid, "--fork-session"])
|
||||
line = self.ledger()[-1]
|
||||
self.assertEqual((line["model"], line["job"], line["cache"]), (MODEL, "1.1", "hit"))
|
||||
for k in ("input", "output", "cache_creation", "cache_read"):
|
||||
self.assertIsInstance(line["tokens"][k], int)
|
||||
self.assertAlmostEqual(line["cost_delta_usd"],
|
||||
line["cost_cumulative_usd"] - self.base_rec()["cost_cumulative_usd"], places=6)
|
||||
self.assertAlmostEqual(line["cost_delta_usd"], 0.02, places=6)
|
||||
|
||||
def test_missing_binary(self):
|
||||
"""B1: a missing binary is refused before launch; no job file."""
|
||||
run = json.loads((self.state / "run.json").read_text(encoding="utf-8"))
|
||||
run["argv"][0] = str(self.tmp / "nope.exe")
|
||||
(self.state / "run.json").write_text(json.dumps(run), encoding="utf-8")
|
||||
code, out = self.train()
|
||||
self.assertEqual(code, 2)
|
||||
self.assertIn("binary", out["error"])
|
||||
self.assertEqual(list((self.state / "jobs").iterdir()), [])
|
||||
self.assertEqual(self.calls(), [])
|
||||
|
||||
def test_train_failures(self):
|
||||
"""B1: no result + exit 1, and an API error result: status error, no base file."""
|
||||
for scenario, reason in (("noresult", "exit_code"), ("api_error", "usage_limit")):
|
||||
code, out = self.train(scenario)
|
||||
self.assertEqual((code, out["status"], out["reason"]), (1, "error", reason), scenario)
|
||||
self.assertIsNone(self.base_rec())
|
||||
|
||||
def test_unknown_base(self):
|
||||
"""B2: an unknown base is refused with no launch."""
|
||||
code, out = self.spawn("1.1", base="NOPE")
|
||||
self.assertEqual((code, out["error"]), (2, "unknown base"))
|
||||
self.assertEqual(self.calls(), [])
|
||||
self.assertFalse((self.state / "jobs" / "1.1.json").exists())
|
||||
|
||||
def test_cache_miss_is_reported(self):
|
||||
"""B2: first-usage cache_read 0 is `cache: miss`, not an error."""
|
||||
self.train()
|
||||
code, out = self.spawn("1.1", scenario="miss")
|
||||
self.assertEqual((code, out["status"], out["cache"]), (0, "ok", "miss"))
|
||||
|
||||
def test_parallel_spawns(self):
|
||||
"""B3: two spawns in parallel threads give two job files and two whole ledger lines."""
|
||||
self.train()
|
||||
ts = [threading.Thread(target=cli.main, args=(["spawn", "--run", self.run_md, "--base", "B1",
|
||||
"--job", j, "--brief", str(self.brief)],))
|
||||
for j in ("p1", "p2")]
|
||||
with contextlib.redirect_stdout(io.StringIO()):
|
||||
for t in ts:
|
||||
t.start()
|
||||
for t in ts:
|
||||
t.join(60)
|
||||
for j in ("p1", "p2"):
|
||||
rec = json.loads((self.state / "jobs" / (j + ".json")).read_text(encoding="utf-8"))
|
||||
self.assertEqual(rec["status"], "ok")
|
||||
lines = self.ledger()
|
||||
self.assertEqual(len(lines), 3)
|
||||
self.assertEqual(sorted(ln["job"] for ln in lines[1:]), ["p1", "p2"])
|
||||
|
||||
def test_stop_signal(self):
|
||||
"""B4: a stop signal to the helper ends the child, marks the job stopped, keeps raw."""
|
||||
self.train()
|
||||
env = dict(os.environ, FAKE_CLAUDE_SCENARIO="sleep")
|
||||
kw = {"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP} if os.name == "nt" else {}
|
||||
h = subprocess.Popen([sys.executable, str(SKILL / "gremlins.py"), "spawn", "--run", self.run_md,
|
||||
"--base", "B1", "--job", "s1", "--brief", str(self.brief)],
|
||||
env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, **kw)
|
||||
jpath, raw = self.state / "jobs" / "s1.json", self.state / "raw" / "s1.jsonl"
|
||||
deadline = time.time() + 30
|
||||
while time.time() < deadline and not (
|
||||
jpath.exists() and json.loads(jpath.read_text(encoding="utf-8")).get("child_pid")
|
||||
and raw.exists() and b'"assistant"' in raw.read_bytes()):
|
||||
time.sleep(0.2)
|
||||
pid = json.loads(jpath.read_text(encoding="utf-8"))["child_pid"]
|
||||
os.kill(h.pid, signal.CTRL_BREAK_EVENT if os.name == "nt" else signal.SIGTERM)
|
||||
out, _ = h.communicate(timeout=30)
|
||||
self.assertEqual(h.returncode, 1)
|
||||
self.assertEqual(json.loads(out.decode("utf-8").splitlines()[-1])["status"], "stopped")
|
||||
self.assertEqual(json.loads(jpath.read_text(encoding="utf-8"))["status"], "stopped")
|
||||
self.assertIn(b'"init"', raw.read_bytes())
|
||||
if os.name == "nt":
|
||||
r = subprocess.run(["tasklist", "/FI", "PID eq %d" % pid, "/NH"], capture_output=True)
|
||||
self.assertNotIn(str(pid).encode(), r.stdout)
|
||||
else:
|
||||
with self.assertRaises(ProcessLookupError):
|
||||
os.kill(pid, 0)
|
||||
|
||||
def test_usage_models_and_cost(self):
|
||||
"""B5: two models both in models_used; no total_cost_usd gives a null cost delta."""
|
||||
self.train("two_models")
|
||||
line = self.ledger()[-1]
|
||||
self.assertEqual(sorted(line["models_used"]), ["claude-haiku-5-5", MODEL])
|
||||
self.assertEqual(line["model"], MODEL)
|
||||
(self.state / "bases" / "B1.json").unlink()
|
||||
self.train("no_cost")
|
||||
self.assertIsNone(self.ledger()[-1]["cost_delta_usd"])
|
||||
|
||||
def test_status_cold_without_hit(self):
|
||||
"""B6: a base with no hit in the ledger is cold (and warm right after training)."""
|
||||
self.train()
|
||||
code, out = self.g("status", "--run", self.run_md)
|
||||
self.assertEqual((code, out["bases"]["B1"]["base_state"]), (0, "warm"))
|
||||
(self.state / "ledger.jsonl").write_bytes(b"")
|
||||
code, out = self.g("status", "--run", self.run_md)
|
||||
self.assertEqual(out["bases"]["B1"]["base_state"], "cold")
|
||||
|
||||
def test_rebase(self):
|
||||
"""B7: rebase resumes without a fork; the base takes the result's session id."""
|
||||
self.train()
|
||||
old = self.base_rec()["session_id"]
|
||||
code, out = self.g("rebase", "--run", self.run_md, "--base", "B1", "--brief", str(self.brief))
|
||||
self.assertEqual((code, out["status"]), (0, "ok"))
|
||||
argv = self.calls()[-1]["argv"]
|
||||
self.assertEqual(argv[-2:], ["--resume", old])
|
||||
self.assertNotIn("--fork-session", argv)
|
||||
new = self.base_rec()["session_id"]
|
||||
self.assertTrue(new.startswith("resumed-"))
|
||||
self.assertEqual(new, out["session_id"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,51 @@
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from grm import launch # noqa: E402
|
||||
|
||||
RUN = {"argv": ["/x/claude", "-p", "--model", "claude-opus-5-5[1m]", "--tools", "Bash,Read"]}
|
||||
|
||||
|
||||
class ArgvTest(unittest.TestCase):
|
||||
def test_fork_argv_extends_stored_argv(self):
|
||||
"""B2: the stored argv byte for byte, then --resume <base> --fork-session."""
|
||||
a = launch.fork_argv(RUN, "base-sid", "new-sid")
|
||||
n = len(RUN["argv"])
|
||||
self.assertEqual(a[:n], RUN["argv"])
|
||||
self.assertEqual(a[n:n + 3], ["--resume", "base-sid", "--fork-session"])
|
||||
self.assertIn("new-sid", a[n + 3:])
|
||||
self.assertEqual(len(RUN["argv"]), n) # the stored list is not mutated
|
||||
|
||||
def test_resume_argv_has_no_fork(self):
|
||||
"""B7: resume the base without forking."""
|
||||
a = launch.resume_argv(RUN, "base-sid")
|
||||
self.assertEqual(a[:len(RUN["argv"])], RUN["argv"])
|
||||
self.assertEqual(a[len(RUN["argv"]):], ["--resume", "base-sid"])
|
||||
self.assertNotIn("--fork-session", a)
|
||||
|
||||
|
||||
class ScrubTest(unittest.TestCase):
|
||||
def test_scrub_drops_pins_and_cleans_path(self):
|
||||
"""B8: SCRUB_DROP and UV_* dropped, pinned vars set, venv PATH entries removed."""
|
||||
venv = os.path.abspath(os.path.join(os.sep, "tmp", "venv"))
|
||||
keep = os.path.abspath(os.path.join(os.sep, "usr", "bin"))
|
||||
environ = {"PATH": os.pathsep.join([os.path.join(venv, "bin"), keep]),
|
||||
"VIRTUAL_ENV": venv, "UV_CACHE_DIR": "c", "ANTHROPIC_API_KEY": "k",
|
||||
"CLAUDE_CODE_SESSION_ID": "s", "DISABLE_AUTOUPDATER": "0", "OTHER": "o"}
|
||||
env = launch.scrubbed_env("/bin/claude", environ)
|
||||
for k in launch.SCRUB_DROP:
|
||||
self.assertNotIn(k, env)
|
||||
self.assertNotIn("UV_CACHE_DIR", env)
|
||||
for k, v in launch.SCRUB_PINNED.items():
|
||||
self.assertEqual(env[k], v)
|
||||
self.assertEqual(env[launch.SCRUB_PINNED_BINARY], "/bin/claude")
|
||||
self.assertEqual(env["PATH"], keep)
|
||||
self.assertEqual(env["OTHER"], "o")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,36 @@
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from grm import store # noqa: E402
|
||||
|
||||
|
||||
class LedgerTest(unittest.TestCase):
|
||||
def test_parallel_appends_give_whole_lines(self):
|
||||
"""B3: two threads appending at once leave only whole, parsable lines."""
|
||||
with tempfile.TemporaryDirectory() as d:
|
||||
run = str(Path(d) / "run.md")
|
||||
store.init_state(run)
|
||||
|
||||
def append(job):
|
||||
for i in range(50):
|
||||
store.ledger_append(run, {"job": job, "i": i, "pad": "x" * 500})
|
||||
threads = [threading.Thread(target=append, args=(j,)) for j in ("a", "b")]
|
||||
for t in threads:
|
||||
t.start()
|
||||
for t in threads:
|
||||
t.join()
|
||||
raw = (store.state_dir(run) / "ledger.jsonl").read_bytes().splitlines()
|
||||
lines = store.ledger_lines(run)
|
||||
self.assertEqual(len(raw), 100)
|
||||
self.assertEqual(len(lines), 100)
|
||||
self.assertEqual(sorted((x["job"], x["i"]) for x in lines),
|
||||
sorted((j, i) for j in ("a", "b") for i in range(50)))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,74 @@
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from grm import stream # noqa: E402
|
||||
|
||||
|
||||
def _ss(result=None):
|
||||
ss = stream.StreamState()
|
||||
if result is not None:
|
||||
ss.feed(dict(result, type="result"), 1.0)
|
||||
return ss
|
||||
|
||||
|
||||
class ClassifyTest(unittest.TestCase):
|
||||
def test_api_errors_map_to_reasons(self):
|
||||
"""B1: an API error result gives status error with its mapped reason."""
|
||||
cases = [({"is_error": True, "api_error_status": 429}, "usage_limit"),
|
||||
({"is_error": True, "api_error_status": 529}, "overloaded"),
|
||||
({"is_error": True, "result": "Prompt is too long"}, "prompt_too_long")]
|
||||
for res, reason in cases:
|
||||
self.assertEqual(stream.api_error_reason(res)[0], reason)
|
||||
self.assertEqual(stream.classify(_ss(res), 1, False), ("error", reason))
|
||||
|
||||
def test_no_result(self):
|
||||
"""B1: no result event is an error, by exit code or no_result."""
|
||||
self.assertEqual(stream.classify(_ss(), 0, False), ("error", "no_result"))
|
||||
self.assertEqual(stream.classify(_ss(), 1, False), ("error", "exit_code"))
|
||||
self.assertEqual(stream.classify(_ss({"is_error": False}), 0, False), ("ok", None))
|
||||
|
||||
|
||||
class UsageTest(unittest.TestCase):
|
||||
def test_cache_miss(self):
|
||||
"""B2: a first-request cache_read of 0 is a miss (reported, not an error)."""
|
||||
self.assertEqual(stream.cache_verdict({"cache_read_input_tokens": 0}, 50000), "miss")
|
||||
|
||||
def test_two_models_and_cost_delta(self):
|
||||
"""B5: both models in models_used; cost delta against the base, rounded at 6."""
|
||||
before = {"claude-opus-5-5": {"inputTokens": 10, "outputTokens": 5}}
|
||||
res = {"usage": {"input_tokens": 3, "output_tokens": 7},
|
||||
"modelUsage": {"claude-opus-5-5": {"inputTokens": 12, "outputTokens": 10},
|
||||
"claude-haiku-5-5": {"inputTokens": 1, "outputTokens": 1}},
|
||||
"total_cost_usd": 0.12345678}
|
||||
u = stream.usage_from_result(res, before, 0.1)
|
||||
self.assertEqual(u["models_used"], ["claude-opus-5-5", "claude-haiku-5-5"])
|
||||
self.assertEqual(u["cost_delta"], round(0.12345678 - 0.1, 6))
|
||||
self.assertEqual(u["total"], 0.123457)
|
||||
|
||||
def test_missing_cost(self):
|
||||
"""B5: no total_cost_usd gives cost_delta None."""
|
||||
self.assertIsNone(stream.usage_from_result({"usage": {}}, {}, 0.5)["cost_delta"])
|
||||
|
||||
|
||||
class WarmthTest(unittest.TestCase):
|
||||
def test_no_hit_is_cold(self):
|
||||
"""B6"""
|
||||
self.assertEqual(stream.warmth(None, 3600, 1000.0)["base_state"], "cold")
|
||||
|
||||
def test_5m_only_ttl(self):
|
||||
"""B6"""
|
||||
self.assertEqual(stream.infer_ttl({"cache_creation_5m": 10, "cache_creation_1h": 0}, None), 300)
|
||||
|
||||
def test_expiring(self):
|
||||
"""B6: a hit at now - ttl + 60 is expiring."""
|
||||
now = 1_000_000.0
|
||||
w = stream.warmth(now - 3600 + 60, 3600, now)
|
||||
self.assertEqual(w["base_state"], "expiring")
|
||||
self.assertEqual(w["ttl_left_s"], 60)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user