From 69f72c09b515b48b122f790fa6f21d3357b7a4bb Mon Sep 17 00:00:00 2001 From: yunyaozhou Date: Mon, 27 Jul 2026 03:26:32 +0800 Subject: [PATCH] feat: add compact unified dashboard --- README.md | 71 +++++--- pyproject.toml | 2 +- sub2api_quota_tui.py | 373 ++++++++++++++++++++++++++++++++++++++++-- tests/test_payload.py | 114 +++++++++++++ uv.lock | 2 +- 5 files changed, 518 insertions(+), 44 deletions(-) diff --git a/README.md b/README.md index ec72928..f624551 100644 --- a/README.md +++ b/README.md @@ -7,17 +7,17 @@ - 部署模型：无 - 对应服务：无 - 对应 stack：无 -- Secret 边界：账号页无 secret；请求日志页可选持有 Sub2API admin API key（仅存本机 `~/.config/shusub2/logs-token`，0600） +- Secret 边界：本地用户配置 - 分发方式：`uvx` / `uv tool install`。 - 包名：`shusub2` - 命令名：`shusub2` - 本地状态路径：`~/.config/shusub2/` -- 目标 API：`cliproxy-codex-quota` 的 `/api/tui/accounts`、可选 `sub2api-status` `/api/status`、可选 Sub2API `/api/v1/admin/usage`（请求日志页，默认 `sub2apicn.shujk.top`）。 -- 备注：账号页不需要 SSH、数据库访问、API key、OAuth credentials 或 plaintext env；请求日志页需要 Sub2API admin API key。 +- 目标 API：`cliproxy-codex-quota` 的 `/api/tui/accounts`、可选 `sub2api-status` `/api/status`，以及可选 Sub2API `/api/v1/admin/usage` 和 `/api/v1/admin/ops/errors`。 +- 备注：账号数据无 secret，不需要 SSH、数据库访问、API key、OAuth credentials 或 plaintext env；请求与错误日志可选持有 Sub2API admin API key，仅存本机 `~/.config/shusub2/logs-token`（0600）或 `SHUSUB2_LOGS_TOKEN`。 -Terminal UI for the token-safe Sub2API account feed exposed by -`cliproxy-codex-quota`, plus an optional request-logs page backed by the -Sub2API admin usage API. +Single-page terminal dashboard for the token-safe Sub2API account feed, +request logs, and merged cn/us error logs. Dedicated full-table views remain +available for each data set. ## Quick Start @@ -79,10 +79,29 @@ Environment variables override the config file: - `SHUSUB2_VERSION_CHECK_TIMEOUT` - `SHUSUB2_NO_VERSION_CHECK` +## Unified Dashboard + +`shusub2` opens a single dashboard with Accounts, Logs, and Errors tables +stacked vertically. Press `a`, `l`, or `e` to focus a table, `/` to filter all +three tables, and `r` to refresh all data. The two-line detail area follows the +selected row. + +The dashboard uses compact columns sized to fit an 80-column terminal without +horizontal scrolling: + +```text +ACCOUNT | Group | Today | Daily | 5h | 7d | Avail +LOG KEY | Account | Model | Cost | Latency | Time +ERR | Status | Account | Model | Phase | Type | Time +``` + +Use `--accounts`, `--logs`, or `--errors` to start a dedicated full-table view. +The corresponding `--once` form still prints only that data set. + ## Request Logs Page -Press `l` on the accounts page to open the request logs page; press `a` to go -back. `shusub2 --logs` starts directly on the logs page, and +The middle dashboard table shows request logs, and `l` focuses it. +`shusub2 --logs` starts the dedicated logs page, while `shusub2 --once --logs` prints one logs snapshot to stdout. The logs page reads the latest requests (default 100, `--logs-limit`) from the @@ -120,16 +139,15 @@ chmod 600 ~/.config/shusub2/logs-token ``` or via `SHUSUB2_LOGS_TOKEN`. The key is sent as the `x-api-key` header and is -never written anywhere else by the client. Without a token the logs page stays -empty and shows a configuration hint; the accounts page keeps working without -any secret. - +never written anywhere else by the client. Without a token the dashboard Logs +and Errors tables stay empty and show a configuration hint; the Accounts table +keeps working without any secret. ## Errors Page -Press `e` from the accounts or logs page to open the merged errors page; press -`a` / `l` to jump back. `shusub2 --errors` starts directly on the errors page, -and `shusub2 --once --errors` prints one merged snapshot to stdout. +The bottom dashboard table shows merged errors, and `e` focuses it. +`shusub2 --errors` starts the dedicated errors page, while +`shusub2 --once --errors` prints one merged snapshot to stdout. The page pulls the latest admin ops error logs from **both** cn and us in parallel, then merges them by `created_at`: @@ -141,10 +159,9 @@ parallel, then merges them by `created_at`: backend, but Cloudflare often returns 1010 for non-browser clients, so the TUI defaults to the fixed server4 origin while still labeling rows as `us`. -While the errors page is active it refreshes every 60 seconds by default -(`--errors-refresh-seconds` / `SHUSUB2_ERRORS_REFRESH_SECONDS`). Leaving the -page stops that interval. Query window defaults to `24h` -(`--errors-time-range` / `SHUSUB2_ERRORS_TIME_RANGE`); each source uses +The dashboard and dedicated errors page refresh every 60 seconds by default +(`--errors-refresh-seconds` / `SHUSUB2_ERRORS_REFRESH_SECONDS`). Query window +defaults to `24h` (`--errors-time-range` / `SHUSUB2_ERRORS_TIME_RANGE`); each source uses `page=1` and `page_size` from `--errors-limit` (default 100, max 500). Columns: @@ -177,15 +194,19 @@ Configuration: - `--logs-token` / `SHUSUB2_LOGS_TOKEN` - `--logs-refresh-seconds` / `SHUSUB2_LOGS_REFRESH_SECONDS` - `--logs-limit` / `SHUSUB2_LOGS_LIMIT` +- `--errors-refresh-seconds` / `SHUSUB2_ERRORS_REFRESH_SECONDS` +- `--errors-limit` / `SHUSUB2_ERRORS_LIMIT` +- `--errors-time-range` / `SHUSUB2_ERRORS_TIME_RANGE` +- `--accounts` / `--logs` / `--errors` - `--save-config` - `--install` - `--refresh-seconds` / `SUB2API_QUOTA_TUI_REFRESH_SECONDS` - `--timeout` / `SUB2API_QUOTA_TUI_TIMEOUT` -The TUI reads `/api/tui/accounts` and can optionally read `sub2api-status` -`/api/status` for channel monitor health. The accounts page does not need SSH, +The dashboard reads `/api/tui/accounts` and can optionally read `sub2api-status` +`/api/status` for channel monitor health. The Accounts table does not need SSH, database access, API keys, OAuth credentials, or plaintext env files. The -optional request logs page authenticates to the Sub2API admin usage API with +Logs and Errors tables authenticate to the Sub2API admin APIs with an admin API key stored only in `~/.config/shusub2/logs-token` (0600) or `SHUSUB2_LOGS_TOKEN`. @@ -195,15 +216,15 @@ automatically enables monitor availability. On startup it also checks the public Gitea repo for a newer package version and prints a short upgrade hint when one is available. -Below the accounts table, a per-key panel shows today's usage for each -API key (`Key | Today | Tokens | Req`, sorted by cost, key names in the +On the dedicated `--accounts` page, a per-key panel shows today's usage for +each API key (`Key | Today | Tokens | Req`, sorted by cost, key names in the same per-key colors as the logs page). It combines the Sub2API admin `dashboard/api-keys-trend` and `dashboard/api-keys-usage` endpoints and needs the same admin API key as the logs page; without a token the panel stays empty and the status line shows a hint. `--once` prints the same panel after the accounts table when a token is configured. -Accounts page columns are ordered for scanning inside zellij: +The dedicated Accounts page keeps the full columns for scanning inside zellij: ```text Name | Provider | Group | Daily | Today | Tokens | Req | Kind | 5h | 7d | Reset | Status | Availability diff --git a/pyproject.toml b/pyproject.toml index 725ec41..e0a8c4d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "shusub2" -version = "0.2.4" +version = "0.2.5" description = "Terminal UI for Sub2API account quota and daily usage" readme = "README.md" requires-python = ">=3.11" diff --git a/sub2api_quota_tui.py b/sub2api_quota_tui.py index 79d195e..4dbd489 100644 --- a/sub2api_quota_tui.py +++ b/sub2api_quota_tui.py @@ -19,7 +19,7 @@ from typing import Any APP_NAME = "shusub2" -FALLBACK_VERSION = "0.2.4" +FALLBACK_VERSION = "0.2.5" DEFAULT_API_URL = "http://127.0.0.1:18318/api/tui/accounts" DEFAULT_CONFIG_FILE = "~/.config/shusub2/api-url" DEFAULT_STATUS_CONFIG_FILE = "~/.config/shusub2/status-url" @@ -1172,7 +1172,7 @@ def run_textual( errors_limit: int, errors_time_range: str, version_message: str = "", - start_page: str = "accounts", + start_page: str = "dashboard", ) -> int: try: from rich.text import Text @@ -1183,6 +1183,336 @@ def run_textual( print("Textual is required. Run with: uv run --with textual python sub2api_quota_tui.py", file=sys.stderr) return 2 + class DashboardScreen(Screen[None]): + AUTO_FOCUS = "#accounts" + BINDINGS = [ + ("r", "refresh", "Refresh"), + ("/", "focus_filter", "Filter"), + ("a", "focus_accounts", "Accounts"), + ("l", "focus_logs", "Logs"), + ("e", "focus_errors", "Errors"), + ] + + def __init__(self) -> None: + super().__init__() + self.accounts_payload: dict[str, Any] = {} + self.status_payload: dict[str, Any] = {} + self.monitor_error = "" + self.accounts_error = "" + self.logs_payload: dict[str, Any] = {} + self.logs_error = "" + self.errors_payload: dict[str, Any] = {} + self.errors_error = "" + self.account_rows: list[dict[str, Any]] = [] + self.log_rows: list[dict[str, Any]] = [] + self.error_rows: list[dict[str, Any]] = [] + self.account_by_key: dict[str, dict[str, Any]] = {} + self.log_by_key: dict[str, dict[str, Any]] = {} + self.error_by_key: dict[str, dict[str, Any]] = {} + self.active_table = "accounts" + + def compose(self) -> ComposeResult: + yield Header(show_clock=True) + yield Static("", id="summary") + yield Input(placeholder="filter", id="filter") + yield DataTable(id="accounts") + yield DataTable(id="logs") + yield DataTable(id="errors") + yield Static("", id="detail") + yield Static("", id="status") + yield Footer() + + @staticmethod + def configure_table(table: DataTable, columns: tuple[tuple[str, int], ...]) -> None: + table.cursor_type = "row" + table.zebra_stripes = True + for label, width in columns: + table.add_column(label, width=width) + + def on_mount(self) -> None: + self.configure_table( + self.query_one("#accounts", DataTable), + ( + ("ACCOUNT", 16), + ("Group", 5), + ("Today", 8), + ("Daily", 10), + ("5h", 9), + ("7d", 9), + ("Avail", 7), + ), + ) + self.configure_table( + self.query_one("#logs", DataTable), + ( + ("LOG KEY", 12), + ("Account", 12), + ("Model", 16), + ("Cost", 8), + ("Latency", 7), + ("Time", 11), + ), + ) + self.configure_table( + self.query_one("#errors", DataTable), + ( + ("ERR", 4), + ("Status", 6), + ("Account", 12), + ("Model", 13), + ("Phase", 8), + ("Type", 10), + ("Time", 11), + ), + ) + self.refresh_all(refresh=True) + self.set_interval(refresh_seconds, self.refresh_accounts) + self.set_interval(logs_refresh_seconds, self.refresh_logs) + self.set_interval(errors_refresh_seconds, self.refresh_errors) + self.focus_table("accounts") + + def action_refresh(self) -> None: + self.refresh_all(refresh=True) + + def action_focus_filter(self) -> None: + self.query_one("#filter", Input).focus() + + def action_focus_accounts(self) -> None: + self.focus_table("accounts") + + def action_focus_logs(self) -> None: + self.focus_table("logs") + + def action_focus_errors(self) -> None: + self.focus_table("errors") + + def focus_table(self, table_id: str) -> None: + self.active_table = table_id + self.query_one(f"#{table_id}", DataTable).focus() + self.render_active_detail() + + def on_input_changed(self, event: Input.Changed) -> None: + if event.input.id == "filter": + self.render_tables() + + def on_input_submitted(self, event: Input.Submitted) -> None: + if event.input.id == "filter": + self.focus_table(self.active_table) + + def refresh_all(self, refresh: bool = False) -> None: + self.query_one("#status", Static).update("refreshing accounts, logs, and errors...") + self.refresh_accounts(refresh=refresh) + self.refresh_logs() + self.refresh_errors() + + def refresh_accounts(self, refresh: bool = False) -> None: + self.accounts_error = "" + try: + self.accounts_payload = fetch_payload(api_url, timeout, refresh=refresh) + self.status_payload, self.monitor_error = fetch_optional_payload(status_url, timeout) + except Exception as exc: + self.accounts_error = str(exc) + self.render_accounts() + self.render_meta() + + def refresh_logs(self) -> None: + self.logs_error = "" + if not str(logs_token or "").strip(): + self.logs_payload = {} + self.logs_error = "admin token not configured" + else: + try: + self.logs_payload = fetch_logs_payload(logs_url, logs_token, timeout, logs_limit) + except Exception as exc: + self.logs_error = str(exc) + self.render_logs() + self.render_meta() + + def refresh_errors(self) -> None: + self.errors_error = "" + if not str(logs_token or "").strip(): + self.errors_payload = {} + self.errors_error = "admin token not configured" + else: + try: + self.errors_payload = fetch_merged_errors( + default_error_sources(errors_cn_url, errors_us_url), + logs_token, + timeout, + errors_limit, + errors_time_range, + ) + except Exception as exc: + self.errors_error = str(exc) + self.render_errors() + self.render_meta() + + def render_tables(self) -> None: + self.render_accounts() + self.render_logs() + self.render_errors() + self.render_meta() + self.render_active_detail() + + def render_accounts(self) -> None: + filter_text = self.query_one("#filter", Input).value + self.account_rows = normalize_account_rows(self.accounts_payload, filter_text) + table = self.query_one("#accounts", DataTable) + table.clear() + self.account_by_key = {} + for index, row in enumerate(self.account_rows): + key = f"account-{row['id']}-{index}" + self.account_by_key[key] = row + table.add_row( + row["name"], + row["routing_group"], + format_cost(row["today_cost_usd"]), + row["daily_quota_cell"], + row["five_hour"], + row["weekly"], + monitor_availability(row, self.status_payload), + key=key, + ) + + def render_logs(self) -> None: + filter_text = self.query_one("#filter", Input).value + self.log_rows = normalize_log_rows(self.logs_payload, filter_text) + table = self.query_one("#logs", DataTable) + table.clear() + self.log_by_key = {} + for index, row in enumerate(self.log_rows): + key = f"log-{row['id']}-{index}" + self.log_by_key[key] = row + color = key_color(row["key"]) + table.add_row( + Text(row["key"], style=color) if color else row["key"], + row["account"], + row["model"], + format_cost(row["cost"]), + format_seconds(row["duration_ms"]), + row["time"], + key=key, + ) + + def render_errors(self) -> None: + filter_text = self.query_one("#filter", Input).value + self.error_rows = normalize_error_rows(self.errors_payload, filter_text) + table = self.query_one("#errors", DataTable) + table.clear() + self.error_by_key = {} + for index, row in enumerate(self.error_rows): + key = f"error-{row['node']}-{row['id']}-{index}" + self.error_by_key[key] = row + table.add_row( + row["node"], + str(row["status_code"]), + row["account"], + row["model"], + row["phase"], + row["type"], + row["time"], + key=key, + ) + + def errors_total(self) -> int: + sources = self.errors_payload.get("sources") if isinstance(self.errors_payload.get("sources"), dict) else {} + totals = [ + as_int(info.get("total")) + for info in sources.values() + if isinstance(info, dict) and info.get("ok") + ] + return sum(totals) if totals else len(self.errors_payload.get("items") or []) + + def render_meta(self) -> None: + totals = self.accounts_payload.get("totals") if isinstance(self.accounts_payload.get("totals"), dict) else {} + account_total = as_int(totals.get("total_accounts")) or len(self.accounts_payload.get("accounts") or []) + usable = as_int(totals.get("usable_accounts")) + logs_total = as_int(logs_envelope(self.logs_payload).get("total")) + if not logs_total: + logs_total = len(log_items(self.logs_payload)) + summary = ( + f"accounts {len(self.account_rows)}/{account_total} ({usable} usable, " + f"{format_cost(totals.get('today_cost_usd'))}) | " + f"logs {len(self.log_rows)}/{logs_total} | " + f"errors {len(self.error_rows)}/{self.errors_total()}" + ) + self.query_one("#summary", Static).update(summary) + + status_bits = [version_message] + if self.accounts_error: + status_bits.append(f"accounts: {self.accounts_error}") + else: + status_bits.extend( + ( + monitor_summary(self.status_payload, self.monitor_error), + f"source {self.accounts_payload.get('source_name') or '-'}", + ) + ) + shared_admin_error = self.logs_error and self.logs_error == self.errors_error + if shared_admin_error: + status_bits.append(f"logs/errors: {self.logs_error}") + else: + if self.logs_error: + status_bits.append(f"logs: {self.logs_error}") + if self.errors_error: + status_bits.append(f"errors: {self.errors_error}") + if not self.errors_error and self.errors_payload: + sources = self.errors_payload.get("sources") if isinstance(self.errors_payload.get("sources"), dict) else {} + source_bits = [] + for node in ("cn", "us"): + info = sources.get(node) if isinstance(sources.get(node), dict) else {} + if info: + source_bits.append(f"{node} {'ok' if info.get('ok') else 'err'}") + if source_bits: + status_bits.append("errors " + "/".join(source_bits)) + self.query_one("#status", Static).update(" | ".join(bit for bit in status_bits if bit) or "ready") + + def render_active_detail(self) -> None: + rows_by_table = { + "accounts": self.account_rows, + "logs": self.log_rows, + "errors": self.error_rows, + } + rows = rows_by_table[self.active_table] + if not rows: + self.query_one("#detail", Static).update(f"no {self.active_table}") + return + self.render_detail(self.active_table, rows[0]) + + def render_detail(self, table_id: str, row: dict[str, Any]) -> None: + if table_id == "accounts": + detail = ( + f"{row['name']} | {row['provider']} | group {row['routing_group']} | {row['kind_label']} | " + f"daily {row['daily_quota_cell']} | today {format_cost(row['today_cost_usd'])}, " + f"{format_count(row['today_tokens'])} tokens, {format_count(row['today_requests'])} req | " + f"{monitor_detail(row, self.status_payload)}" + ) + error = str(row.get("error") or "").strip() + if error: + detail += f" | {error}" + elif table_id == "logs": + detail = log_detail_line(row) + else: + detail = error_detail_line(row) + self.query_one("#detail", Static).update(detail) + + def on_data_table_row_highlighted(self, event: DataTable.RowHighlighted) -> None: + table_id = str(event.control.id or "") + if table_id not in {"accounts", "logs", "errors"}: + return + if event.control.has_focus: + self.active_table = table_id + if table_id != self.active_table: + return + row_maps = { + "accounts": self.account_by_key, + "logs": self.log_by_key, + "errors": self.error_by_key, + } + row = row_maps[table_id].get(str(event.row_key.value)) + if row: + self.render_detail(table_id, row) + class AccountsScreen(Screen[None]): AUTO_FOCUS = "#accounts" BINDINGS = [ @@ -1274,7 +1604,9 @@ def run_textual( def render_keys(self) -> None: table = self.query_one("#keys", DataTable) table.clear() - for row in normalize_key_rows(self.keys_payload): + rows = normalize_key_rows(self.keys_payload) + table.display = bool(rows) + for row in rows: color = key_color(row["name"]) table.add_row( Text(str(row["name"]), style=color) if color else str(row["name"]), @@ -1556,12 +1888,13 @@ def run_textual( class Sub2APIQuotaApp(App[None]): CSS = """ #summary { height: 1; padding: 0 1; color: $accent; } - #filter { height: 3; } - #accounts { height: 2fr; } - #keys { height: 1fr; border-top: solid $panel; } - #logs { height: 1fr; } - #errors { height: 1fr; } - #detail { height: 3; padding: 0 1; border-top: solid $panel; } + #filter { height: 1; border: none; padding: 0 1; } + #filter:focus { border: none; } + #accounts { height: 1fr; min-height: 3; } + #keys { height: auto; max-height: 6; } + #logs { height: 1fr; min-height: 3; } + #errors { height: 1fr; min-height: 3; } + #detail { height: 2; padding: 0 1; background: $surface-lighten-1; } #status { height: 1; padding: 0 1; color: $text-muted; } """ BINDINGS = [ @@ -1569,12 +1902,14 @@ def run_textual( ] def on_mount(self) -> None: - if start_page == "logs": + if start_page == "accounts": + self.push_screen(AccountsScreen()) + elif start_page == "logs": self.push_screen(LogsScreen()) elif start_page == "errors": self.push_screen(ErrorsScreen()) else: - self.push_screen(AccountsScreen()) + self.push_screen(DashboardScreen()) Sub2APIQuotaApp().run() return 0 @@ -1640,8 +1975,9 @@ def build_parser() -> argparse.ArgumentParser: ) parser.add_argument("--timeout", type=int, default=env_int("SUB2API_QUOTA_TUI_TIMEOUT", DEFAULT_TIMEOUT_SECONDS)) parser.add_argument("--once", action="store_true", help="print one snapshot and exit") - parser.add_argument("--logs", action="store_true", help="start on the request logs page (with --once: print a logs snapshot)") - parser.add_argument("--errors", action="store_true", help="start on the merged cn+us errors page (with --once: print an errors snapshot)") + parser.add_argument("--accounts", action="store_true", help="start on the dedicated accounts page (with --once: print the accounts snapshot)") + parser.add_argument("--logs", action="store_true", help="start on the dedicated request logs page (with --once: print a logs snapshot)") + parser.add_argument("--errors", action="store_true", help="start on the dedicated merged cn+us errors page (with --once: print an errors snapshot)") parser.add_argument("--filter", default="", help="initial filter for --once output") return parser @@ -1652,8 +1988,9 @@ def main(argv: list[str] | None = None) -> int: logs_limit = max(1, min(args.logs_limit, 1000)) errors_limit = max(1, min(args.errors_limit, 500)) errors_time_range = str(args.errors_time_range or DEFAULT_ERRORS_TIME_RANGE).strip() or DEFAULT_ERRORS_TIME_RANGE - if args.logs and args.errors: - print("choose only one of --logs or --errors", file=sys.stderr) + selected_pages = sum(bool(value) for value in (args.accounts, args.logs, args.errors)) + if selected_pages > 1: + print("choose only one of --accounts, --logs, or --errors", file=sys.stderr) return 2 if args.save_config or args.install: config_path = write_api_url_config(args.api_url) @@ -1711,8 +2048,10 @@ def main(argv: list[str] | None = None) -> int: except Exception as exc: print(f"keys error: {exc}", file=sys.stderr) return 0 - start_page = "accounts" - if args.logs: + start_page = "dashboard" + if args.accounts: + start_page = "accounts" + elif args.logs: start_page = "logs" elif args.errors: start_page = "errors" diff --git a/tests/test_payload.py b/tests/test_payload.py index b5d781b..4dc4c17 100644 --- a/tests/test_payload.py +++ b/tests/test_payload.py @@ -668,5 +668,119 @@ class Sub2APILogsTests(unittest.TestCase): 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_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) + for selector in ("#accounts", "#logs", "#errors"): + table = screen.query_one(selector) + self.assertGreaterEqual(table.region.height, 5) + self.assertLessEqual(table.virtual_size.width, table.region.width) + for key, expected_id in (("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() diff --git a/uv.lock b/uv.lock index 9f47fad..3992d83 100644 --- a/uv.lock +++ b/uv.lock @@ -85,7 +85,7 @@ wheels = [ [[package]] name = "shusub2" -version = "0.2.3" +version = "0.2.5" source = { editable = "." } dependencies = [ { name = "textual" },