shujakuin/shusub2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
492e6c882f5dffdadd794e538e03959f499fa3e3
shusub2/tests/test_payload.py
T

2005 lines
86 KiB
Python
Raw Blame History

from __future__ import annotations
import contextlib
import copy
import datetime as dt
import io
import json
import os
from pathlib import Path
import importlib.util
import sys
import tempfile
import threading
import time
import unittest
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from unittest import mock
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
def workspace_payload_fixture() -> dict[str, object]:
zone = dt.timezone(dt.timedelta(hours=8), name="Asia/Shanghai")
local_start = dt.datetime.combine(dt.datetime.now(zone).date(), dt.time.min, tzinfo=zone)
local_end = local_start + dt.timedelta(days=1)
return {
"service": "sub2api-pricing-monitor",
"view": "workspace",
"generated_at": "2026-08-03T12:00:00Z",
"state": {"ok": True, "partial": False},
"components": {
"accounts": {"ok": True, "stale": False},
"status": {"ok": True, "stale": False},
"traffic": {"ok": True, "stale": False},
},
"sources": [
{
"name": "code-plan",
"source_kind": "newapi",
"health_state": "healthy",
"balance_available": True,
"balance": {"available": 5000000, "available_cny": 10, "unit": "quota"},
}
],
"accounts": {
"source_name": "workspace",
"totals": {
"total_accounts": 1,
"usable_accounts": 1,
"today_cost_usd": 1.25,
"today_tokens": 300,
"today_requests": 2,
},
"accounts": [
{
"id": "9007199254740993",
"name": "oai-quota-code-plan",
"routing_group": "fast",
"provider": "openai",
"kind": "quota_limited",
"usable": True,
"account_rate_multiplier": 0.06,
"account_rate_multiplier_cny": 0.003,
}
],
},
"status": {
"channel_monitors": {
"items": [
{
"id": "9007199254740995",
"name": "oai-quota-code-plan",
"provider": "openai",
"latest_status": "success",
}
]
}
},
"traffic": {
"period_kind": "calendar_day",
"date": local_start.date().isoformat(),
"timezone": "Asia/Shanghai",
"started_at": local_start.isoformat(),
"ends_at": local_end.isoformat(),
"sample_limit": 1000,
"sample_limited": False,
"usage_rollups_full_day": True,
"partial": False,
"instances": [
{"instance": "server6", "ok": True, "error_total": 4},
{"instance": "server4", "ok": True, "error_total": 0},
],
"requests": [
{
"instance": "server6",
"id": "9007199254740997",
"created_at": "2026-08-03T11:59:00Z",
"api_key_id": "7",
"api_key_name": "wmy",
"account_id": "9",
"account_name": "oai-quota-code-plan",
"account_rate_multiplier": 0.06,
"account_rate_multiplier_cny": 0.003,
"model": "gpt-5.5",
"input_tokens": 100,
"output_tokens": 50,
"cache_read_tokens": 25,
"actual_cost": 0.25,
"duration_ms": 1000,
}
],
"errors": [
{
"instance": "server6",
"id": "21",
"created_at": "2026-08-03T11:58:00Z",
"status_code": 502,
"inbound_status_code": 500,
"upstream_status_code": 502,
"api_key_id": "7",
"api_key_name": "wmy",
"account_id": "9",
"account_name": "oai-quota-code-plan",
"model": "gpt-5.5",
"error_type": "api_error",
"error_source": "upstream_http",
}
],
"keys": [
{
"instance": "server6",
"api_key_id": "7",
"api_key_name": "wmy",
"status": "active",
"request_count": 2,
"token_count": 175,
"actual_cost": 0.25,
"latest_at": "2026-08-03T11:59:00Z",
}
],
},
}
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.assertNotIn("five_hour_balance", rows[1])
self.assertNotIn("weekly_balance", rows[1])
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_account_rows_attach_only_canonical_pricing_source_balances(self) -> None:
mod = load_module()
accounts_payload = {
"accounts": [
{"id": 1, "name": "oai-quota-code-plan", "routing_group": "fast", "kind": "quota_limited"},
{"id": 2, "name": "oai-quotaonly-code-plan", "routing_group": "fast", "kind": "quota_limited"},
{"id": 3, "name": "oai-sub-code-plan", "routing_group": "fast", "kind": "quota_limited"},
{"id": 4, "name": "oai-quota-unknown", "routing_group": "fast", "kind": "quota_limited"},
{"id": 5, "name": "oai-quota-ycy", "routing_group": "fast", "kind": "quota_limited"},
{"id": 6, "name": "oai-quota-code--plan", "routing_group": "fast", "kind": "quota_limited"},
{"id": 7, "name": "oai-quota-code-plan-", "routing_group": "fast", "kind": "quota_limited"},
]
}
pricing_payload = {
"sources": [
{
"name": "code-plan",
"source_kind": "newapi",
"health_state": "healthy",
"last_success_at": "2026-08-02T12:00:00Z",
"balance": {"available": 5000000, "available_cny": 10, "unit": "quota"},
},
{
"name": "ycy",
"source_kind": "newapi",
"health_state": "error",
"last_error": "upstream request failed",
"balance": {"available": 1000000, "available_cny": 2, "unit": "quota"},
},
]
}
rows = {row["name"]: row for row in mod.normalize_account_rows(accounts_payload, pricing_payload=pricing_payload)}
self.assertEqual(rows["oai-quota-code-plan"]["pricing_source"], "code-plan")
self.assertEqual(rows["oai-quota-code-plan"]["pricing_cny"], 10)
self.assertEqual(rows["oai-quotaonly-code-plan"]["pricing_source"], "code-plan")
self.assertEqual(rows["oai-sub-code-plan"]["pricing_source"], "-")
self.assertEqual(rows["oai-quota-unknown"]["pricing_source"], "-")
self.assertEqual(rows["oai-quota-ycy"]["pricing_source"], "ycy")
self.assertEqual(rows["oai-quota-ycy"]["pricing_state"], "error")
self.assertIsNone(rows["oai-quota-ycy"]["pricing_cny"])
self.assertEqual(rows["oai-quota-code--plan"]["pricing_source"], "-")
self.assertEqual(rows["oai-quota-code-plan-"]["pricing_source"], "-")
ambiguous = dict(pricing_payload, sources=[*pricing_payload["sources"], {"name": "codeplan", "health_state": "healthy", "balance": {"available_cny": 12}}])
ambiguous_rows = {row["name"]: row for row in mod.normalize_account_rows(accounts_payload, pricing_payload=ambiguous)}
self.assertEqual(ambiguous_rows["oai-quota-code-plan"]["pricing_source"], "-")
def test_normalize_pricing_rows_uses_pricing_monitor_source_balances(self) -> None:
mod = load_module()
payload = {
"generated_at": "2026-08-02T12:00:00Z",
"sources": [
{
"name": "code-plan",
"source_kind": "newapi",
"health_state": "healthy",
"last_success_at": "2026-08-02T11:59:00Z",
"balance_available": True,
"balance": {"available": 5000000, "available_cny": 10, "unit": "quota"},
},
{
"name": "kedaya",
"source_kind": "remote_monitor",
"health_state": "error",
"last_error": "upstream request failed",
"balance_available": False,
"balance": {"available": 99, "available_cny": 99, "unit": "balance"},
},
],
}
rows = mod.normalize_pricing_rows(payload)
self.assertEqual([row["name"] for row in rows], ["code-plan", "kedaya"])
self.assertEqual(rows[0]["balance"], 5000000)
self.assertEqual(rows[0]["balance_cny"], 10)
self.assertEqual(rows[0]["unit"], "quota")
self.assertEqual(rows[0]["status"], "healthy")
self.assertEqual(rows[1]["status"], "error")
self.assertIsNone(rows[1]["balance"])
self.assertIsNone(rows[1]["balance_cny"])
unavailable_rows = mod.normalize_pricing_rows(
{
"sources": [
{
"name": "unavailable",
"source_kind": "newapi",
"health_state": "healthy",
"balance_available": False,
"balance": {"available": 99, "available_cny": 99, "unit": "quota"},
}
]
}
)
self.assertEqual(unavailable_rows[0]["status"], "healthy")
self.assertIsNone(unavailable_rows[0]["balance"])
self.assertIsNone(unavailable_rows[0]["balance_cny"])
self.assertEqual([row["name"] for row in mod.normalize_pricing_rows(payload, "keday")], ["kedaya"])
self.assertEqual(mod.pricing_summary(payload, rows), "sources 1/2 healthy | CNY ¥10")
out = io.StringIO()
with contextlib.redirect_stdout(out):
mod.print_pricing_once(payload)
self.assertIn("code-plan", out.getvalue())
self.assertIn("5,000,000", out.getvalue())
self.assertIn("¥10", out.getvalue())
def test_default_pricing_url_reads_env_or_config_file(self) -> None:
mod = load_module()
with tempfile.TemporaryDirectory() as tmp:
config_file = Path(tmp) / "shusub2" / "pricing-url"
config_file.parent.mkdir(parents=True)
config_file.write_text("https://pricing.example.test/api/ui-data?view=accounts\n", encoding="utf-8")
old_url = os.environ.get("SHUSUB2_PRICING_URL")
old_file = os.environ.get("SHUSUB2_PRICING_URL_FILE")
os.environ["SHUSUB2_PRICING_URL_FILE"] = str(config_file)
os.environ.pop("SHUSUB2_PRICING_URL", None)
try:
self.assertEqual(
mod.default_pricing_url(),
"https://pricing.example.test/api/ui-data?view=accounts",
)
os.environ["SHUSUB2_PRICING_URL"] = "https://env.example.test/api/ui-data?view=accounts"
self.assertEqual(
mod.default_pricing_url(),
"https://env.example.test/api/ui-data?view=accounts",
)
finally:
if old_url is None:
os.environ.pop("SHUSUB2_PRICING_URL", None)
else:
os.environ["SHUSUB2_PRICING_URL"] = old_url
if old_file is None:
os.environ.pop("SHUSUB2_PRICING_URL_FILE", None)
else:
os.environ["SHUSUB2_PRICING_URL_FILE"] = old_file
def test_fetch_pricing_payload_requires_accounts_projection_shape(self) -> None:
mod = load_module()
original_fetch = mod.fetch_payload
valid_payload = {
"view": "accounts",
"sources": [
{
"name": "code-plan",
"source_kind": "newapi",
"health_state": "healthy",
"balance_available": True,
"balance": {"available": 1, "available_cny": 1, "unit": "quota"},
}
],
}
mod.fetch_payload = lambda *args, **kwargs: valid_payload
try:
self.assertEqual(
mod.fetch_pricing_payload("https://price.example/api/ui-data?view=accounts", 1),
valid_payload,
)
for invalid_payload in (
{"view": "overview", "sources": []},
{"view": "accounts"},
{"view": "accounts", "sources": [{"name": "code-plan"}]},
{
"view": "accounts",
"sources": [
{"name": "", "source_kind": "newapi", "health_state": "healthy", "balance_available": True, "balance": {}}
],
},
):
mod.fetch_payload = lambda *args, _payload=invalid_payload, **kwargs: _payload
with self.assertRaisesRegex(RuntimeError, "accounts source projection"):
mod.fetch_pricing_payload("https://price.example/api/ui-data?view=accounts", 1)
finally:
mod.fetch_payload = original_fetch
def test_optional_pricing_failure_is_nonblocking_and_redacted(self) -> None:
mod = load_module()
original_fetch = mod.fetch_pricing_payload
mod.fetch_pricing_payload = lambda *args, **kwargs: (_ for _ in ()).throw(
RuntimeError("https://pricing.example.test/?token=private")
)
try:
payload, error = mod.fetch_optional_pricing_payload(
"https://pricing.example.test/api/ui-data?view=accounts", 1
)
finally:
mod.fetch_pricing_payload = original_fetch
self.assertEqual(payload, {})
self.assertEqual(error, "unavailable")
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, size=-1):
return compressed if size < 0 else compressed[:size]
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, size=-1):
raw = mod.json.dumps(payload).encode("utf-8")
return raw if size < 0 else raw[:size]
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_json_decoder_bounds_plain_and_gzip_payloads(self) -> None:
mod = load_module()
class Response:
def __init__(self, raw: bytes, encoding: str = "") -> None:
self.raw = raw
self.headers = {"Content-Encoding": encoding} if encoding else {}
def read(self, size=-1):
return self.raw if size < 0 else self.raw[:size]
oversized = mod.json.dumps({"value": "x" * 256}).encode("utf-8")
with self.assertRaisesRegex(RuntimeError, "bounded"):
mod.decode_json_response(Response(oversized), "bounded response", maximum_bytes=64)
with self.assertRaisesRegex(RuntimeError, "bounded"):
mod.decode_json_response(
Response(mod.gzip.compress(oversized), "gzip"),
"bounded response",
maximum_bytes=64,
)
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.assertNotIn("5h bal", 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 WorkspaceTests(unittest.TestCase):
def test_workspace_adapters_preserve_ids_and_map_compact_traffic(self) -> None:
mod = load_module()
payload = workspace_payload_fixture()
payload["traffic"]["requests"].append(
{"instance": "server4", "id": "71", "created_at": "2026-08-03T11:57:00Z"}
)
payload["traffic"]["keys"].append(
{"instance": "server4", "api_key_id": "71", "api_key_name": "server4-copy"}
)
payload["traffic"]["errors"].append(
{
"instance": "server4",
"id": "22",
"created_at": "2026-08-03T11:57:00Z",
"status_code": 503,
"api_key_id": "8",
"account_id": "10",
"error_type": "upstream",
"error_source": "upstream_http",
"error_detail": "forbidden-detail",
"request_id": "forbidden-request-id",
"user_email": "forbidden@example.test",
}
)
accounts = mod.workspace_accounts_payload(payload)
status = mod.workspace_status_payload(payload)
pricing = mod.workspace_pricing_payload(payload)
logs = mod.workspace_logs_payload(payload)
keys = mod.workspace_keys_payload(payload)
errors = mod.workspace_errors_payload(payload)
self.assertEqual(accounts["accounts"][0]["id"], "9007199254740993")
self.assertEqual(status["channel_monitors"]["items"][0]["id"], "9007199254740995")
self.assertEqual(pricing["sources"][0]["name"], "code-plan")
self.assertEqual(len(logs["data"]["items"]), 1)
self.assertEqual(logs["data"]["items"][0]["id"], "9007199254740997")
self.assertEqual(logs["data"]["items"][0]["instance"], "server6")
normalized_accounts = mod.normalize_account_rows(accounts, pricing_payload=pricing)
normalized_logs = mod.normalize_log_rows(logs)
self.assertEqual(normalized_accounts[0]["id"], 9007199254740993)
self.assertEqual(normalized_accounts[0]["account_multiplier"], 0.003)
self.assertEqual(normalized_logs[0]["id"], 9007199254740997)
self.assertEqual(normalized_logs[0]["account_multiplier"], 0.003)
self.assertEqual(normalized_logs[0]["cost"], 0.25)
self.assertEqual(mod.as_int("9007199254740999"), 9007199254740999)
self.assertEqual(len(keys["items"]), 1)
self.assertEqual(mod.normalize_key_rows(keys)[0]["name"], "wmy")
self.assertEqual(mod.normalize_key_rows(keys)[0]["cost"], 0.25)
self.assertEqual(keys["period_kind"], "calendar_day")
self.assertEqual(keys["date"], payload["traffic"]["date"])
self.assertEqual(keys["timezone"], "Asia/Shanghai")
self.assertTrue(keys["usage_rollups_full_day"])
self.assertIn(f"today {keys['date']} (Asia/Shanghai)", mod.usage_period_label(keys))
self.assertEqual(errors["time_range"], f"today {keys['date']} (Asia/Shanghai)")
self.assertEqual({item["_node"] for item in errors["items"]}, {"server6", "server4"})
self.assertNotIn("forbidden-detail", json.dumps(errors))
self.assertNotIn("forbidden-request-id", json.dumps(errors))
self.assertNotIn("forbidden@example.test", json.dumps(errors))
self.assertEqual(errors["items"][0]["id"], "21")
self.assertEqual(errors["items"][0]["created_at"], "2026-08-03T11:58:00Z")
self.assertNotIn("error_count", errors["items"][0])
self.assertEqual(errors["items"][0]["phase"], "upstream_http")
self.assertEqual(errors["sources"]["server6"]["total"], 4)
self.assertEqual(errors["sources"]["server4"]["total"], 0)
self.assertEqual(
mod.workspace_state_summary(payload, "workspace unavailable"),
"workspace endpoint unavailable (using last good)",
)
fallback_error = copy.deepcopy(errors)
fallback_error["items"][0].pop("api_key_name", None)
fallback_error["items"][0].pop("account_name", None)
fallback_error["items"][0]["api_key_id"] = "9007199254740999"
fallback_error["items"][0]["account_id"] = "9007199254740998"
fallback_row = mod.normalize_error_rows(fallback_error)[0]
self.assertEqual(fallback_row["key"], "#9007199254740999")
self.assertEqual(fallback_row["account"], "#9007199254740998")
normalized_errors = mod.normalize_error_rows(errors)
self.assertNotIn("count", normalized_errors[0])
self.assertNotIn("| count ", mod.error_detail_line(normalized_errors[0]))
def test_workspace_payload_validation_rejects_nonfinite_and_oversized_rows(self) -> None:
mod = load_module()
nonfinite = workspace_payload_fixture()
nonfinite["traffic"]["requests"][0]["actual_cost"] = float("nan")
with mock.patch.object(mod, "fetch_workspace_response", return_value=(nonfinite, "")):
with self.assertRaisesRegex(RuntimeError, "invalid workspace projection"):
mod.fetch_workspace_payload("https://workspace.example.test/data", 3)
oversized = workspace_payload_fixture()
oversized["traffic"]["requests"] = [
{"id": str(index)} for index in range(mod.MAX_WORKSPACE_REQUESTS + 1)
]
with mock.patch.object(mod, "fetch_workspace_response", return_value=(oversized, "")):
with self.assertRaisesRegex(RuntimeError, "invalid workspace requests"):
mod.fetch_workspace_payload("https://workspace.example.test/data", 3)
invalid_period = workspace_payload_fixture()
invalid_period["traffic"].pop("ends_at")
with mock.patch.object(mod, "fetch_workspace_response", return_value=(invalid_period, "")):
with self.assertRaisesRegex(RuntimeError, "invalid workspace traffic period"):
mod.fetch_workspace_payload("https://workspace.example.test/data", 3)
utc_period = workspace_payload_fixture()
utc_start = dt.datetime.now(dt.timezone.utc).replace(
hour=0, minute=0, second=0, microsecond=0
)
utc_period["traffic"].update(
{
"date": utc_start.date().isoformat(),
"timezone": "UTC",
"started_at": utc_start.isoformat(),
"ends_at": (utc_start + dt.timedelta(days=1)).isoformat(),
}
)
with mock.patch.object(mod, "fetch_workspace_response", return_value=(utc_period, "")):
with self.assertRaisesRegex(RuntimeError, "invalid workspace traffic period"):
mod.fetch_workspace_payload("https://workspace.example.test/data", 3)
incomplete_usage = workspace_payload_fixture()
incomplete_usage["traffic"]["usage_rollups_full_day"] = False
with mock.patch.object(mod, "fetch_workspace_response", return_value=(incomplete_usage, "")):
with self.assertRaisesRegex(RuntimeError, "incomplete workspace traffic usage"):
mod.fetch_workspace_payload("https://workspace.example.test/data", 3)
noncanonical_request = workspace_payload_fixture()
noncanonical_request["traffic"]["requests"][0]["instance"] = "server4"
with mock.patch.object(mod, "fetch_workspace_response", return_value=(noncanonical_request, "")):
with self.assertRaisesRegex(RuntimeError, "noncanonical workspace requests"):
mod.fetch_workspace_payload("https://workspace.example.test/data", 3)
missing_key_provenance = workspace_payload_fixture()
missing_key_provenance["traffic"]["keys"][0].pop("instance")
with mock.patch.object(mod, "fetch_workspace_response", return_value=(missing_key_provenance, "")):
with self.assertRaisesRegex(RuntimeError, "noncanonical workspace keys"):
mod.fetch_workspace_payload("https://workspace.example.test/data", 3)
def test_workspace_conditional_response_sends_validator_and_accepts_bodyless_304(self) -> None:
mod = load_module()
validator = 'W/"workspace-v1"'
requests = []
class NotModifiedOpener:
def open(self, request, timeout):
requests.append((request, timeout))
raise mod.urllib.error.HTTPError(
request.full_url,
304,
"Not Modified",
{"ETag": validator},
None,
)
with mock.patch.object(mod.urllib.request, "build_opener", return_value=NotModifiedOpener()):
with self.assertRaises(mod.WorkspaceNotModified) as raised:
mod.fetch_workspace_response(
"https://workspace.example.test/api/ui-data?view=workspace",
3,
if_none_match=validator,
)
self.assertEqual(raised.exception.etag, validator)
self.assertEqual(len(requests), 1)
request, timeout = requests[0]
self.assertEqual(timeout, 3)
self.assertEqual(
request.full_url,
"https://workspace.example.test/api/ui-data?view=workspace",
)
self.assertEqual(request.get_header("Accept-encoding"), "gzip")
self.assertEqual(request.get_header("If-none-match"), validator)
with mock.patch.object(
mod,
"fetch_workspace_response",
return_value=(workspace_payload_fixture(), validator),
):
verified = mod.fetch_workspace_payload(
"https://workspace.example.test/api/ui-data?view=workspace", 3
)
self.assertEqual(verified.etag, validator)
def test_workspace_response_rejects_redirects_without_forwarding_validators(self) -> None:
mod = load_module()
validator = 'W/"workspace-v1"'
origin_validators: list[str] = []
redirected_validators: list[str] = []
class RedirectTargetHandler(BaseHTTPRequestHandler):
def do_GET(self) -> None: # noqa: N802
redirected_validators.append(self.headers.get("If-None-Match", ""))
self.send_response(200)
self.end_headers()
def log_message(self, format: str, *args: object) -> None: # noqa: A003
return
target = ThreadingHTTPServer(("127.0.0.1", 0), RedirectTargetHandler)
target_thread = threading.Thread(target=target.serve_forever, daemon=True)
target_thread.start()
self.addCleanup(target.server_close)
self.addCleanup(lambda: target_thread.join(timeout=2))
self.addCleanup(target.shutdown)
target_host, target_port = target.server_address[:2]
class RedirectOriginHandler(BaseHTTPRequestHandler):
def do_GET(self) -> None: # noqa: N802
origin_validators.append(self.headers.get("If-None-Match", ""))
self.send_response(302)
self.send_header("Location", f"http://{target_host}:{target_port}/redirected")
self.end_headers()
def log_message(self, format: str, *args: object) -> None: # noqa: A003
return
origin = ThreadingHTTPServer(("127.0.0.1", 0), RedirectOriginHandler)
origin_thread = threading.Thread(target=origin.serve_forever, daemon=True)
origin_thread.start()
self.addCleanup(origin.server_close)
self.addCleanup(lambda: origin_thread.join(timeout=2))
self.addCleanup(origin.shutdown)
origin_host, origin_port = origin.server_address[:2]
with self.assertRaises(mod.urllib.error.HTTPError) as redirected:
mod.fetch_workspace_response(
f"http://{origin_host}:{origin_port}/workspace",
3,
if_none_match=validator,
)
self.assertEqual(redirected.exception.code, 302)
self.assertEqual(origin_validators, [validator])
self.assertEqual(redirected_validators, [])
def test_workspace_cache_revalidates_current_payload_with_etag(self) -> None:
mod = load_module()
initial = workspace_payload_fixture()
changed = workspace_payload_fixture()
changed["accounts"]["totals"]["today_requests"] = 3
initial_validator = 'W/"workspace-v1"'
changed_validator = 'W/"workspace-v2"'
cache = mod.WorkspaceCache("https://workspace.example.test/api/ui-data?view=workspace", 3, 300)
with mock.patch.object(
mod,
"fetch_workspace_payload",
side_effect=[
mod.WorkspacePayload(initial, initial_validator),
mod.WorkspaceNotModified(initial_validator),
mod.WorkspacePayload(changed, changed_validator),
],
) as fetch:
first = cache.get()
revalidated = cache.get(force=True)
replaced = cache.get(force=True)
self.assertEqual(first, revalidated)
self.assertIsNot(first, revalidated)
self.assertEqual(replaced["accounts"]["totals"]["today_requests"], 3)
self.assertEqual(cache.etag, changed_validator)
self.assertEqual(cache.error, "")
self.assertEqual(cache.network_fetches, 3)
self.assertEqual(
fetch.call_args_list,
[
mock.call(cache.url, 3),
mock.call(cache.url, 3, if_none_match=initial_validator),
mock.call(cache.url, 3, if_none_match=initial_validator),
],
)
def test_workspace_cache_drops_validator_outside_current_day_and_fails_closed_on_304(self) -> None:
mod = load_module()
validator = 'W/"workspace-v1"'
cache = mod.WorkspaceCache("https://workspace.example.test/api/ui-data?view=workspace", 3, 300)
expired = workspace_payload_fixture()
zone = dt.timezone(dt.timedelta(hours=8), name="Asia/Shanghai")
local_start = dt.datetime.combine(
dt.datetime.now(zone).date() - dt.timedelta(days=1),
dt.time.min,
tzinfo=zone,
)
expired["traffic"].update(
{
"date": local_start.date().isoformat(),
"started_at": local_start.isoformat(),
"ends_at": (local_start + dt.timedelta(days=1)).isoformat(),
}
)
cache.payload = expired
cache.etag = validator
with mock.patch.object(
mod,
"fetch_workspace_payload",
return_value=mod.WorkspacePayload(workspace_payload_fixture(), ""),
) as fetch:
cache.get(force=True)
cache.get(force=True)
self.assertEqual(fetch.call_args_list, [mock.call(cache.url, 3), mock.call(cache.url, 3)])
self.assertEqual(cache.etag, "")
empty = mod.WorkspaceCache(cache.url, 3, 300)
empty.etag = validator
with mock.patch.object(
mod,
"fetch_workspace_payload",
side_effect=mod.WorkspaceNotModified(validator),
) as fetch:
with self.assertRaisesRegex(RuntimeError, "current-day traffic unavailable"):
empty.get(force=True)
fetch.assert_called_once_with(empty.url, 3)
self.assertEqual(empty.etag, "")
def test_workspace_cache_rechecks_current_day_after_network_completion(self) -> None:
mod = load_module()
validator = 'W/"workspace-v1"'
def cached_workspace() -> object:
cache = mod.WorkspaceCache(
"https://workspace.example.test/api/ui-data?view=workspace", 3, 300
)
cache.payload = workspace_payload_fixture()
cache.etag = validator
return cache
not_modified_cache = cached_workspace()
with (
mock.patch.object(
mod,
"fetch_workspace_payload",
side_effect=mod.WorkspaceNotModified(validator),
) as fetch,
mock.patch.object(mod, "workspace_traffic_is_current", side_effect=[True, False]),
):
with self.assertRaisesRegex(RuntimeError, "current-day traffic unavailable"):
not_modified_cache.get(force=True)
fetch.assert_called_once_with(
not_modified_cache.url, 3, if_none_match=validator
)
self.assertEqual(not_modified_cache.etag, "")
failed_cache = cached_workspace()
with (
mock.patch.object(
mod,
"fetch_workspace_payload",
side_effect=RuntimeError("private failure"),
) as fetch,
mock.patch.object(mod, "workspace_traffic_is_current", side_effect=[True, False]),
):
with self.assertRaisesRegex(RuntimeError, "workspace unavailable"):
failed_cache.get(force=True)
fetch.assert_called_once_with(failed_cache.url, 3, if_none_match=validator)
self.assertEqual(failed_cache.etag, "")
invalid = workspace_payload_fixture()
invalid["traffic"]["usage_rollups_full_day"] = False
invalid_cache = cached_workspace()
with (
mock.patch.object(
mod,
"fetch_workspace_response",
return_value=(invalid, 'W/"workspace-v2"'),
) as fetch,
mock.patch.object(mod, "workspace_traffic_is_current", side_effect=[True, False]),
):
with self.assertRaisesRegex(RuntimeError, "workspace unavailable"):
invalid_cache.get(force=True)
fetch.assert_called_once_with(invalid_cache.url, 3, if_none_match=validator)
self.assertEqual(invalid_cache.etag, "")
def test_workspace_cache_retains_last_good_validator_when_a_200_is_invalid(self) -> None:
mod = load_module()
validator = 'W/"workspace-v1"'
invalid = workspace_payload_fixture()
invalid["traffic"]["usage_rollups_full_day"] = False
cache = mod.WorkspaceCache("https://workspace.example.test/api/ui-data?view=workspace", 3, 300)
with mock.patch.object(
mod,
"fetch_workspace_response",
side_effect=[
(workspace_payload_fixture(), validator),
(invalid, 'W/"workspace-v2"'),
],
) as fetch:
first = cache.get()
stale = cache.get(force=True)
self.assertEqual(stale, first)
self.assertEqual(cache.etag, validator)
self.assertEqual(cache.error, "workspace unavailable")
self.assertEqual(
fetch.call_args_list[1],
mock.call(cache.url, 3, if_none_match=validator),
)
def test_workspace_cache_coalesces_reads_and_returns_copies(self) -> None:
mod = load_module()
calls: list[str] = []
call_lock = threading.Lock()
def fake_request(url: str, timeout: int):
with call_lock:
calls.append(url)
time.sleep(0.03)
return workspace_payload_fixture()
cache = mod.WorkspaceCache("https://workspace.example.test/api/ui-data?view=workspace", 3, 300)
results: list[dict[str, object]] = []
threads = [
threading.Thread(target=lambda: results.append(cache.get()))
for _ in range(12)
]
with mock.patch.object(mod, "fetch_workspace_payload", side_effect=fake_request):
for thread in threads:
thread.start()
for thread in threads:
thread.join(timeout=2)
self.assertEqual(len(calls), 1)
self.assertEqual(len(results), 12)
results[0]["view"] = "mutated"
self.assertEqual(cache.get()["view"], "workspace")
cache.get(force=True)
self.assertEqual(len(calls), 2)
def test_workspace_cache_rejects_previous_calendar_day(self) -> None:
mod = load_module()
self.assertEqual(
mod.usage_day_date(dt.datetime(2026, 8, 3, 16, tzinfo=dt.timezone.utc)),
"2026-08-04",
)
payload = workspace_payload_fixture()
yesterday = dt.datetime.now(dt.timezone.utc) - dt.timedelta(days=1)
local_start = yesterday.astimezone(dt.timezone(dt.timedelta(hours=8))).replace(
hour=0, minute=0, second=0, microsecond=0
)
local_end = local_start + dt.timedelta(days=1)
payload["traffic"].update(
{
"date": local_start.date().isoformat(),
"started_at": local_start.isoformat(),
"ends_at": local_end.isoformat(),
}
)
cache = mod.WorkspaceCache("https://workspace.example.test/data", 3, 300)
with mock.patch.object(mod, "fetch_workspace_payload", return_value=payload):
with self.assertRaisesRegex(RuntimeError, "current-day traffic unavailable"):
cache.get()
def test_workspace_cache_backs_off_failures_with_and_without_last_good(self) -> None:
mod = load_module()
cache = mod.WorkspaceCache("https://workspace.example.test/data", 3, 300)
with mock.patch.object(
mod,
"fetch_workspace_payload",
side_effect=[
workspace_payload_fixture(),
RuntimeError("private failure"),
workspace_payload_fixture(),
],
) as fetch:
first = cache.get()
stale = cache.get(force=True)
repeated = cache.get()
self.assertEqual(fetch.call_count, 2)
cache.last_attempt_at -= cache.retry_seconds + 1
recovered = cache.get()
self.assertEqual(first["view"], "workspace")
self.assertEqual(stale["view"], "workspace")
self.assertEqual(repeated["view"], "workspace")
self.assertEqual(recovered["view"], "workspace")
self.assertEqual(fetch.call_count, 3)
self.assertEqual(cache.network_fetches, 3)
self.assertEqual(cache.error, "")
empty = mod.WorkspaceCache("https://workspace.example.test/data", 3, 300)
with mock.patch.object(mod, "fetch_workspace_payload", side_effect=RuntimeError("private failure")) as fetch:
with self.assertRaisesRegex(RuntimeError, "workspace unavailable"):
empty.get()
with self.assertRaisesRegex(RuntimeError, "workspace unavailable"):
empty.get()
fetch.assert_called_once()
def test_default_main_uses_one_workspace_request_without_loading_admin_token(self) -> None:
mod = load_module()
out = io.StringIO()
with (
mock.patch.object(mod, "fetch_workspace_payload", return_value=workspace_payload_fixture()) as workspace_fetch,
mock.patch.object(mod, "default_logs_token", side_effect=AssertionError("admin token must not be read")) as token_read,
mock.patch.object(mod, "fetch_payload", side_effect=AssertionError("legacy accounts must not be read")),
mock.patch.object(mod, "fetch_optional_payload", side_effect=AssertionError("legacy status must not be read")),
mock.patch.object(mod, "fetch_optional_pricing_payload", side_effect=AssertionError("legacy pricing must not be read")),
contextlib.redirect_stdout(out),
):
rc = mod.main(
[
"--once",
"--workspace-url",
"https://workspace.example.test/api/ui-data?view=workspace",
"--no-version-check",
]
)
self.assertEqual(rc, 0)
workspace_fetch.assert_called_once_with(
"https://workspace.example.test/api/ui-data?view=workspace", 10
)
token_read.assert_not_called()
self.assertIn("oai-quota-code-plan", out.getvalue())
self.assertIn("wmy", out.getvalue())
def test_empty_workspace_url_falls_back_to_legacy_direct_mode(self) -> None:
mod = load_module()
accounts = workspace_payload_fixture()["accounts"]
with (
mock.patch.object(mod, "default_logs_token", return_value="") as token_read,
mock.patch.object(mod, "fetch_payload", return_value=accounts) as account_fetch,
mock.patch.object(mod, "fetch_optional_payload", return_value=({}, "")),
mock.patch.object(mod, "fetch_optional_pricing_payload", return_value=({"sources": []}, "")),
mock.patch.object(mod, "fetch_workspace_payload", side_effect=AssertionError("workspace must not be read")),
contextlib.redirect_stdout(io.StringIO()),
):
rc = mod.main(["--once", "--workspace-url", "", "--no-version-check"])
self.assertEqual(rc, 0)
token_read.assert_called_once_with()
account_fetch.assert_called_once()
def test_sources_and_requests_flags_are_aliases(self) -> None:
mod = load_module()
self.assertTrue(mod.build_parser().parse_args(["--sources"]).pricing)
self.assertTrue(mod.build_parser().parse_args(["--requests"]).logs)
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", "--legacy-direct", "--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"), [])
unsafe = {
"items": [
{
"id": 3,
"_node": "server6",
"created_at": "2026-07-24T12:00:00+08:00",
"status_code": 502,
"api_key_name": "safe-key",
"account_name": "safe-account",
"error_type": "upstream",
"error_source": "upstream_http",
"error_detail": "private detail token=redacted-test-value",
"message": "private message",
"request_id": "private-request-id",
"client_request_id": "private-client-request-id",
"user_email": "private@example.test",
}
]
}
safe_row = mod.normalize_error_rows(unsafe)[0]
serialized = json.dumps(safe_row)
for private_value in (
"redacted-test-value",
"private message",
"private-request-id",
"private-client-request-id",
"private@example.test",
):
self.assertNotIn(private_value, serialized)
self.assertEqual(safe_row["message"], "")
self.assertEqual(safe_row["request_id"], "")
self.assertEqual(safe_row["user"], "-")
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.assertEqual(payload["sources"]["us"]["error"], "unavailable")
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", "--legacy-direct", "--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": "oai-quota-code-plan",
"routing_group": "fast",
"provider": "openai",
"kind": "pay_as_you_go",
"status": "ok",
}
],
}
mod.fetch_optional_payload = lambda *args, **kwargs: ({}, "")
mod.fetch_pricing_payload = lambda *args, **kwargs: {
"generated_at": "2026-07-24T12:00:00+08:00",
"sources": [
{
"name": "code-plan",
"source_kind": "newapi",
"health_state": "healthy",
"last_success_at": "2026-07-24T12:00:00+08:00",
"balance_available": True,
"balance": {"available": 5000000, "available_cny": 10, "unit": "quota"},
}
],
}
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/pricing?view=accounts",
"https://example/usage",
"token",
"https://example/cn",
"https://example/us",
999,
999,
999,
1,
100,
100,
"24h",
legacy_direct=True,
)
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.assertEqual(screen.account_rows[0]["pricing_source"], "code-plan")
self.assertEqual(screen.account_rows[0]["pricing_cny"], 10)
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)
await pilot.press("p")
await pilot.pause()
self.assertEqual(type(app.screen).__name__, "PricingScreen")
self.assertEqual(app.screen.focused.id, "pricing")
self.assertEqual(app.screen.rows[0]["name"], "code-plan")
async def test_workspace_dashboard_handles_initial_endpoint_failure(self) -> None:
from textual.app import App
mod = load_module()
captured: dict[str, object] = {}
original_run = App.run
App.run = lambda self, *args, **kwargs: captured.setdefault("app", self)
try:
rc = mod.run_textual(
"",
"",
"",
"",
"",
"",
"",
999,
999,
999,
1,
100,
100,
"24h",
workspace_url="https://workspace.example.test/api/ui-data?view=workspace",
legacy_direct=False,
)
finally:
App.run = original_run
self.assertEqual(rc, 0)
app = captured["app"]
with mock.patch.object(
mod,
"fetch_workspace_payload",
side_effect=RuntimeError("private endpoint failure"),
) as fetch:
async with app.run_test(size=(100, 30)) as pilot:
await pilot.pause()
await pilot.pause()
self.assertEqual(type(app.screen).__name__, "DashboardScreen")
self.assertIn("workspace unavailable", str(app.screen.query_one("#status").render()))
self.assertEqual(fetch.call_count, 1)
async def test_workspace_dashboard_uses_one_snapshot_and_has_return_navigation(self) -> None:
from textual.app import App
mod = load_module()
captured: dict[str, object] = {}
original_run = App.run
App.run = lambda self, *args, **kwargs: captured.setdefault("app", self)
try:
rc = mod.run_textual(
"",
"",
"",
"",
"",
"",
"",
999,
999,
999,
1,
100,
100,
"24h",
workspace_url="https://workspace.example.test/api/ui-data?view=workspace",
legacy_direct=False,
)
finally:
App.run = original_run
self.assertEqual(rc, 0)
app = captured["app"]
with mock.patch.object(mod, "fetch_workspace_payload", return_value=workspace_payload_fixture()) as fetch:
async with app.run_test(size=(100, 30)) as pilot:
await pilot.pause()
await pilot.pause()
screen = app.screen
self.assertEqual(type(screen).__name__, "DashboardScreen")
self.assertEqual(app.sub_title, "Dashboard")
self.assertEqual(screen.account_rows[0]["name"], "oai-quota-code-plan")
self.assertEqual(screen.error_rows[0]["key"], "wmy")
self.assertNotIn("count", screen.error_rows[0])
self.assertEqual(len(fetch.call_args_list), 1)
await pilot.press("p")
await pilot.pause()
self.assertEqual(type(app.screen).__name__, "PricingScreen")
self.assertEqual(app.sub_title, "Sources")
await pilot.press("d")
await pilot.pause()
self.assertEqual(type(app.screen).__name__, "DashboardScreen")
self.assertEqual(app.sub_title, "Dashboard")
self.assertEqual(len(fetch.call_args_list), 1)
class PageSelectionTests(unittest.TestCase):
def test_once_pricing_prints_pricing_monitor_sources(self) -> None:
mod = load_module()
original_fetch = mod.fetch_pricing_payload
mod.fetch_pricing_payload = lambda *args, **kwargs: {
"generated_at": "2026-08-02T12:00:00Z",
"sources": [
{
"name": "code-plan",
"source_kind": "newapi",
"health_state": "healthy",
"last_success_at": "2026-08-02T11:59:00Z",
"balance": {"available": 5000000, "available_cny": 10, "unit": "quota"},
}
],
}
out = io.StringIO()
try:
with contextlib.redirect_stdout(out):
rc = mod.main(["--once", "--pricing", "--legacy-direct", "--no-version-check"])
finally:
mod.fetch_pricing_payload = original_fetch
self.assertEqual(rc, 0)
self.assertIn("code-plan", out.getvalue())
self.assertIn("¥10", out.getvalue())
def test_once_pricing_failure_is_redacted(self) -> None:
mod = load_module()
original_fetch = mod.fetch_pricing_payload
mod.fetch_pricing_payload = lambda *args, **kwargs: (_ for _ in ()).throw(
RuntimeError("https://price.example.test/?token=private")
)
err = io.StringIO()
try:
with contextlib.redirect_stderr(err):
rc = mod.main(["--once", "--pricing", "--legacy-direct", "--no-version-check"])
finally:
mod.fetch_pricing_payload = original_fetch
self.assertEqual(rc, 1)
self.assertEqual(err.getvalue().strip(), "sources unavailable")
def test_once_accounts_includes_only_mapped_pricing_source_balance(self) -> None:
mod = load_module()
original_fetch = mod.fetch_payload
original_optional = mod.fetch_optional_payload
original_pricing_optional = mod.fetch_optional_pricing_payload
accounts_payload = {
"source_name": "test",
"totals": {"total_accounts": 2, "usable_accounts": 2, "today_cost_usd": 0},
"accounts": [
{"id": 1, "name": "oai-quota-code-plan", "routing_group": "fast", "kind": "quota_limited"},
{"id": 2, "name": "oai-sub-code-plan", "routing_group": "fast", "kind": "quota_limited"},
],
}
pricing_payload = {
"sources": [
{
"name": "code-plan",
"source_kind": "newapi",
"health_state": "healthy",
"last_success_at": "2026-08-02T12:00:00Z",
"balance": {"available": 5000000, "available_cny": 10, "unit": "quota"},
}
]
}
mod.fetch_payload = lambda *args, **kwargs: accounts_payload
mod.fetch_optional_payload = lambda *args, **kwargs: ({}, "")
mod.fetch_optional_pricing_payload = lambda *args, **kwargs: (pricing_payload, "")
out = io.StringIO()
try:
with contextlib.redirect_stdout(out):
rc = mod.main(
[
"--legacy-direct",
"--once",
"--api-url",
"https://accounts.example.test/api/tui/accounts",
"--pricing-url",
"https://pricing.example.test/api/ui-data?view=accounts",
"--no-version-check",
]
)
finally:
mod.fetch_payload = original_fetch
mod.fetch_optional_payload = original_optional
mod.fetch_optional_pricing_payload = original_pricing_optional
text = out.getvalue()
self.assertEqual(rc, 0)
self.assertIn("src cny", text)
self.assertIn("oai-quota-code-plan", text)
self.assertIn("oai-sub-code-plan", text)
self.assertGreaterEqual(text.count("code-plan"), 2)
self.assertIn("¥10", text)
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()
Reference in New Issue View Git Blame Copy Permalink