vikingowl
73ce3ad0df
Env-driven Config fails fast on missing required values; LOCAL_MODE relaxes the K8s-specific knobs so the bot can run from a plain docker container against a bind-mounted oauth-tokens dir. Entrypoint wires together token refresh, usage poll, alert computation, optional armed cold-start message, E2EE Matrix send, and atomic state persistence. Adds a pytest -m live smoke test that hits the real /api/oauth/usage endpoint, skipped by default so CI stays offline.
65 lines
2.0 KiB
Python
65 lines
2.0 KiB
Python
"""Live smoke test against the real Anthropic /api/oauth/usage endpoint.
|
|
|
|
Skipped by default. Run with:
|
|
|
|
LIVE_TOKENS_DIR=$PWD/local/oauth-tokens pytest -m live tests/
|
|
|
|
Purpose: catch silent schema changes from Anthropic (e.g. utilization unit flip,
|
|
key rename, type change of resets_at). Unit tests can't see this; only a live
|
|
call can.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from claude_matrix_bot.clients import anthropic
|
|
|
|
|
|
pytestmark = pytest.mark.live
|
|
|
|
|
|
def _tokens_dir() -> str:
|
|
path = os.environ.get("LIVE_TOKENS_DIR", "").strip()
|
|
if not path:
|
|
pytest.skip("LIVE_TOKENS_DIR not set")
|
|
if not os.path.isdir(path):
|
|
pytest.skip(f"LIVE_TOKENS_DIR does not exist: {path}")
|
|
return path
|
|
|
|
|
|
def test_live_usage_endpoint_shape() -> None:
|
|
tokens = anthropic.load_tokens_from_dir(_tokens_dir())
|
|
if anthropic.needs_refresh(tokens, leeway_sec=60):
|
|
pytest.skip(
|
|
"access token near expiry; refresh outside the test "
|
|
"(via a normal local docker run) and retry"
|
|
)
|
|
|
|
fresh = anthropic.fetch_usage(
|
|
tokens.access_token,
|
|
api_base="https://api.anthropic.com",
|
|
)
|
|
|
|
assert len(fresh) > 0, "no recognized windows in /api/oauth/usage response"
|
|
|
|
now = time.time()
|
|
for name, snap in fresh.items():
|
|
# Catches the 0-100 vs 0-1 unit regression.
|
|
assert 0.0 <= snap.utilization <= 1.0, (
|
|
f"{name}.utilization out of fractional range: {snap.utilization!r}"
|
|
)
|
|
# resets_at must be plausibly future (or just slightly past, for races
|
|
# right around a window boundary). Catches int-vs-ms or seconds-vs-ms drift.
|
|
assert snap.resets_at > now - 600, (
|
|
f"{name}.resets_at unreasonably in the past: "
|
|
f"{snap.resets_at} vs now {int(now)}"
|
|
)
|
|
assert snap.resets_at < now + 14 * 86400, (
|
|
f"{name}.resets_at unreasonably far in future "
|
|
f"(>14d): {snap.resets_at}"
|
|
)
|