wenil/clawdmeter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

v3 M3: finalize z.ai against the real usage endpoint (monitor/quota/limit)

Browse Source 
Probing a live GLM Coding Plan key showed the Anthropic-compat /v1/messages
response carries NO rate-limit headers. z.ai instead exposes a dedicated status
endpoint (found via github.com/rygel/AIUsageTracker):

    GET https://api.z.ai/api/monitor/usage/quota/limit
    Authorization: <raw key>   (no "Bearer")   Accept-Language: en-US,en
    -> data.limits[] of TOKENS_LIMIT windows {percentage 0-100, nextResetTime ms, unit,number}
       + data.level (plan tier)

ZaiProvider now GETs that: the shortest TOKENS_LIMIT window -> the 5h bar (s/sr),
the longest -> the weekly bar (w/wr); the monthly TIME_LIMIT (web-tool quota) is
ignored; level -> status ("Lite"/"Pro"/…), any window at 100% -> "limited". It's
a status GET, so it does NOT spend the prompt-metered plan — no self-throttle
needed, polls on the normal cadence. Monitor URL derives from the configured
base host.

Verified live: s=1% (5h, resets ~4.9h), w=2% (weekly, resets ~6.5d), st="Lite".
test_zai rewritten for the JSON envelope (9 tests). probe_zai.py repointed to the
monitor endpoint. 122 passed, 2 Linux-only failures (baseline).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
wenil
2026-07-10 08:53:32 +03:00
co-authored by Claude Opus 4.8
parent ac0742d92c
commit 4eeb6e242b
3 changed files with 192 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files
daemon/tests/test_zai.py
+96 -76
View File
@@ -1,7 +1,8 @@
"""v3 M3 — z.ai (GLM) provider: Anthropic-compatible poll + quota-safe throttle.
"""v3 M3 — z.ai (GLM) provider: usage from the monitor/quota endpoint.
Uses httpx.MockTransport so no real z.ai call is made. Covers header mapping,
auth rejection, the no-known-headers fallback, and the self-throttle cache.
Uses httpx.MockTransport so no real z.ai call is made. Covers the token-window
mapping (5h -> s, weekly -> w), raw-key auth, the ignored monthly TIME_LIMIT,
and auth/error handling.
"""
import asyncio
@@ -10,7 +11,7 @@ import time
import httpx
from daemon.providers.zai import ZaiProvider, DEFAULT_BASE
from daemon.providers.zai import ZaiProvider, DEFAULT_HOST, _MONITOR_PATH
def _run(coro):
@@ -33,108 +34,127 @@ def _config(tmp_path, monkeypatch, zai: dict):
monkeypatch.setenv("LOCALAPPDATA", str(tmp_path))
def _transport(handler):
return httpx.MockTransport(handler)
def _tok(pct, reset_ms, unit=3, number=5):
return {"type": "TOKENS_LIMIT", "unit": unit, "number": number,
"percentage": pct, "nextResetTime": reset_ms}
def test_maps_unified_rate_limit_headers(tmp_path, monkeypatch):
_config(tmp_path, monkeypatch, {"enabled": True, "api_key": "sk-zai"})
now = time.time()
def _time_limit(pct=0):
return {"type": "TIME_LIMIT", "unit": 5, "number": 1, "usage": 100,
"currentValue": 0, "remaining": 100, "percentage": pct,
"nextResetTime": 9999999999999, "usageDetails": []}
def _ok_resp(limits, level="lite"):
return httpx.Response(200, json={"code": 200, "msg": "Operation successful",
"success": True, "data": {"level": level, "limits": limits}})
def _provider_returning(resp_or_fn):
def handler(request):
return resp_or_fn(request) if callable(resp_or_fn) else resp_or_fn
return ZaiProvider(transport=httpx.MockTransport(handler))
def test_maps_token_windows(tmp_path, monkeypatch):
_config(tmp_path, monkeypatch, {"enabled": True, "api_key": "sk-zai",
"base_url": "https://api.z.ai/api/anthropic"})
now_ms = time.time() * 1000
seen = {}
def handler(request: httpx.Request) -> httpx.Response:
seen["url"] = str(request.url)
seen["auth"] = request.headers.get("authorization")
return httpx.Response(200, headers={
"anthropic-ratelimit-unified-5h-utilization": "0.42",
"anthropic-ratelimit-unified-5h-reset": str(now + 3600),
"anthropic-ratelimit-unified-5h-status": "allowed",
"anthropic-ratelimit-unified-7d-utilization": "0.10",
"anthropic-ratelimit-unified-7d-reset": str(now + 7200),
}, json={"ok": True})
seen["lang"] = request.headers.get("accept-language")
seen["method"] = request.method
return _ok_resp([
_time_limit(), # ignored monthly web-tool quota
_tok(1, now_ms + 3600_000, unit=3, number=5), # 5h window -> short
_tok(2, now_ms + 6 * 86400_000, unit=6, number=1), # weekly window -> weekly
])
st = _run(ZaiProvider(transport=_transport(handler)).poll())
st = _run(ZaiProvider(transport=httpx.MockTransport(handler)).poll())
assert st.ok is True
assert st.s == 42.0 and 58 <= st.sr <= 60
assert st.w == 10.0 and 118 <= st.wr <= 120
assert st.st == "allowed"
assert seen["url"] == f"{DEFAULT_BASE}/v1/messages" # default base used
assert seen["auth"] == "Bearer sk-zai" # Bearer auth
assert st.s == 1.0 and 58 <= st.sr <= 60
assert st.w == 2.0 and 8000 <= st.wr <= 8700 # ~6 days in minutes
assert st.st == "Lite"
# raw key auth (no "Bearer"), monitor endpoint on the configured host, GET
assert seen["method"] == "GET"
assert seen["url"] == f"{DEFAULT_HOST}{_MONITOR_PATH}"
assert seen["auth"] == "sk-zai"
assert seen["lang"] == "en-US,en"
def test_no_key_reports_no_key(tmp_path, monkeypatch):
def test_no_key(tmp_path, monkeypatch):
_config(tmp_path, monkeypatch, {"enabled": True, "api_key": ""})
called = {"n": 0}
calls = {"n": 0}
def handler(request):
called["n"] += 1
return httpx.Response(200)
calls["n"] += 1
return _ok_resp([])
st = _run(ZaiProvider(transport=_transport(handler)).poll())
st = _run(ZaiProvider(transport=httpx.MockTransport(handler)).poll())
assert st.ok is False and st.st == "no key" and st.auth_problem is True
assert called["n"] == 0 # never hit the network
assert calls["n"] == 0
def test_auth_rejection(tmp_path, monkeypatch):
def test_http_401_is_bad_key(tmp_path, monkeypatch):
_config(tmp_path, monkeypatch, {"enabled": True, "api_key": "bad"})
def handler(request):
return httpx.Response(401, json={"error": "unauthorized"})
st = _run(ZaiProvider(transport=_transport(handler)).poll())
p = _provider_returning(httpx.Response(401, json={"error": "unauthorized"}))
st = _run(p.poll())
assert st.ok is False and st.st == "bad key" and st.auth_problem is True
def test_connected_without_known_headers(tmp_path, monkeypatch):
def test_body_success_false_is_bad_key(tmp_path, monkeypatch):
_config(tmp_path, monkeypatch, {"enabled": True, "api_key": "bad"})
p = _provider_returning(httpx.Response(200, json={"code": 401, "msg": "auth", "success": False, "data": None}))
st = _run(p.poll())
assert st.ok is False and st.st == "bad key" and st.auth_problem is True
def test_single_token_window(tmp_path, monkeypatch):
_config(tmp_path, monkeypatch, {"enabled": True, "api_key": "sk"})
now_ms = time.time() * 1000
p = _provider_returning(_ok_resp([_tok(37, now_ms + 3600_000)]))
st = _run(p.poll())
assert st.s == 37.0 and st.w == 0.0 and st.wr == -1 and st.ok is True
def test_hundred_percent_is_limited(tmp_path, monkeypatch):
_config(tmp_path, monkeypatch, {"enabled": True, "api_key": "sk"})
now_ms = time.time() * 1000
p = _provider_returning(_ok_resp([
_tok(100, now_ms + 60_000, unit=3, number=5),
_tok(40, now_ms + 6 * 86400_000, unit=6, number=1)]))
st = _run(p.poll())
assert st.st == "limited" and st.s == 100.0
def test_only_time_limit_reports_connected(tmp_path, monkeypatch):
_config(tmp_path, monkeypatch, {"enabled": True, "api_key": "sk"})
p = _provider_returning(_ok_resp([_time_limit(5)], level="pro"))
st = _run(p.poll())
assert st.ok is True and st.st == "Pro" and st.s == 0.0
def test_transient_error_reported(tmp_path, monkeypatch):
_config(tmp_path, monkeypatch, {"enabled": True, "api_key": "sk"})
def handler(request):
return httpx.Response(200, headers={"x-some-quota": "5"}, json={"ok": True})
raise httpx.ConnectError("boom")
st = _run(ZaiProvider(transport=_transport(handler)).poll())
assert st.ok is True and st.st == "allowed" and st.s == 0.0
st = _run(ZaiProvider(transport=httpx.MockTransport(handler)).poll())
assert st.ok is False and st.st == "offline"
def test_custom_base_url_and_model(tmp_path, monkeypatch):
_config(tmp_path, monkeypatch, {
"enabled": True, "api_key": "sk", "base_url": "https://zzz.example/anthropic/"})
def test_custom_base_host(tmp_path, monkeypatch):
_config(tmp_path, monkeypatch, {"enabled": True, "api_key": "sk",
"base_url": "https://zzz.example/api/anthropic"})
seen = {}
def handler(request):
seen["url"] = str(request.url)
seen["body"] = json.loads(request.content)
return httpx.Response(200, json={"ok": True})
return _ok_resp([])
_run(ZaiProvider(transport=_transport(handler)).poll())
assert seen["url"] == "https://zzz.example/anthropic/v1/messages" # trailing slash trimmed
assert seen["body"]["model"] == "glm-4.6"
def test_self_throttle_serves_cache(tmp_path, monkeypatch):
_config(tmp_path, monkeypatch, {"enabled": True, "api_key": "sk"})
calls = {"n": 0}
def handler(request):
calls["n"] += 1
return httpx.Response(200, json={"ok": True})
p = ZaiProvider(transport=_transport(handler))
st1 = _run(p.poll())
st2 = _run(p.poll()) # within the 15-min interval
assert calls["n"] == 1 # second served from cache
assert st1 is st2
def test_transient_error_not_cached(tmp_path, monkeypatch):
_config(tmp_path, monkeypatch, {"enabled": True, "api_key": "sk"})
calls = {"n": 0}
def handler(request):
calls["n"] += 1
raise httpx.ConnectError("boom")
p = ZaiProvider(transport=_transport(handler))
st1 = _run(p.poll())
st2 = _run(p.poll())
assert st1.ok is False and st1.st == "offline"
assert calls["n"] == 2 # blip retried, not cached
_run(ZaiProvider(transport=httpx.MockTransport(handler)).poll())
assert seen["url"] == f"https://zzz.example{_MONITOR_PATH}" # monitor path on the same host