tools/usage_check.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Verify the plan-usage probe: it reads a description, and only a description.
The point of `clique/usage.py` is that the panel learns nothing about any
vendor. A block of TOML says where the token is, which URL answers, and which
fields hold the numbers; this runs that and nothing else. So the checks here
are mostly about a probe pointed at a local server that is not anybody's real
API, which is the strongest evidence the code is genuinely generic.
The rest is about failing quietly. A status bar that shouts because it could
not reach an API it does not need is worse than one that says nothing.
python3 tools/usage_check.py
Its own throwaway HTTP server. 0 on pass, 1 on fail.
"""
from __future__ import annotations
import json
import sys
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from clique import usage
from clique.registry import Registry
passed = failed = 0
seen: list[str] = []
def check(label: str, ok: bool, detail: object = "") -> None:
global passed, failed
if ok:
passed += 1
print(f" ok {label}")
else:
failed += 1
print(f" FAIL {label} {detail}")
REPLY = {
"alpha": {"used": 12.5, "until": "2026-09-01T00:00:00+00:00"},
"beta": {"used": 140, "until": None},
"gamma": {"used": "not a number"},
"epsilon": {"used": 5, "until": 1e300},
}
class Handler(BaseHTTPRequestHandler):
def do_GET(self) -> None:
seen.append(self.headers.get("Authorization", ""))
if self.path == "/huge":
body = json.dumps({"pad": "x" * (usage.MAX_BYTES + 1000)}).encode()
elif self.path == "/notjson":
body = b"<html>nope</html>"
elif self.path == "/expired":
self.send_response(401)
self.end_headers()
return
elif self.path == "/redirect":
self.send_response(302)
self.send_header("Location", "/usage")
self.end_headers()
return
elif self.path == "/teapot":
self.send_response(418)
self.end_headers()
return
else:
body = json.dumps(REPLY).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, *args) -> None:
pass
def allow_all(url: str) -> None:
"""Stand-in for the SSRF guard. The real one refuses loopback for models;
a probe on 127.0.0.1 is exactly what this check needs."""
if not url.startswith("http"):
raise ValueError("not http")
def main() -> int:
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()
base = f"http://127.0.0.1:{server.server_address[1]}"
home = Path("/tmp/clique-usage-check")
home.mkdir(parents=True, exist_ok=True)
creds = home / "creds.json"
creds.write_text(json.dumps({"outer": {"inner": {"tok": "SECRET-TOKEN"}}}))
spec = {
"url": base + "/usage",
"token_file": str(creds),
"token_field": "outer.inner.tok",
"headers": {"X-Probe": "yes"},
"window": [
{"label": "A", "percent": "alpha.used", "resets": "alpha.until"},
{"label": "B", "percent": "beta.used", "resets": "beta.until"},
{"label": "C", "percent": "gamma.used"},
{"label": "D", "percent": "delta.nothing"},
],
}
print("a probe is a description, and nothing here knows whose it is")
got = usage.read("fake", spec, allow_all, force=True)
check("it reads", bool(got), got)
labels = [w["label"] for w in (got or {}).get("windows", [])]
check("every window that produced a number is kept", labels == ["A", "B"], labels)
check("a field holding something that is not a number is dropped", "C" not in labels)
check("and a path that matches nothing is dropped", "D" not in labels)
first = (got or {}).get("windows", [{}])[0]
check("the number comes through", first.get("percent") == 12.5, first)
check("so does the reset time", first.get("resets_at", "").startswith("2026-09-01"), first)
over = (got or {}).get("windows", [{}, {}])[1]
# A width on a bar, so an API answering 140 must not paint past the end.
check("a percentage over a hundred is clamped", over.get("percent") == 100.0, over)
check("a missing reset time is null, not a guess", over.get("resets_at") is None, over)
print("the token is spent, not spread around")
check("it went out as a bearer token", seen and seen[-1] == "Bearer SECRET-TOKEN", seen[-1:])
check("and never rides the result", "SECRET-TOKEN" not in json.dumps(got), got)
print("everything that can go wrong says why, quietly")
def miss(label: str, override: dict, want: str) -> None:
usage.forget()
got = usage.read("f", {**spec, **override}, allow_all, force=True) or {}
check(label, got.get("windows") == [] and want in got.get("error", ""), got)
miss("no token file", {"token_file": str(home / "nope.json")}, "no token")
miss("a token field that is not there", {"token_field": "outer.missing"}, "no token")
miss("a refused URL", {"url": "file:///etc/passwd"}, "refused")
miss("a reply that is not JSON", {"url": base + "/notjson"}, "not JSON")
miss("a reply that is too big", {"url": base + "/huge"}, "too large")
miss("a bad status", {"url": base + "/teapot"}, "HTTP 418")
before = len(seen)
miss("a redirect is refused, not followed", {"url": base + "/redirect"}, "HTTP 302")
check("so the token went to one place only", len(seen) - before == 1, len(seen) - before)
usage.forget()
far = usage.read("f", {**spec, "window": [{"label": "E", "percent": "epsilon.used", "resets": "epsilon.until"}]},
allow_all, force=True) or {}
check("a reset time past any clock is null, not a crash",
far.get("windows") == [{"label": "E", "percent": 5.0, "resets_at": None}], far)
miss("a reply with none of the declared numbers", {"window": [{"label": "X", "percent": "nope"}]}, "no usage numbers")
miss("a command that fails", {"cmd": ["false"]}, "exited 1")
miss("a command that is not there", {"cmd": ["/nonexistent/probe"]}, "could not run")
check("no probe declared, no reading", usage.read("f", {}, allow_all, force=True) is None)
print("the cache is shared, so extra browsers are free")
usage.forget()
before = len(seen)
for _ in range(5):
usage.read("fake", spec, allow_all)
check("five reads are one request", len(seen) - before == 1, len(seen) - before)
usage.forget("fake")
usage.read("fake", spec, allow_all)
check("and forgetting one goes out again", len(seen) - before == 2, len(seen) - before)
usage.forget()
fails = len(seen)
for _ in range(3):
usage.read("f", {**spec, "url": base + "/teapot"}, allow_all)
check(
"a failure is cached too, so a box with no key stops asking",
len(seen) - fails == 1,
len(seen) - fails,
)
miss("an expired sign-in says so", {"url": base + "/expired"}, "sign in again")
print("an old reading answers at once, and one blip does not wipe it")
usage.forget()
good = usage.read("s", spec, allow_all) or {}
usage._cache["s"] = (0.0, good) # long past its TTL
before = len(seen)
stale = usage.read("s", spec, allow_all)
check("a stale reading is answered without waiting", stale is good, stale)
for _ in range(50):
if usage._cache["s"][0] > 0:
break
time.sleep(0.05)
check("and refreshed behind it", len(seen) - before == 1 and usage._cache["s"][0] > 0, len(seen) - before)
usage._cache["s"] = (0.0, good)
broken = usage.read("s", {**spec, "url": base + "/teapot"}, allow_all, force=True) or {}
check("a failed refresh keeps the last good bars",
broken["windows"] == good["windows"] and "418" in broken.get("error", ""), broken)
newer = {**good, "checked": 2**40}
usage._cache["s"] = (2.0**40, newer) # a forced probe already landed
late = usage._probe("s", spec, allow_all, good)
check("a slow probe does not overwrite a newer answer",
usage._cache["s"][1] is newer and late is newer, usage._cache["s"])
print("the shipped catalogue still says what it said")
claude = Registry(ROOT / "clique" / "config" / "clis.toml").types()["claude"]
# A sub-table opened in the wrong place silently swallows every bare key
# after it. That happened while this feature was being written, and these
# are the keys it ate.
check("claude still owns its input", claude.own_input is True)
check("claude still speaks hooks", claude.hooks is True)
check("claude still has a status page", "url" in claude.status, claude.status)
check("claude still has its attention patterns", "waiting" in claude.attention)
check("and it declares a usage probe", "url" in claude.usage, list(claude.usage))
check("with both windows", [w["label"] for w in claude.usage.get("window", [])] == ["5H", "7D"])
server.shutdown()
print(f"\n{passed} passed, {failed} failed")
return 0 if failed == 0 else 1
if __name__ == "__main__":
sys.exit(main())