287 lines
12 KiB
Python
287 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import io
|
|
from pathlib import Path
|
|
import importlib.util
|
|
import sys
|
|
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_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_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())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|