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_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["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("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://sub2api.server4.shujk.top:19857/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://sub2api.server4.shujk.top:19857/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_fits_three_tables_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", "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": "key-one", "account_name": "account-one", "model": "gpt-5.5-codex", "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, "account_name": "account-one", "requested_model": "gpt-5.5-codex", "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) self.assertLessEqual(table.virtual_size.width, table.region.width) 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()