Gabriel Vidal
c675cc1a8e
The notify CLI prices the transcript itself (main transcript only, family rates, flat cache multipliers), so a --final push carried two disagreeing figures ($46.55 from the CLI, $35.46 appended by the hub). The hub now rewrites the CLI's 'Claude Code · $x' with the run's exact list price (subagents included) and puts '≈Y% wk' right after it on a --final push. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
181 lines
8.5 KiB
Python
181 lines
8.5 KiB
Python
"""Unit tests for the plan-usage measuring path (``python3 -m unittest
|
|
test_usage_measuring`` from ``backend/``): the per-version price book, the
|
|
parser's one-usage-per-API-call dedup, and the calibration fit."""
|
|
|
|
import json
|
|
import pathlib
|
|
import tempfile
|
|
import unittest
|
|
|
|
import conversations
|
|
from conversations import pricing
|
|
from usage import calibrate
|
|
from usage.poller import UsagePoller
|
|
from usage.store import UsageStore
|
|
|
|
M = 1_000_000
|
|
|
|
|
|
def _u(inp=0, out=0, cr=0, c1=0, c5=0):
|
|
return {"input": inp, "output": out, "cacheRead": cr,
|
|
"cacheWriteTokens": c1 + c5, "cacheWriteUnits": c5 * 1.25 + c1 * 2.0}
|
|
|
|
|
|
class PriceBook(unittest.TestCase):
|
|
def test_exact_versions(self):
|
|
self.assertEqual(pricing.rates_for("claude-opus-5-5"), (4.0, 20.0, 0.20))
|
|
self.assertEqual(pricing.rates_for("claude-opus-5"), (5.0, 25.0, 0.50))
|
|
self.assertEqual(pricing.rates_for("claude-fable-5-1"), (10.0, 50.0, 0.25))
|
|
self.assertEqual(pricing.rates_for("claude-haiku-4-5-20251001"), (1.0, 5.0, 0.10))
|
|
|
|
def test_unknown_version_is_newest_of_family(self):
|
|
self.assertEqual(pricing.rates_for("claude-opus-9"), pricing.MODEL_PRICES["claude-opus-5-5"])
|
|
|
|
def test_cost_per_type(self):
|
|
self.assertAlmostEqual(pricing.usage_cost(_u(cr=M), "claude-opus-5-5"), 0.20)
|
|
self.assertAlmostEqual(pricing.usage_cost(_u(c1=M), "claude-opus-5-5"), 8.0)
|
|
self.assertAlmostEqual(pricing.usage_cost(_u(out=M), "claude-fable-5-1"), 50.0)
|
|
|
|
def test_fast_mode_doubles(self):
|
|
u = _u(inp=M, out=M)
|
|
self.assertAlmostEqual(pricing.usage_cost(u, "claude-opus-5-5", "fast"),
|
|
2 * pricing.usage_cost(u, "claude-opus-5-5"))
|
|
|
|
def test_haiku_long_prompt_tier(self):
|
|
short, long_ = _u(inp=50_000, out=1000), _u(inp=150_000, out=1000)
|
|
self.assertAlmostEqual(pricing.usage_cost(long_, "claude-haiku-5-5"),
|
|
5 * (150_000 * 0.10 + 1000 * 0.50) / M)
|
|
self.assertAlmostEqual(pricing.usage_cost(short, "claude-haiku-5-5"),
|
|
(50_000 * 0.10 + 1000 * 0.50) / M)
|
|
|
|
def test_plan_units_meter_new_models_at_predecessor_price(self):
|
|
u = _u(cr=M, out=M)
|
|
self.assertAlmostEqual(pricing.plan_units(u, "claude-opus-5-5"),
|
|
pricing.usage_cost(u, "claude-opus-5"))
|
|
self.assertAlmostEqual(pricing.plan_units(u, "claude-fable-5-1"),
|
|
1.75 * pricing.usage_cost(u, "claude-fable-5"))
|
|
self.assertEqual(pricing.plan_units(u, "qwen/qwen3.6-35b-a3b"), 0.0)
|
|
|
|
|
|
def _rec(mid, rid, blocks, usage, model="claude-opus-5-5", ts="2026-10-10T10:00:00Z"):
|
|
return {"type": "assistant", "sessionId": "s1", "timestamp": ts, "requestId": rid,
|
|
"effort": "xhigh", "message": {"id": mid, "model": model, "role": "assistant",
|
|
"content": blocks, "usage": usage}}
|
|
|
|
|
|
class ParserDedup(unittest.TestCase):
|
|
USAGE = {"input_tokens": 10, "output_tokens": 500, "cache_read_input_tokens": 100_000,
|
|
"cache_creation_input_tokens": 2000,
|
|
"cache_creation": {"ephemeral_1h_input_tokens": 2000, "ephemeral_5m_input_tokens": 0}}
|
|
|
|
def _parse(self, records):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
p = pathlib.Path(d) / "s1.jsonl"
|
|
p.write_text("\n".join(json.dumps(r) for r in records))
|
|
return conversations.parse_conversation(p, full=True)
|
|
|
|
def test_one_usage_per_api_call(self):
|
|
recs = [
|
|
{"type": "user", "sessionId": "s1", "timestamp": "2026-10-10T09:59:59Z",
|
|
"message": {"role": "user", "content": "hi"}},
|
|
_rec("m1", "r1", [{"type": "thinking", "thinking": ""}], self.USAGE),
|
|
_rec("m1", "r1", [{"type": "text", "text": "Looking."}], self.USAGE),
|
|
_rec("m1", "r1", [{"type": "tool_use", "id": "t1", "name": "Bash",
|
|
"input": {"command": "ls"}}], self.USAGE),
|
|
]
|
|
s = self._parse(recs)
|
|
nu = conversations.norm_usage(self.USAGE)
|
|
self.assertEqual(s["tokens"], conversations.usage_tokens(nu))
|
|
self.assertAlmostEqual(s["cost"], pricing.usage_cost(nu, "claude-opus-5-5"))
|
|
self.assertEqual(s["assistantTurns"], 1)
|
|
bm = s["byModel"]["claude-opus-5-5"]
|
|
self.assertEqual(bm["turns"], 1)
|
|
self.assertEqual(list(bm["efforts"]), ["xhigh"])
|
|
self.assertAlmostEqual(s["planUnits"], pricing.plan_units(nu, "claude-opus-5-5"))
|
|
# every split still sums to the conversation total
|
|
self.assertAlmostEqual(sum(v["cost"] for v in s["byTool"].values()), s["cost"])
|
|
self.assertAlmostEqual(sum(v[0] for v in s["meterBins"].values()), s["cost"])
|
|
# the call drove a Bash command: its charge is re-homed off `msg`
|
|
self.assertGreater(s["byTool"]["bash"]["cost"], 0)
|
|
self.assertLess(s["byTool"].get("msg", {"cost": 0})["cost"], s["byTool"]["bash"]["cost"])
|
|
# …and the per-turn badge sits on the tool card, once
|
|
th = s["thread"]
|
|
carriers = [i for i in th if "turnCost" in i]
|
|
self.assertEqual(len(carriers), 1)
|
|
self.assertAlmostEqual(sum(i["turnCost"] for i in carriers), s["cost"])
|
|
|
|
|
|
class Fit(unittest.TestCase):
|
|
def test_increments_skip_resets_and_stale(self):
|
|
H = calibrate.HOUR
|
|
pts = [(1 * H - 1, 10, 100), (2 * H - 1, 12, 100), (3 * H - 1, 15, 100),
|
|
(4 * H - 1, 1, 200), # reset (level drops, resets_at moves)
|
|
(5 * H - 1, 4, 200)]
|
|
inc = calibrate.hourly_increments(pts, 0, 6)
|
|
# hour 3 holds the reset; hour 5 keeps a fresh level (no rise)
|
|
self.assertEqual(inc, {1: 2, 2: 3, 4: 3, 5: 0})
|
|
# a gap longer than FRESH_S: the rise across it is never pinned on an hour
|
|
gap = calibrate.hourly_increments([(1 * H - 1, 10, 1), (9 * H - 1, 20, 1)], 0, 10)
|
|
self.assertEqual(sum(gap.values()), 0)
|
|
|
|
def test_lsq_recovers_rate_and_intercept(self):
|
|
xs = [float(i % 7) * 10 for i in range(60)]
|
|
ys = [x / 45.0 + 0.05 for x in xs]
|
|
a, c, r2 = calibrate._lsq(xs, ys)
|
|
self.assertAlmostEqual(1 / a, 45.0, places=6)
|
|
self.assertAlmostEqual(c, 0.05, places=6)
|
|
self.assertAlmostEqual(r2, 1.0, places=6)
|
|
|
|
|
|
class Poller(unittest.TestCase):
|
|
def test_records_changes_and_heartbeat_only(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
sd = pathlib.Path(d) / "state"
|
|
sd.mkdir()
|
|
store = UsageStore(str(pathlib.Path(d) / "x.db"))
|
|
poller = UsagePoller(store, dirs={"personal": sd})
|
|
|
|
def snap(t, pct):
|
|
(sd / "rate-limits.json").write_text(json.dumps({
|
|
"updated_at": t, "rate_limits": {
|
|
"five_hour": {"used_percentage": pct, "resets_at": 5000},
|
|
"seven_day": {"used_percentage": pct, "resets_at": 9000}}}))
|
|
snap(1000, 5)
|
|
self.assertEqual(poller.poll_once(), 2)
|
|
snap(1100, 5) # unchanged, too soon: nothing
|
|
self.assertEqual(poller.poll_once(), 0)
|
|
snap(1200, 6) # moved
|
|
self.assertEqual(poller.poll_once(), 2)
|
|
snap(1200 + 1800, 6) # heartbeat
|
|
self.assertEqual(poller.poll_once(), 2)
|
|
self.assertEqual(len(store.series("personal", 0)["seven_day"]), 3)
|
|
|
|
|
|
class RestateCost(unittest.TestCase):
|
|
def setUp(self):
|
|
self._st, self._pp = calibrate.session_totals, calibrate.plan_pct
|
|
calibrate.session_totals = lambda deps, sid: (35.456, 70.0, "personal")
|
|
calibrate.plan_pct = lambda deps, a, u: {"week": u / 46.0, "h5": u / 11.0}
|
|
|
|
def tearDown(self):
|
|
calibrate.session_totals, calibrate.plan_pct = self._st, self._pp
|
|
|
|
def test_cli_line_restated_with_plan_share_on_final(self):
|
|
msg = "all done\nClaude Code · $46.55 · 76.3M tokens · 2h 3m 39s"
|
|
self.assertEqual(calibrate.restate_cost(None, "s", msg, True),
|
|
"all done\nClaude Code · $35.46 · ≈1.5% wk · 76.3M tokens · 2h 3m 39s")
|
|
|
|
def test_non_final_only_reprices(self):
|
|
msg = "step\nClaude Code · $1,046.55 · 9k tokens"
|
|
self.assertEqual(calibrate.restate_cost(None, "s", msg, False),
|
|
"step\nClaude Code · $35.46 · 9k tokens")
|
|
|
|
def test_no_cli_line(self):
|
|
self.assertEqual(calibrate.restate_cost(None, "s", "done", True),
|
|
"done\n$35.46 · ≈1.5% wk")
|
|
self.assertEqual(calibrate.restate_cost(None, "s", "progress", False), "progress")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|