872 lines
37 KiB
Python
872 lines
37 KiB
Python
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import io
|
|
import os
|
|
from pathlib import Path
|
|
import importlib.util
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
def load_module():
|
|
module_path = Path(__file__).resolve().parents[1] / "sub2api_quota_tui.py"
|
|
spec = importlib.util.spec_from_file_location("sub2api_quota_tui", module_path)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class Sub2APIQuotaTUITests(unittest.TestCase):
|
|
def test_normalize_rows_sorts_by_usage_and_formats_windows(self) -> None:
|
|
mod = load_module()
|
|
payload = {
|
|
"generated_at": "2026-06-09T12:00:00+08:00",
|
|
"totals": {"total_accounts": 2, "usable_accounts": 2, "today_cost_usd": 0.75, "today_tokens": 3000, "today_requests": 9},
|
|
"accounts": [
|
|
{
|
|
"id": 1,
|
|
"name": "quota-low",
|
|
"routing_group": "slow",
|
|
"provider": "anthropic",
|
|
"kind": "quota_limited",
|
|
"status": "ok",
|
|
"account_type": "oauth",
|
|
"today_cost_usd": 0.25,
|
|
"today_tokens": 2000,
|
|
"today_requests": 4,
|
|
"windows": [
|
|
{"id": "five-hour", "used_percent": 80, "remaining_percent": 20, "reset": "2026-06-09T15:00:00+08:00"},
|
|
{"id": "weekly", "used_percent": 10, "remaining_percent": 90, "reset": "2026-06-10T15:00:00+08:00"},
|
|
],
|
|
},
|
|
{
|
|
"id": 2,
|
|
"name": "payg-high",
|
|
"routing_group": "fast",
|
|
"provider": "openai",
|
|
"kind": "pay_as_you_go",
|
|
"status": "ok",
|
|
"account_type": "apikey",
|
|
"daily_quota_limit": 100,
|
|
"daily_quota_used": 91.2,
|
|
"daily_quota_remaining": 8.8,
|
|
"daily_quota_used_percent": 91.2,
|
|
"today_cost_usd": 0.5,
|
|
"today_tokens": 1000,
|
|
"today_requests": 5,
|
|
"windows": [],
|
|
},
|
|
],
|
|
}
|
|
|
|
rows = mod.normalize_account_rows(payload)
|
|
|
|
self.assertEqual([row["name"] for row in rows], ["payg-high", "quota-low"])
|
|
self.assertEqual(rows[0]["routing_group"], "fast")
|
|
self.assertEqual(rows[0]["provider"], "openai")
|
|
self.assertEqual(rows[0]["daily_quota_cell"], "91.2/100")
|
|
self.assertEqual(rows[0]["kind_label"], "usage")
|
|
self.assertEqual(rows[1]["five_hour"], "80%/20%")
|
|
self.assertEqual(rows[1]["weekly"], "10%/90%")
|
|
self.assertIn("today $0.75", mod.summary_line(payload))
|
|
|
|
def test_filter_matches_kind_and_name(self) -> None:
|
|
mod = load_module()
|
|
payload = {
|
|
"accounts": [
|
|
{"id": 1, "name": "alpha", "routing_group": "slow", "provider": "anthropic", "kind": "quota_limited", "today_cost_usd": 0},
|
|
{"id": 2, "name": "beta", "routing_group": "sfast", "provider": "openai", "kind": "pay_as_you_go", "today_cost_usd": 0},
|
|
]
|
|
}
|
|
|
|
self.assertEqual([row["name"] for row in mod.normalize_account_rows(payload, "pay")], ["beta"])
|
|
self.assertEqual([row["name"] for row in mod.normalize_account_rows(payload, "alp")], ["alpha"])
|
|
self.assertEqual([row["name"] for row in mod.normalize_account_rows(payload, "sfast")], ["beta"])
|
|
self.assertEqual([row["name"] for row in mod.normalize_account_rows(payload, "anthropic")], ["alpha"])
|
|
|
|
def test_infers_public_status_url_from_accounts_url(self) -> None:
|
|
mod = load_module()
|
|
|
|
self.assertEqual(
|
|
mod.inferred_status_url("https://codex.server2.shujk.top/1232131231313123/api/tui/accounts"),
|
|
"https://codex.server2.shujk.top/1232131231313123/api/status",
|
|
)
|
|
self.assertEqual(mod.inferred_status_url("https://example.com/nope"), "")
|
|
|
|
def test_version_update_message_only_for_newer_versions(self) -> None:
|
|
mod = load_module()
|
|
|
|
self.assertEqual(mod.latest_version_from_text('name = "shusub2"\nversion = "0.1.8"\n'), "0.1.8")
|
|
self.assertIn("0.1.7 -> 0.1.8", mod.version_update_message("0.1.8", "0.1.7"))
|
|
self.assertEqual(mod.version_update_message("0.1.7", "0.1.7"), "")
|
|
self.assertEqual(mod.version_update_message("0.1.6", "0.1.7"), "")
|
|
|
|
def test_gzip_json_requests_decode_responses_and_request_compression(self) -> None:
|
|
mod = load_module()
|
|
payload = {"ok": True}
|
|
compressed = mod.gzip.compress(mod.json.dumps(payload).encode("utf-8"))
|
|
requests = []
|
|
|
|
class FakeResponse:
|
|
headers = {"Content-Encoding": "gzip"}
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, traceback):
|
|
return False
|
|
|
|
def read(self):
|
|
return compressed
|
|
|
|
old_urlopen = mod.urllib.request.urlopen
|
|
mod.urllib.request.urlopen = lambda request, timeout: (requests.append(request) or FakeResponse())
|
|
try:
|
|
self.assertEqual(mod.fetch_payload("https://example.test/accounts", 1), payload)
|
|
self.assertEqual(mod.fetch_logs_payload("https://example.test/usage", "test-token", 1), payload)
|
|
self.assertEqual(mod.fetch_admin_json("https://example.test/admin", "test-token", 1, body={"ids": [1]}), payload)
|
|
self.assertEqual(mod.fetch_errors_payload("https://example.test/errors", "test-token", 1), payload)
|
|
finally:
|
|
mod.urllib.request.urlopen = old_urlopen
|
|
|
|
class PlainResponse:
|
|
headers = {}
|
|
|
|
def read(self):
|
|
return mod.json.dumps(payload).encode("utf-8")
|
|
|
|
self.assertEqual(mod.decode_json_response(PlainResponse(), "plain JSON failed"), payload)
|
|
self.assertEqual(len(requests), 4)
|
|
self.assertTrue(all(request.get_header("Accept-encoding") == "gzip" for request in requests))
|
|
|
|
def test_default_refresh_intervals_are_five_minutes(self) -> None:
|
|
mod = load_module()
|
|
|
|
self.assertEqual(mod.DEFAULT_REFRESH_SECONDS, 300)
|
|
self.assertEqual(mod.DEFAULT_LOGS_REFRESH_SECONDS, 300)
|
|
self.assertEqual(mod.DEFAULT_ERRORS_REFRESH_SECONDS, 300)
|
|
|
|
def test_write_api_url_config_uses_config_file_env(self) -> None:
|
|
mod = load_module()
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
config_file = Path(tmp) / "shusub2" / "api-url"
|
|
old_value = os.environ.get("SHUSUB2_API_URL_FILE")
|
|
os.environ["SHUSUB2_API_URL_FILE"] = str(config_file)
|
|
try:
|
|
written = mod.write_api_url_config("https://example.com/api/tui/accounts")
|
|
self.assertEqual(written, config_file)
|
|
self.assertEqual(config_file.read_text(encoding="utf-8"), "https://example.com/api/tui/accounts\n")
|
|
finally:
|
|
if old_value is None:
|
|
os.environ.pop("SHUSUB2_API_URL_FILE", None)
|
|
else:
|
|
os.environ["SHUSUB2_API_URL_FILE"] = old_value
|
|
|
|
def test_once_output_is_name_first_and_includes_daily_quota(self) -> None:
|
|
mod = load_module()
|
|
payload = {
|
|
"generated_at": "2026-06-09T12:00:00+08:00",
|
|
"totals": {"total_accounts": 1, "usable_accounts": 1, "today_cost_usd": 1, "today_tokens": 100, "today_requests": 2},
|
|
"accounts": [
|
|
{
|
|
"id": 2,
|
|
"name": "input 300",
|
|
"routing_group": "fast",
|
|
"provider": "openai",
|
|
"kind": "daily_limited",
|
|
"status": "ok",
|
|
"account_type": "apikey",
|
|
"daily_quota_limit": 300,
|
|
"daily_quota_used": 96.6,
|
|
"daily_quota_remaining": 203.4,
|
|
"daily_quota_used_percent": 32.2,
|
|
"daily_quota_reset_at": "2026-06-09T16:00:00Z",
|
|
"today_cost_usd": 1,
|
|
"today_tokens": 100,
|
|
"today_requests": 2,
|
|
"windows": [],
|
|
}
|
|
],
|
|
}
|
|
|
|
rows = mod.normalize_account_rows(payload)
|
|
|
|
self.assertEqual(rows[0]["kind_label"], "daily")
|
|
self.assertEqual(rows[0]["routing_group"], "fast")
|
|
self.assertEqual(rows[0]["provider"], "openai")
|
|
self.assertEqual(rows[0]["daily_quota_cell"], "96.6/300")
|
|
self.assertTrue(rows[0]["reset"].endswith("00:00") or rows[0]["reset"] != "-")
|
|
|
|
def test_daily_quota_cell_is_dash_without_limit(self) -> None:
|
|
mod = load_module()
|
|
|
|
self.assertEqual(mod.daily_quota_cell({"daily_quota_used": 0}), "-")
|
|
|
|
def test_monitor_availability_column_uses_bound_monitor_status(self) -> None:
|
|
mod = load_module()
|
|
payload = {
|
|
"generated_at": "2026-06-09T12:00:00+08:00",
|
|
"totals": {"total_accounts": 1, "usable_accounts": 1, "today_cost_usd": 1, "today_tokens": 100, "today_requests": 2},
|
|
"accounts": [
|
|
{
|
|
"id": 2,
|
|
"name": "input 300",
|
|
"base_url_hash": "input-hash",
|
|
"routing_group": "fast",
|
|
"provider": "openai",
|
|
"kind": "daily_limited",
|
|
"status": "ok",
|
|
"account_type": "apikey",
|
|
"today_cost_usd": 1,
|
|
"today_tokens": 100,
|
|
"today_requests": 2,
|
|
"windows": [],
|
|
}
|
|
],
|
|
}
|
|
status_payload = {
|
|
"channel_monitors": {
|
|
"items": [
|
|
{
|
|
"name": "other responses",
|
|
"base_url_hash": "input-hash",
|
|
"enabled": True,
|
|
"latest_status": "operational",
|
|
"latency_ms": 2606,
|
|
}
|
|
]
|
|
}
|
|
}
|
|
rows = mod.normalize_account_rows(payload)
|
|
out = io.StringIO()
|
|
|
|
with contextlib.redirect_stdout(out):
|
|
mod.print_once(payload, status_payload=status_payload)
|
|
|
|
self.assertEqual(mod.monitor_availability(rows[0], status_payload), "ok")
|
|
self.assertIn("availability", out.getvalue())
|
|
self.assertIn(" ok\n", out.getvalue())
|
|
|
|
def test_monitor_availability_is_dash_without_bound_monitor(self) -> None:
|
|
mod = load_module()
|
|
self.assertEqual(mod.monitor_availability({"name": "alpha"}, {}), "-")
|
|
|
|
def test_routing_group_sort_shows_sfast_first(self) -> None:
|
|
mod = load_module()
|
|
payload = {
|
|
"accounts": [
|
|
{"id": 1, "name": "manual", "routing_group": "id", "kind": "pay_as_you_go", "today_cost_usd": 10},
|
|
{"id": 2, "name": "slow", "routing_group": "slow", "kind": "pay_as_you_go", "today_cost_usd": 1},
|
|
{"id": 3, "name": "fast", "routing_group": "fast", "kind": "pay_as_you_go", "today_cost_usd": 1},
|
|
{"id": 4, "name": "sfast", "routing_group": "sfast", "kind": "pay_as_you_go", "today_cost_usd": 1},
|
|
]
|
|
}
|
|
|
|
rows = mod.normalize_account_rows(payload)
|
|
|
|
self.assertEqual([row["name"] for row in rows], ["sfast", "fast", "slow", "manual"])
|
|
|
|
def test_monitor_summary_and_account_match(self) -> None:
|
|
mod = load_module()
|
|
status_payload = {
|
|
"channel_monitors": {
|
|
"enabled": 6,
|
|
"latest_ok": 5,
|
|
"latest_failed": 0,
|
|
"latest_unknown": 1,
|
|
"latest_checked_max": "2026-06-09 15:07:40+08:00",
|
|
"items": [
|
|
{
|
|
"name": "input responses",
|
|
"provider": "openai",
|
|
"enabled": True,
|
|
"primary_model": "gpt-5.5",
|
|
"latest_status": "operational",
|
|
"latency_ms": 2606,
|
|
"checked_at": "2026-06-09 15:03:28+08:00",
|
|
"message": "",
|
|
},
|
|
{
|
|
"name": "kedaya responses",
|
|
"provider": "openai",
|
|
"enabled": True,
|
|
"primary_model": "gpt-5.5",
|
|
"latest_status": "degraded",
|
|
"latency_ms": 25923,
|
|
"checked_at": "2026-06-09 15:07:40+08:00",
|
|
"message": "slow response: 25923ms",
|
|
},
|
|
],
|
|
}
|
|
}
|
|
row = {"name": "input 300"}
|
|
|
|
self.assertIn("monitors 5 ok / 0 failed / 1 unknown", mod.monitor_summary(status_payload))
|
|
self.assertIn("monitor operational 2606ms", mod.monitor_detail(row, status_payload))
|
|
|
|
def test_monitor_match_prefers_base_url_hash_over_name_tokens(self) -> None:
|
|
mod = load_module()
|
|
status_payload = {
|
|
"channel_monitors": {
|
|
"items": [
|
|
{
|
|
"name": "input responses",
|
|
"base_url_hash": "wrong-hash",
|
|
"latest_status": "operational",
|
|
"latency_ms": 111,
|
|
},
|
|
{
|
|
"name": "unrelated responses",
|
|
"base_url_hash": "right-hash",
|
|
"latest_status": "degraded",
|
|
"latency_ms": 222,
|
|
},
|
|
],
|
|
}
|
|
}
|
|
row = {"name": "input 300", "base_url_hash": "right-hash"}
|
|
|
|
self.assertEqual(mod.matching_monitor(row, status_payload)["name"], "unrelated responses")
|
|
self.assertIn("monitor degraded 222ms", mod.monitor_detail(row, status_payload))
|
|
|
|
def test_monitor_error_is_non_blocking_in_once_output(self) -> None:
|
|
mod = load_module()
|
|
payload = {
|
|
"generated_at": "2026-06-09T12:00:00+08:00",
|
|
"totals": {"total_accounts": 1, "usable_accounts": 1, "today_cost_usd": 0, "today_tokens": 0, "today_requests": 0},
|
|
"accounts": [
|
|
{"id": 1, "name": "alpha", "routing_group": "slow", "provider": "openai", "kind": "pay_as_you_go", "status": "ok"}
|
|
],
|
|
}
|
|
out = io.StringIO()
|
|
|
|
with contextlib.redirect_stdout(out):
|
|
mod.print_once(payload, status_error="timeout")
|
|
|
|
text = out.getvalue()
|
|
self.assertIn("monitors error: timeout", text)
|
|
self.assertIn("alpha", text)
|
|
|
|
def test_once_output_includes_monitor_summary(self) -> None:
|
|
mod = load_module()
|
|
payload = {
|
|
"generated_at": "2026-06-09T12:00:00+08:00",
|
|
"totals": {"total_accounts": 0, "usable_accounts": 0, "today_cost_usd": 0, "today_tokens": 0, "today_requests": 0},
|
|
"accounts": [],
|
|
}
|
|
status_payload = {"channel_monitors": {"enabled": 1, "latest_ok": 1, "latest_failed": 0, "latest_unknown": 0}}
|
|
out = io.StringIO()
|
|
|
|
with contextlib.redirect_stdout(out):
|
|
mod.print_once(payload, status_payload=status_payload)
|
|
|
|
self.assertIn("monitors 1 ok / 0 failed / 0 unknown", out.getvalue())
|
|
|
|
|
|
def sample_logs_payload() -> dict:
|
|
return {
|
|
"code": 0,
|
|
"message": "success",
|
|
"data": {
|
|
"total": 2345,
|
|
"page": 1,
|
|
"page_size": 100,
|
|
"items": [
|
|
{
|
|
"id": 11,
|
|
"api_key_id": 3,
|
|
"api_key": {"id": 3, "name": "codex-main"},
|
|
"account_id": 7,
|
|
"account": {"id": 7, "name": "oai-sub-1"},
|
|
"user": {"id": 1, "name": "shujakuin"},
|
|
"model": "gpt-5.5",
|
|
"upstream_model": "gpt-5.5-codex",
|
|
"request_type": "stream",
|
|
"stream": True,
|
|
"input_tokens": 1200,
|
|
"output_tokens": 340,
|
|
"cache_creation_tokens": 50,
|
|
"cache_read_tokens": 4100,
|
|
"total_cost": 0.0123,
|
|
"actual_cost": 0.0061,
|
|
"duration_ms": 5321,
|
|
"first_token_ms": 800,
|
|
"request_id": "req-abc",
|
|
"created_at": "2026-07-21T10:00:00+08:00",
|
|
},
|
|
{
|
|
"id": 12,
|
|
"api_key_id": 4,
|
|
"account_id": 9,
|
|
"account": {"id": 9, "name": "claude-max"},
|
|
"model": "claude-sonnet-5",
|
|
"request_type": "",
|
|
"stream": False,
|
|
"input_tokens": 100,
|
|
"output_tokens": 20,
|
|
"total_cost": 0.001,
|
|
"duration_ms": 900,
|
|
"created_at": "2026-07-21T10:05:00+08:00",
|
|
},
|
|
],
|
|
},
|
|
}
|
|
|
|
|
|
class Sub2APILogsTests(unittest.TestCase):
|
|
def test_normalize_log_rows_maps_columns_and_sorts_newest_first(self) -> None:
|
|
mod = load_module()
|
|
|
|
rows = mod.normalize_log_rows(sample_logs_payload())
|
|
|
|
self.assertEqual([row["id"] for row in rows], [12, 11])
|
|
newest, oldest = rows
|
|
self.assertEqual(newest["key"], "#4")
|
|
self.assertEqual(newest["account"], "claude-max")
|
|
self.assertEqual(newest["type"], "sync")
|
|
self.assertEqual(newest["tokens"], 120)
|
|
self.assertEqual(oldest["key"], "codex-main")
|
|
self.assertEqual(oldest["account"], "oai-sub-1")
|
|
self.assertEqual(oldest["user"], "shujakuin")
|
|
self.assertEqual(oldest["type"], "stream")
|
|
self.assertEqual(oldest["tokens"], 1200 + 340 + 50 + 4100)
|
|
self.assertEqual(oldest["input_tokens"], 1200)
|
|
self.assertEqual(oldest["output_tokens"], 340)
|
|
self.assertEqual(oldest["cache_tokens"], 50 + 4100)
|
|
self.assertEqual(oldest["first_token_ms"], 800)
|
|
self.assertGreater(oldest["tokens_per_second"], 0)
|
|
self.assertEqual(oldest["cost"], 0.0123)
|
|
self.assertEqual(oldest["duration_ms"], 5321)
|
|
self.assertTrue(oldest["time"].endswith("10:00") or oldest["time"] != "-")
|
|
|
|
def test_relative_age_uses_compact_minutes_and_hours(self) -> None:
|
|
mod = load_module()
|
|
now = mod.dt.datetime(2026, 7, 24, 12, 5, tzinfo=mod.dt.timezone.utc)
|
|
|
|
self.assertEqual(mod.relative_age("2026-07-24T12:05:00Z", now), "now")
|
|
self.assertEqual(mod.relative_age("2026-07-24T12:00:00Z", now), "5m ago")
|
|
self.assertEqual(mod.relative_age("2026-07-24T10:05:00Z", now), "2h ago")
|
|
self.assertEqual(mod.relative_age("bad", now), "-")
|
|
|
|
def test_normalize_log_rows_filter_matches_key_account_model(self) -> None:
|
|
mod = load_module()
|
|
payload = sample_logs_payload()
|
|
|
|
self.assertEqual([row["id"] for row in mod.normalize_log_rows(payload, "codex-main")], [11])
|
|
self.assertEqual([row["id"] for row in mod.normalize_log_rows(payload, "claude")], [12])
|
|
self.assertEqual([row["id"] for row in mod.normalize_log_rows(payload, "stream")], [11])
|
|
self.assertEqual(mod.normalize_log_rows(payload, "no-match"), [])
|
|
|
|
def test_log_items_supports_flat_and_enveloped_payloads(self) -> None:
|
|
mod = load_module()
|
|
|
|
self.assertEqual(len(mod.log_items(sample_logs_payload())), 2)
|
|
self.assertEqual(len(mod.log_items({"items": [{"id": 1}]})), 1)
|
|
self.assertEqual(mod.log_items({"data": {"items": "nope"}}), [])
|
|
self.assertEqual(mod.log_items({}), [])
|
|
|
|
def test_log_type_label_falls_back_to_legacy_flags(self) -> None:
|
|
mod = load_module()
|
|
|
|
self.assertEqual(mod.log_type_label({"request_type": "ws_v2"}), "ws_v2")
|
|
self.assertEqual(mod.log_type_label({"request_type": "unknown", "openai_ws_mode": True}), "ws_v2")
|
|
self.assertEqual(mod.log_type_label({"stream": True}), "stream")
|
|
self.assertEqual(mod.log_type_label({"stream": False}), "sync")
|
|
self.assertEqual(mod.log_type_label({}), "-")
|
|
|
|
def test_logs_request_url_forces_first_page_and_limit(self) -> None:
|
|
mod = load_module()
|
|
|
|
url = mod.logs_request_url("https://sub2apicn.shujk.top/api/v1/admin/usage?page=9&page_size=5&user_id=2", 100)
|
|
parsed = dict(part.split("=") for part in url.split("?", 1)[1].split("&"))
|
|
self.assertEqual(parsed["page"], "1")
|
|
self.assertEqual(parsed["page_size"], "100")
|
|
self.assertEqual(parsed["user_id"], "2")
|
|
|
|
capped = mod.logs_request_url("https://sub2apicn.shujk.top/api/v1/admin/usage", 9000)
|
|
self.assertIn("page_size=1000", capped)
|
|
|
|
def test_print_logs_once_renders_requested_columns(self) -> None:
|
|
mod = load_module()
|
|
out = io.StringIO()
|
|
|
|
with contextlib.redirect_stdout(out):
|
|
mod.print_logs_once(sample_logs_payload())
|
|
|
|
text = out.getvalue()
|
|
self.assertIn("key", text)
|
|
self.assertIn("codex-main", text)
|
|
self.assertIn("oai-sub-1", text)
|
|
self.assertIn("gpt-5.5", text)
|
|
self.assertIn("stream", text)
|
|
self.assertIn("input", text)
|
|
self.assertIn("output", text)
|
|
self.assertIn("cache", text)
|
|
self.assertIn("duration", text)
|
|
self.assertIn("1.2K", text)
|
|
self.assertIn("4.2K", text)
|
|
self.assertIn("5.7K", text)
|
|
self.assertIn("$0.012", text)
|
|
self.assertIn("5.3s", text)
|
|
self.assertIn("total 2.3K records", text)
|
|
|
|
def test_default_logs_token_reads_config_file(self) -> None:
|
|
mod = load_module()
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
token_file = Path(tmp) / "logs-token"
|
|
token_file.write_text("# comment\nsk-admin-token\n", encoding="utf-8")
|
|
old_value = os.environ.get("SHUSUB2_LOGS_TOKEN_FILE")
|
|
os.environ["SHUSUB2_LOGS_TOKEN_FILE"] = str(token_file)
|
|
try:
|
|
self.assertEqual(mod.default_logs_token(), "sk-admin-token")
|
|
written = mod.write_config_value(mod.logs_token_config_file_path(), "sk-new-token")
|
|
self.assertEqual(written, token_file)
|
|
self.assertEqual(token_file.read_text(encoding="utf-8"), "sk-new-token\n")
|
|
finally:
|
|
if old_value is None:
|
|
os.environ.pop("SHUSUB2_LOGS_TOKEN_FILE", None)
|
|
else:
|
|
os.environ["SHUSUB2_LOGS_TOKEN_FILE"] = old_value
|
|
|
|
def test_once_logs_without_token_prints_hint_and_exits_2(self) -> None:
|
|
mod = load_module()
|
|
old_token = os.environ.pop("SHUSUB2_LOGS_TOKEN", None)
|
|
old_file = os.environ.get("SHUSUB2_LOGS_TOKEN_FILE")
|
|
os.environ["SHUSUB2_LOGS_TOKEN_FILE"] = "/nonexistent/shusub2/logs-token"
|
|
err = io.StringIO()
|
|
try:
|
|
with contextlib.redirect_stderr(err):
|
|
rc = mod.main(["--once", "--logs", "--no-version-check"])
|
|
finally:
|
|
if old_token is not None:
|
|
os.environ["SHUSUB2_LOGS_TOKEN"] = old_token
|
|
if old_file is None:
|
|
os.environ.pop("SHUSUB2_LOGS_TOKEN_FILE", None)
|
|
else:
|
|
os.environ["SHUSUB2_LOGS_TOKEN_FILE"] = old_file
|
|
|
|
self.assertEqual(rc, 2)
|
|
self.assertIn("logs token not configured", err.getvalue())
|
|
|
|
def test_default_logs_url_defaults_to_sub2apicn(self) -> None:
|
|
mod = load_module()
|
|
old_url = os.environ.pop("SHUSUB2_LOGS_URL", None)
|
|
old_file = os.environ.get("SHUSUB2_LOGS_URL_FILE")
|
|
os.environ["SHUSUB2_LOGS_URL_FILE"] = "/nonexistent/shusub2/logs-url"
|
|
try:
|
|
self.assertEqual(mod.default_logs_url(), "https://sub2apicn.shujk.top/api/v1/admin/usage")
|
|
finally:
|
|
if old_url is not None:
|
|
os.environ["SHUSUB2_LOGS_URL"] = old_url
|
|
if old_file is None:
|
|
os.environ.pop("SHUSUB2_LOGS_URL_FILE", None)
|
|
else:
|
|
os.environ["SHUSUB2_LOGS_URL_FILE"] = old_file
|
|
|
|
def test_errors_request_url_forces_page_limit_and_time_range(self) -> None:
|
|
mod = load_module()
|
|
|
|
url = mod.errors_request_url(
|
|
"https://sub2apicn.shujk.top/api/v1/admin/ops/errors?page=9&page_size=5&q=x",
|
|
100,
|
|
"6h",
|
|
)
|
|
parsed = dict(part.split("=") for part in url.split("?", 1)[1].split("&"))
|
|
self.assertEqual(parsed["page"], "1")
|
|
self.assertEqual(parsed["page_size"], "100")
|
|
self.assertEqual(parsed["time_range"], "6h")
|
|
self.assertEqual(parsed["q"], "x")
|
|
|
|
capped = mod.errors_request_url("https://sub2apicn.shujk.top/api/v1/admin/ops/errors", 9000)
|
|
self.assertIn("page_size=500", capped)
|
|
self.assertIn("time_range=24h", capped)
|
|
|
|
def test_normalize_error_rows_merges_filters_and_sorts(self) -> None:
|
|
mod = load_module()
|
|
payload = {
|
|
"items": [
|
|
{
|
|
"_node": "cn",
|
|
"id": 1,
|
|
"created_at": "2026-07-24T10:00:00+08:00",
|
|
"status_code": 429,
|
|
"api_key_name": "key-a",
|
|
"account_name": "acc-a",
|
|
"model": "gpt-5.5",
|
|
"phase": "upstream",
|
|
"type": "rate_limit",
|
|
"error_owner": "provider",
|
|
"error_source": "upstream_http",
|
|
"message": "too many requests",
|
|
"request_id": "req-a",
|
|
},
|
|
{
|
|
"_node": "us",
|
|
"id": 2,
|
|
"created_at": "2026-07-24T11:00:00+08:00",
|
|
"status_code": 500,
|
|
"api_key_id": 9,
|
|
"account_id": 12,
|
|
"requested_model": "claude-opus",
|
|
"phase": "internal",
|
|
"type": "api_error",
|
|
"error_owner": "platform",
|
|
"error_source": "gateway",
|
|
"message": "upstream failed",
|
|
"request_id": "req-b",
|
|
},
|
|
]
|
|
}
|
|
|
|
rows = mod.normalize_error_rows(payload)
|
|
self.assertEqual([row["id"] for row in rows], [2, 1])
|
|
self.assertEqual(rows[0]["node"], "us")
|
|
self.assertEqual(rows[0]["key"], "#9")
|
|
self.assertEqual(rows[0]["account"], "#12")
|
|
self.assertEqual(rows[0]["model"], "claude-opus")
|
|
self.assertEqual([row["id"] for row in mod.normalize_error_rows(payload, "rate_limit")], [1])
|
|
self.assertEqual([row["id"] for row in mod.normalize_error_rows(payload, "us")], [2])
|
|
self.assertEqual(mod.normalize_error_rows(payload, "no-match"), [])
|
|
|
|
def test_fetch_merged_errors_labels_nodes_and_tolerates_partial_failure(self) -> None:
|
|
mod = load_module()
|
|
|
|
def fake_fetch(url: str, token: str, timeout: int, limit: int = 100, time_range: str = "24h"):
|
|
if "cn" in url:
|
|
return {
|
|
"data": {
|
|
"items": [
|
|
{
|
|
"id": 10,
|
|
"created_at": "2026-07-24T12:00:00+08:00",
|
|
"status_code": 404,
|
|
"model": "gpt-a",
|
|
"message": "cn miss",
|
|
}
|
|
],
|
|
"total": 3,
|
|
}
|
|
}
|
|
raise RuntimeError("us down")
|
|
|
|
old = mod.fetch_errors_payload
|
|
mod.fetch_errors_payload = fake_fetch
|
|
try:
|
|
payload = mod.fetch_merged_errors(
|
|
[("cn", "https://example/cn"), ("us", "https://example/us")],
|
|
"token",
|
|
5,
|
|
50,
|
|
"1h",
|
|
)
|
|
finally:
|
|
mod.fetch_errors_payload = old
|
|
|
|
self.assertEqual(len(payload["items"]), 1)
|
|
self.assertEqual(payload["items"][0]["_node"], "cn")
|
|
self.assertTrue(payload["sources"]["cn"]["ok"])
|
|
self.assertEqual(payload["sources"]["cn"]["total"], 3)
|
|
self.assertFalse(payload["sources"]["us"]["ok"])
|
|
self.assertIn("us down", payload["sources"]["us"]["error"])
|
|
self.assertIn("cn 1/3", mod.errors_summary_line(payload, 1))
|
|
self.assertIn("us err", mod.errors_summary_line(payload, 1))
|
|
|
|
def test_default_errors_urls(self) -> None:
|
|
mod = load_module()
|
|
old_cn = os.environ.pop("SHUSUB2_ERRORS_CN_URL", None)
|
|
old_us = os.environ.pop("SHUSUB2_ERRORS_US_URL", None)
|
|
old_cn_file = os.environ.get("SHUSUB2_ERRORS_CN_URL_FILE")
|
|
old_us_file = os.environ.get("SHUSUB2_ERRORS_US_URL_FILE")
|
|
os.environ["SHUSUB2_ERRORS_CN_URL_FILE"] = "/nonexistent/shusub2/errors-cn-url"
|
|
os.environ["SHUSUB2_ERRORS_US_URL_FILE"] = "/nonexistent/shusub2/errors-us-url"
|
|
try:
|
|
self.assertEqual(mod.default_errors_cn_url(), "https://sub2apicn.shujk.top/api/v1/admin/ops/errors")
|
|
self.assertEqual(
|
|
mod.default_errors_us_url(),
|
|
"https://us.sub2.shujk.top/api/v1/admin/ops/errors",
|
|
)
|
|
self.assertEqual(
|
|
mod.default_error_sources(mod.default_errors_cn_url(), mod.default_errors_us_url()),
|
|
[
|
|
("cn", "https://sub2apicn.shujk.top/api/v1/admin/ops/errors"),
|
|
("us", "https://us.sub2.shujk.top/api/v1/admin/ops/errors"),
|
|
],
|
|
)
|
|
finally:
|
|
if old_cn is not None:
|
|
os.environ["SHUSUB2_ERRORS_CN_URL"] = old_cn
|
|
if old_us is not None:
|
|
os.environ["SHUSUB2_ERRORS_US_URL"] = old_us
|
|
if old_cn_file is None:
|
|
os.environ.pop("SHUSUB2_ERRORS_CN_URL_FILE", None)
|
|
else:
|
|
os.environ["SHUSUB2_ERRORS_CN_URL_FILE"] = old_cn_file
|
|
if old_us_file is None:
|
|
os.environ.pop("SHUSUB2_ERRORS_US_URL_FILE", None)
|
|
else:
|
|
os.environ["SHUSUB2_ERRORS_US_URL_FILE"] = old_us_file
|
|
|
|
def test_once_errors_without_token_prints_hint_and_exits_2(self) -> None:
|
|
mod = load_module()
|
|
old_token = os.environ.pop("SHUSUB2_LOGS_TOKEN", None)
|
|
old_file = os.environ.get("SHUSUB2_LOGS_TOKEN_FILE")
|
|
os.environ["SHUSUB2_LOGS_TOKEN_FILE"] = "/nonexistent/shusub2/logs-token"
|
|
err = io.StringIO()
|
|
try:
|
|
with contextlib.redirect_stderr(err):
|
|
rc = mod.main(["--once", "--errors", "--no-version-check"])
|
|
finally:
|
|
if old_token is not None:
|
|
os.environ["SHUSUB2_LOGS_TOKEN"] = old_token
|
|
if old_file is None:
|
|
os.environ.pop("SHUSUB2_LOGS_TOKEN_FILE", None)
|
|
else:
|
|
os.environ["SHUSUB2_LOGS_TOKEN_FILE"] = old_file
|
|
|
|
self.assertEqual(rc, 2)
|
|
self.assertIn("logs token not configured", err.getvalue())
|
|
|
|
|
|
class DashboardLayoutTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_dashboard_supports_wide_columns_at_80_by_24_and_focuses_each(self) -> None:
|
|
from textual.app import App
|
|
|
|
mod = load_module()
|
|
mod.fetch_payload = lambda *args, **kwargs: {
|
|
"source_name": "test",
|
|
"totals": {"total_accounts": 1, "usable_accounts": 1, "today_cost_usd": 1},
|
|
"accounts": [
|
|
{
|
|
"id": 1,
|
|
"name": "account-one-with-a-complete-expanded-dashboard-name",
|
|
"routing_group": "fast",
|
|
"provider": "openai",
|
|
"kind": "pay_as_you_go",
|
|
"status": "ok",
|
|
}
|
|
],
|
|
}
|
|
mod.fetch_optional_payload = lambda *args, **kwargs: ({}, "")
|
|
mod.fetch_key_usage_payload = lambda *args, **kwargs: {
|
|
"date": "2026-07-24",
|
|
"trend": [{"api_key_id": 1, "key_name": "wmy", "requests": 3, "tokens": 300}],
|
|
"stats": {"1": {"today_actual_cost": 0.42}},
|
|
}
|
|
mod.fetch_logs_payload = lambda *args, **kwargs: {
|
|
"data": {
|
|
"total": 1,
|
|
"items": [
|
|
{
|
|
"id": 1,
|
|
"api_key_name": "wmy-production-key-with-a-complete-expanded-name",
|
|
"account_name": "account-one-with-a-complete-expanded-dashboard-name",
|
|
"model": "gpt-5.5-codex-with-a-complete-expanded-dashboard-model-name",
|
|
"input_tokens": 1200,
|
|
"output_tokens": 340,
|
|
"cache_creation_tokens": 50,
|
|
"cache_read_tokens": 4100,
|
|
"first_token_ms": 200,
|
|
"duration_ms": 1200,
|
|
"created_at": "2026-07-24T12:00:00+08:00",
|
|
}
|
|
],
|
|
}
|
|
}
|
|
mod.fetch_merged_errors = lambda *args, **kwargs: {
|
|
"items": [
|
|
{
|
|
"_node": "cn",
|
|
"id": 1,
|
|
"status_code": 500,
|
|
"api_key_name": "wmy-production-key-with-a-complete-expanded-name",
|
|
"account_name": "account-one-with-a-complete-expanded-dashboard-name",
|
|
"requested_model": "gpt-5.5-codex-with-a-complete-expanded-dashboard-model-name",
|
|
"phase": "upstream",
|
|
"type": "api_error",
|
|
"created_at": "2026-07-24T12:01:00+08:00",
|
|
}
|
|
],
|
|
"sources": {"cn": {"ok": True, "total": 1}},
|
|
}
|
|
|
|
captured = {}
|
|
original_run = App.run
|
|
App.run = lambda self, *args, **kwargs: captured.setdefault("app", self)
|
|
try:
|
|
rc = mod.run_textual(
|
|
"https://example/accounts",
|
|
"",
|
|
"https://example/usage",
|
|
"token",
|
|
"https://example/cn",
|
|
"https://example/us",
|
|
999,
|
|
999,
|
|
999,
|
|
1,
|
|
100,
|
|
100,
|
|
"24h",
|
|
)
|
|
finally:
|
|
App.run = original_run
|
|
|
|
self.assertEqual(rc, 0)
|
|
app = captured["app"]
|
|
async with app.run_test(size=(80, 24)) as pilot:
|
|
await pilot.pause()
|
|
await pilot.pause()
|
|
screen = app.screen
|
|
self.assertEqual(type(screen).__name__, "DashboardScreen")
|
|
self.assertEqual(screen.query_one("#filter").region.height, 1)
|
|
self.assertEqual(screen.query_one("#detail").region.height, 2)
|
|
tables = [screen.query_one(selector) for selector in ("#accounts", "#keys", "#logs", "#errors")]
|
|
for table in tables:
|
|
self.assertGreaterEqual(table.region.height, 4)
|
|
logs_table = screen.query_one("#logs")
|
|
errors_table = screen.query_one("#errors")
|
|
self.assertGreater(logs_table.virtual_size.width, logs_table.region.width)
|
|
self.assertGreater(errors_table.virtual_size.width, errors_table.region.width)
|
|
self.assertGreater(logs_table.max_scroll_x, 0)
|
|
self.assertGreater(errors_table.max_scroll_x, 0)
|
|
self.assertEqual(screen.log_rows[0]["cache_tokens"], 4150)
|
|
self.assertGreater(screen.log_rows[0]["tokens_per_second"], 0)
|
|
self.assertEqual(len({table.styles.background for table in tables}), 4)
|
|
self.assertIn("wmy", str(screen.query_one("#summary").render()))
|
|
for key, expected_id in (("k", "keys"), ("l", "logs"), ("e", "errors"), ("a", "accounts")):
|
|
await pilot.press(key)
|
|
await pilot.pause()
|
|
self.assertEqual(screen.focused.id, expected_id)
|
|
|
|
|
|
class PageSelectionTests(unittest.TestCase):
|
|
def test_dedicated_page_flags_are_mutually_exclusive(self) -> None:
|
|
mod = load_module()
|
|
old_token = os.environ.pop("SHUSUB2_LOGS_TOKEN", None)
|
|
old_file = os.environ.get("SHUSUB2_LOGS_TOKEN_FILE")
|
|
os.environ["SHUSUB2_LOGS_TOKEN_FILE"] = "/nonexistent/shusub2/logs-token"
|
|
err = io.StringIO()
|
|
try:
|
|
with contextlib.redirect_stderr(err):
|
|
rc = mod.main(["--accounts", "--logs", "--no-version-check"])
|
|
finally:
|
|
if old_token is not None:
|
|
os.environ["SHUSUB2_LOGS_TOKEN"] = old_token
|
|
if old_file is None:
|
|
os.environ.pop("SHUSUB2_LOGS_TOKEN_FILE", None)
|
|
else:
|
|
os.environ["SHUSUB2_LOGS_TOKEN_FILE"] = old_file
|
|
|
|
self.assertEqual(rc, 2)
|
|
self.assertIn("choose only one", err.getvalue())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|