alee/6krrt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
neuralwatt-router-service
6krrt/tests/test_tui.py
adlee-was-taken 943b2d5cde fix: resolve TUI live-routing panel review findings
2026-08-24 00:22:01 -04:00

842 lines
28 KiB
Python
Raw Permalink Blame History

"""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"
Reference in New Issue View Git Blame Copy Permalink