"""Tests for the Textual monitoring dashboard (tui.py). Offline, no real router / network. The data layer (``build_model`` and ``fetch_metrics``) is importable without a running TUI, so nearly all assertions are on the rendered panel PAYLOADS (plain dicts of display rows), not on pixels. ``App.run_test`` drives the app itself with a stubbed fetcher. This file imports ``tui`` and ``textual`` deliberately — it is the ONE test file that may. No non-tui module imports textual. """ from __future__ import annotations import asyncio import time import pytest import requests import tui import tui_model from tui_model import build_model, fetch_metrics def _fixture() -> dict: """A representative GET /metrics payload (matches dispatcher's shape).""" return { "quota": { "plan_kwh": 6.25, "metered_kwh_30d": 1.25, "metered_fraction_of_plan": 0.2, "metered_calls_30d": 18, "note": "router-metered only", }, "coverage": { "routable_models": 13, "with_energy_data": 10, "with_proficiency_data": 12, "quota": { "plan_kwh": 6.25, "metered_kwh_30d": 1.25, "metered_fraction_of_plan": 0.2, "metered_calls_30d": 18, "note": "router-metered only", }, "warnings": [ "3/13 routable models have no reference-workload observations", "1/13 routable models have no proficiency data", ], }, "recent_decisions": [ { "id": 42, "observed_at": "2026-08-23T10:00:00+00:00", "kind": "chat", "task_category": "coding_general", "task_tier": 2, "selected_model": "deepseek-v4-flash", "selected_provider": "neuralwatt", "est_cost_usd": 0.00016, }, { "id": 41, "observed_at": "2026-08-23T09:59:00+00:00", "kind": "route", "task_category": "docs_writing", "task_tier": 3, "selected_model": "kimi-k2.7-code", "selected_provider": "neuralwatt", "est_cost_usd": 0.0136, }, { "id": 40, "observed_at": "2026-08-23T09:58:00+00:00", "kind": "chat", "task_category": "debugging", "task_tier": 1, "selected_model": None, "selected_provider": None, "est_cost_usd": None, }, ], "per_model": [ { "model_id": "deepseek-v4-flash", "provider": "neuralwatt", "calls": 12, "sum_cost_usd": 0.0012, "sum_energy_kwh": 2.5e-05, "sum_carbon_g_co2eq": 1.2e-04, }, { "model_id": "kimi-k3", "provider": "neuralwatt", "calls": 4, "sum_cost_usd": 0.08, "sum_energy_kwh": 1.0e-03, "sum_carbon_g_co2eq": 3.0e-03, }, ], "verdict_mix": {"ok": 5, "unverifiable": 2, "truncated": 1}, "top_proficiency": [ {"model_id": "deepseek-v4-flash", "provider": "neuralwatt", "blended_score": 1.0, "source": "self_eval_thin", "self_eval_samples": 3} ], "generated_at": "2026-08-23T10:01:00+00:00", } # -------------------------------------------------------------------------- # Direct unit tests of the data layer (no TUI running). # -------------------------------------------------------------------------- def test_build_model_quota_panel(): m = build_model(_fixture()) rows = m["quota"] # plan number is surfaced as a display row joined = " ".join(r["label"] + "=" + str(r["value"]) for r in rows) assert "plan_kwh=6.25" in joined assert "metered_kwh_30d=1.25" in joined assert "fraction=0.2" in joined assert "calls=18" in joined def test_build_model_per_model_lists_seeded_models(): m = build_model(_fixture()) rows = m["per_model"] assert rows[0]["model"] == "deepseek-v4-flash" assert rows[0]["calls"] == 12 assert rows[1]["model"] == "kimi-k3" # every row keeps numeric cost/energy/carbon for display assert rows[0]["cost_usd"] == 0.0012 assert rows[0]["energy_kwh"] == 2.5e-05 assert rows[0]["carbon_g_co2eq"] == 1.2e-04 def test_build_model_verdict_mix(): m = build_model(_fixture()) mix = m["verdict_mix"] by_verdict = {r["verdict"]: r["count"] for r in mix} assert by_verdict == {"ok": 5, "unverifiable": 2, "truncated": 1} def test_build_model_recent_decisions_top_rows(): m = build_model(_fixture()) rows = m["recent_decisions"] # DESC by id: first row is id 42 assert rows[0]["id"] == 42 assert rows[0]["kind"] == "chat" assert rows[0]["category"] == "coding_general" assert rows[0]["tier"] == 2 assert rows[0]["selected"] == "deepseek-v4-flash" assert rows[1]["kind"] == "route" assert rows[2]["selected"] == "none" # no-candidate row renders "none" def test_build_model_warnings_from_coverage(): m = build_model(_fixture()) warnings = m["warnings"] assert len(warnings) == 2 assert "reference-workload" in warnings[0] assert "proficiency data" in warnings[1] def test_build_model_handles_missing_quota(): """Empty DB / null plan: the quota section degrades to a notice.""" data = _fixture() data["quota"] = None data["coverage"]["quota"] = None m = build_model(data) joined = " ".join(str(r) for r in m["quota"]) assert "unset" in joined.lower() or "no plan" in joined.lower() or "n/a" in joined.lower() def test_fetch_metrics_returns_parsed_dict(monkeypatch): """fetch_metrics hits the right URL and returns parsed JSON.""" captured = {} class _FakeResp: def raise_for_status(self): return None def json(self): return {"quota": None, "ok": True} def _fake_get(url, timeout=None): captured["url"] = url captured["timeout"] = timeout return _FakeResp() monkeypatch.setattr(tui_model.requests, "get", _fake_get) out = fetch_metrics("http://testhost:8081") assert out == {"quota": None, "ok": True} assert captured["url"] == "http://testhost:8081/metrics" def test_fetch_metrics_raises_on_http_error(monkeypatch): class _Err: def raise_for_status(self): raise RuntimeError("500") monkeypatch.setattr(tui_model.requests, "get", lambda *a, **k: _Err()) with pytest.raises(Exception): fetch_metrics("http://x") def test_fetch_metrics_raises_on_network_error(monkeypatch): def _boom(*a, **k): raise ConnectionError("refused") monkeypatch.setattr(tui_model.requests, "get", _boom) with pytest.raises(ConnectionError): fetch_metrics("http://x") # -------------------------------------------------------------------------- # App-level tests via App.run_test with a stubbed fetch_metrics. # -------------------------------------------------------------------------- class _StubFetcher: """Swappable fake for fetch_metrics the App calls.""" def __init__(self): self.payload = None self.error = None self.calls = 0 def __call__(self, base_url): self.calls += 1 if self.error is not None: raise self.error return self.payload @pytest.mark.parametrize("fetcher_arg", ["callable", "subclass"]) def test_app_run_test_populates_quota_and_model_panels(fetcher_arg): stub = _StubFetcher() stub.payload = _fixture() app = tui.DashboardApp(fetcher=stub) def _assert(a): stash = getattr(a, "_last_model", None) assert stash is not None, "build_model result was not stashed on the app" # quota plan number surfaced quota_text = " ".join( f"{r['label']}={r['value']}" for r in stash["quota"] ) assert "plan_kwh=6.25" in quota_text # per-model lists the seeded models models = {r["model"] for r in stash["per_model"]} assert {"deepseek-v4-flash", "kimi-k3"} <= models assert stub.calls == 1 _run_app(app, _assert) def test_app_run_test_recent_and_warnings_panels(): stub = _StubFetcher() stub.payload = _fixture() app = tui.DashboardApp(fetcher=stub) def _assert(a): stash = a._last_model assert stash["recent_decisions"][0]["selected"] == "deepseek-v4-flash" assert len(stash["warnings"]) == 2 _run_app(app, _assert) def test_app_run_test_quota_panel_static_shows_plan(): """The rendered Static widget carries the plan number after a good fetch.""" stub = _StubFetcher() stub.payload = _fixture() app = tui.DashboardApp(fetcher=stub) def _assert(a): quota_widget = a.query_one("#quota-panel") assert "6.25" in str(quota_widget.content) _run_app(app, _assert) def test_app_run_test_failure_shows_error_and_does_not_crash(): """fetch raising -> error panel visible, run_test completes without raising.""" stub = _StubFetcher() stub.error = ConnectionError("cannot reach router") app = tui.DashboardApp(fetcher=stub, base_url="http://127.0.0.1:8080") def _assert(a): error_widget = a.query_one("#error-panel") assert "cannot reach router" in str(error_widget.content) assert a._last_error is not None _run_app(app, _assert) # must not raise def _run_app(app: tui.DashboardApp, body) -> None: """Drive the app via Textual App.run_test synchronously. ``body(app)`` runs while the app is mounted, so queries and the data model are live. Assertion failures inside propagate out of ``asyncio.run`` as normal test failures. """ async def _go(): async with app.run_test() as pilot: await pilot.pause() body(app) asyncio.run(_go()) # -------------------------------------------------------------------------- # Auto-refresh, error resilience and keyboard controls. # # These drive the app through App.run_test with a tiny REFRESH_SECONDS and no # real wall-clock sleep. Textual 8.2.8 schedules set_interval timers on the # asyncio event loop, so repeatedly awaiting ``pilot.pause()`` lets due ticks # fire without the test asserting on elapsed time or calling time.sleep(). # Assertions are on the re-rendered panel payload (``app._last_model``), not # on mock-call counts. # -------------------------------------------------------------------------- class _CountingFetcher: """Hands back a metrics payload whose first per-model call count equals the invocation number, so each fetch produces a distinct, inspectable payload with no script to exhaust.""" def __init__(self): self.calls = 0 def __call__(self, base_url): self.calls += 1 return _variant(self.calls) def _variant(calls: int) -> dict: """Return a metrics payload whose per-model call count is ``calls``.""" data = _fixture() data["per_model"][0]["calls"] = calls return data def _first_model_calls(app) -> int: """Read the re-rendered per-model payload's first row call count.""" return app._last_model["per_model"][0]["calls"] def test_auto_refresh_rerenders_updated_payload(): """Interval ticks re-fetch and re-render: the re-rendered panel tracks the fetcher's latest payload. No real sleep: the interval is tiny and every tick fires while the event loop is pumped through ``pilot.pause()``. The assertion reads the actual re-rendered payload back, not a mock-call count. """ fetcher = _CountingFetcher() app = tui.DashboardApp(fetcher=fetcher, refresh_seconds=0.05) async def _go(): async with app.run_test() as pilot: await pilot.pause() # >1 fetch means an auto-tick fired beyond the on_mount refresh. for _ in range(30): await pilot.pause() assert fetcher.calls > 1 # The displayed panel reflects the fetcher's newest payload. assert _first_model_calls(app) == fetcher.calls assert app._refreshing is False asyncio.run(_go()) def test_error_resilience_keeps_last_good_data_and_recovers(): """A transient fetch failure keeps the app running and the last good data displayed; a later successful refresh re-renders the new payload. A large interval means no stray auto-ticks, so the 2nd fetch is exactly the failing one, driven deterministically through the same ``_on_interval`` callback the timer invokes — no sleep, no timing race. """ calls = {"n": 0} def _scripted(base_url): calls["n"] += 1 if calls["n"] == 2: raise ConnectionError("transient blip") return _variant(calls["n"]) app = tui.DashboardApp(fetcher=_scripted, refresh_seconds=60) async def _go(): async with app.run_test() as pilot: await pilot.pause() assert _first_model_calls(app) == 1 # initial good fetch displayed # advance one tick (the failing 2nd fetch) app._on_interval() await pilot.pause() assert app._last_error is not None, "transient failure was not seen" assert not app._exit, "app must not exit on a transient failure" err_widget = app.query_one("#error-panel") assert "cannot reach router" in str(err_widget.content) assert "visible" in err_widget.classes # last good data is still the displayed payload assert _first_model_calls(app) == 1 # recover on the next refresh (3rd fetch, now successful) app._on_interval() await pilot.pause() assert _first_model_calls(app) == 3 assert app._last_error is None asyncio.run(_go()) def test_force_refresh_binding_reloads_on_r(): """Pressing ``r`` immediately re-fetches and re-renders a new payload.""" fetcher = _CountingFetcher() # A large interval ensures only the forced refresh advances the payload. app = tui.DashboardApp(fetcher=fetcher, refresh_seconds=60) async def _go(): async with app.run_test() as pilot: await pilot.pause() baseline = fetcher.calls await pilot.press("r") await pilot.pause() assert fetcher.calls == baseline + 1 assert _first_model_calls(app) == fetcher.calls asyncio.run(_go()) @pytest.mark.parametrize("key", ["q", "Q", "ctrl+c"]) def test_quit_bindings_exit_app(key): """q, Q and Ctrl+C all quit the running app.""" fetcher = _CountingFetcher() app = tui.DashboardApp(fetcher=fetcher, refresh_seconds=60) async def _go(): async with app.run_test() as pilot: await pilot.pause() assert not app._exit await pilot.press(key) assert app._exit asyncio.run(_go()) @pytest.mark.parametrize( "key,panel", [ ("1", "model-table"), ("2", "verdict-table"), ("3", "decision-table"), ("4", "category-table"), ("5", "quota-panel"), ("6", "warnings-panel"), ], ) def test_number_bindings_focus_panel(key, panel): """Number keys 1-5 focus the corresponding panel.""" fetcher = _CountingFetcher() app = tui.DashboardApp(fetcher=fetcher, refresh_seconds=60) async def _go(): async with app.run_test() as pilot: await pilot.pause() await pilot.press(key) await pilot.pause() widget = app.query_one(f"#{panel}") assert widget.has_focus, f"{panel} should have focus after {key!r}" asyncio.run(_go()) @pytest.fixture(autouse=True) def _no_real_network(monkeypatch): """Safety net: even if the fetcher is mis-wired, never hit a real router.""" def _guard(base_url): raise AssertionError(f"real fetch_metrics called with {base_url!r}") monkeypatch.setattr(tui, "fetch_metrics", _guard) # Also guard the SSE consumer's requests so a stray DecisionStream never # reaches the network even if a test forgets to disable live_events. import tui_sse def _sse_guard(*args, **kwargs): raise AssertionError( f"real requests.get called from tui_sse with {args!r} {kwargs!r}" ) monkeypatch.setattr(tui_sse.requests, "get", _sse_guard) # -------------------------------------------------------------------------- # Category breakdown and enriched decision fields (pure data-layer tests). # -------------------------------------------------------------------------- def test_build_category_breakdown_majority_and_share(): """One category, two different winners: majority is the most common and the share is its fraction of the count.""" decisions = [ {"id": 3, "category": "coding_general", "tier": 2, "selected": "a"}, {"id": 2, "category": "coding_general", "tier": 2, "selected": "a"}, {"id": 1, "category": "coding_general", "tier": 2, "selected": "b"}, ] rows = tui_model.build_category_breakdown(decisions) assert len(rows) == 1 row = rows[0] assert row["category"] == "coding_general" assert row["tier"] == 2 assert row["count"] == 3 assert row["majority"] == "a" assert row["share"] == round(2 / 3, 2) def test_build_category_breakdown_separates_tiers(): """Same category, different tiers are separate buckets.""" decisions = [ {"id": 2, "category": "coding_general", "tier": 1, "selected": "tiny"}, {"id": 1, "category": "coding_general", "tier": 3, "selected": "big"}, ] rows = tui_model.build_category_breakdown(decisions) assert len(rows) == 2 tiers = {r["tier"] for r in rows} assert tiers == {1, 3} def test_build_category_breakdown_handles_empty_and_none_selected(): """No decisions: empty list. Decisions with no selected model get majority 'none' and share 1.0 (they all count toward the bucket).""" assert tui_model.build_category_breakdown([]) == [] rows = tui_model.build_category_breakdown( [{"id": 1, "category": "x", "tier": 1, "selected": None}] ) assert rows[0]["majority"] == "none" def test_build_model_recent_decisions_carry_enriched_fields(): """The enriched /metrics row fields must reach the TUI data model so the detail popup can render the full decision in full.""" data = _fixture() # Ensure the fixture's first row has the fields the new model surfaces. data["recent_decisions"][0].update( { "required_context_tokens": 50000, "confidence": 0.92, "classifier_ms": 1800, "classification_source": "classifier", "latency_tolerance": "interactive", "candidates_considered": 8, "runner_up_models": '[{"model_id":"kimi-k3","provider":"neuralwatt"}]', "est_proficiency": 0.9, "rejected_reason": None, "tools": 0, "images": 0, "json_mode": 0, "streamed": 1, } ) m = build_model(data) row = m["recent_decisions"][0] assert row["required_context_tokens"] == 50000 assert row["confidence"] == 0.92 assert row["runner_up_models"].startswith("[{") assert row["streamed"] == 1 # The breakdown is always present (even an empty list proves the key). assert "category_breakdown" in m # -------------------------------------------------------------------------- # Detail popup and live SSE decision handling (App-level tests). # -------------------------------------------------------------------------- def test_show_decision_detail_pushes_modal_with_full_row(): """Pressing ``e`` on the decisions table opens a modal whose body contains the full JSON of the selected row — not just the table columns.""" stub = _StubFetcher() stub.payload = _fixture() stub.payload["recent_decisions"][0].update( { "required_context_tokens": 50000, "confidence": 0.92, "runner_up_models": '[{"model_id":"kimi-k3"}]', "rejected_reason": None, } ) app = tui.DashboardApp(fetcher=stub, refresh_seconds=60) async def _go(): async with app.run_test() as pilot: await pilot.pause() # Focus the decisions table and move to the first row, then open # the detail popup via the dedicated binding. app.query_one("#decision-table").focus() await pilot.pause() await pilot.press("e") await pilot.pause() # A modal screen is now active and carries the selected decision. from textual.screen import ModalScreen assert isinstance(app.screen, ModalScreen) decision = app.screen.decision assert decision["id"] == 42 assert decision["required_context_tokens"] == 50000 assert "kimi-k3" in decision["runner_up_models"] asyncio.run(_go()) def test_live_decision_inserts_row_at_front_and_rerenders(): """A decision delivered via the SSE callback is prepended to the model and re-renders the decisions and category tables without a full re-fetch.""" stub = _StubFetcher() stub.payload = _fixture() app = tui.DashboardApp(fetcher=stub, refresh_seconds=60) async def _go(): async with app.run_test() as pilot: await pilot.pause() before = len(app._last_model["recent_decisions"]) # Simulate the SSE consumer handing in a brand-new decision. # Same (category, tier) as the fixture's first row so the bucket # count for coding_general/tier-2 rises to 2. app._handle_live_decision( { "id": 999, "kind": "chat", "task_category": "coding_general", "task_tier": 2, "selected_model": "deepseek-v4-flash", "est_cost_usd": 0.0002, } ) await pilot.pause() after = app._last_model["recent_decisions"] assert len(after) == before + 1 assert after[0]["id"] == 999 # prepended, newest-first # The table was re-rendered: the first row shows the new id. dt = app.query_one("#decision-table") first_row_text = " ".join(str(c) for c in dt.get_row_at(0)) assert "999" in first_row_text asyncio.run(_go()) def test_live_decision_caps_recent_decisions_at_fifty(): """The live feed never grows the in-memory list past the /metrics cap, so the dashboard's view stays consistent with a /metrics refresh.""" stub = _StubFetcher() # Start with exactly 50 rows so one live addition must evict the oldest. base = _fixture()["recent_decisions"][0] stub.payload = {"recent_decisions": [dict(base, id=i) for i in range(50, 0, -1)]} app = tui.DashboardApp(fetcher=stub, refresh_seconds=60) async def _go(): async with app.run_test() as pilot: await pilot.pause() assert len(app._last_model["recent_decisions"]) == 50 app._handle_live_decision( {"id": 1, "kind": "chat", "task_category": "x", "task_tier": 1} ) await pilot.pause() assert len(app._last_model["recent_decisions"]) == 50 asyncio.run(_go()) def test_live_decision_dedup_skips_duplicate_id(): stub = _StubFetcher() stub.payload = _fixture() app = tui.DashboardApp(fetcher=stub, refresh_seconds=60) async def _go(): async with app.run_test() as pilot: await pilot.pause() before = len(app._last_model["recent_decisions"]) app._handle_live_decision( { "id": 42, "kind": "chat", "task_category": "coding_general", "task_tier": 2, "selected_model": "deepseek-v4-flash", } ) await pilot.pause() after = app._last_model["recent_decisions"] assert len(after) == before, "duplicate id did not get skipped" # first row is id=41 (42 was skipped as duplicate; 42 is now at pos 0) assert after[0]["id"] == 42 and after[1]["id"] == 41, ( "first row is id 42 after dedup" ) app._handle_live_decision( { "id": 888, "kind": "route", "task_category": "coding_general", "task_tier": 2, "selected_model": "kimi-k3", } ) await pilot.pause() assert len(app._last_model["recent_decisions"]) == before + 1 assert app._last_model["recent_decisions"][0]["id"] == 888 asyncio.run(_go()) def test_live_decision_new_bucket_rebuilds_breakdown(): stub = _StubFetcher() stub.payload = _fixture() app = tui.DashboardApp(fetcher=stub, refresh_seconds=60) async def _go(): async with app.run_test() as pilot: await pilot.pause() buckets_before = set( (r["category"], r["tier"]) for r in app._last_model["category_breakdown"] ) app._handle_live_decision( { "id": 1000, "kind": "route", "task_category": "summarization", "task_tier": 1, "selected_model": "qwen3.6-35b", } ) await pilot.pause() buckets_after = set( (r["category"], r["tier"]) for r in app._last_model["category_breakdown"] ) assert ("summarization", 1) in buckets_after - buckets_before row = next( r for r in app._last_model["category_breakdown"] if r["category"] == "summarization" and r["tier"] == 1 ) assert row["count"] == 1 assert row["majority"] == "qwen3.6-35b" asyncio.run(_go()) # -------------------------------------------------------------------------- # DecisionStream: callback exceptions and stop behavior (tui_sse.py). # -------------------------------------------------------------------------- class _MockResp: def __enter__(self): return self def __exit__(self, *a): pass def raise_for_status(self): pass def iter_lines(self, decode_unicode=False): yield "data: {\"id\": 1}" raise requests.exceptions.ConnectionError("broken pipe") def test_callback_raises_doesnt_break_reconnect(monkeypatch): """A callback that raises RuntimeError is caught; the stream still survives and processes subsequent decisions after a reconnection.""" import tui_sse monkeypatch.setattr(tui_sse.requests, "get", lambda *a, **kw: _MockResp()) call_count = {"n": 0} def failing_callback(decision): call_count["n"] += 1 if call_count["n"] == 1: raise RuntimeError("app loop gone") # second call succeeds — proves reconnect worked s = tui_sse.DecisionStream( "http://127.0.0.1", failing_callback, reconnect_seconds=0.1, ) s.start() time.sleep(0.6) s.stop() s.join(timeout=2) assert not s.is_alive() assert call_count["n"] >= 2, ( f"Expected reconnection after callback failure, got {call_count['n']} call(s)" ) def test_stopped_stream_exits_without_reconnect_sleep(monkeypatch): """After stop(), the thread should NOT wait for reconnect_seconds before exiting — the _stopped guard is checked before the sleep.""" import tui_sse class _MockResp: def __enter__(self): return self def __exit__(self, *a): pass def raise_for_status(self): pass def iter_lines(self, decode_unicode=False): raise requests.exceptions.ConnectionError("closed") monkeypatch.setattr(tui_sse.requests, "get", lambda *a, **kw: _MockResp()) stop_at = time.monotonic() s = tui_sse.DecisionStream( "http://127.0.0.1", lambda x: None, reconnect_seconds=5.0, ) s.start() # Let the first request attempt begin. time.sleep(0.2) # Record when stop is called. s.stop() stopped_at = time.monotonic() # Thread should exit BEFORE the 5s reconnect backoff (use 3s as margin). s.join(timeout=3) assert not s.is_alive(), "Thread should exit promptly after stop()" assert stopped_at - stop_at < 3.0, "Thread slept through reconnect_seconds"