| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059 |
- """Smoke test: the Textual app mounts, subscribes, and ingests broadcasts."""
- import asyncio
- import os
- import shutil
- import sys
- import tempfile
- import pytest
- from stt_daemon.assistant.control import ControlServer
- from stt_daemon.assistant.service import AssistantService
- from stt_daemon.config import Config
- class FakeSubscriber:
- def events(self):
- return iter(())
- def close(self):
- pass
- @pytest.fixture
- def server():
- directory = tempfile.mkdtemp(prefix="stta-")
- path = os.path.join(directory, "c.sock")
- cfg = Config()
- cfg.assistant.control_socket = path
- service = AssistantService(cfg, FakeSubscriber())
- control = ControlServer(cfg, service, path=path)
- service.set_sinks(emit=control.broadcast, relay=control.relay)
- control.start()
- yield path, service
- control.close()
- service.stop()
- shutil.rmtree(directory, ignore_errors=True)
- def test_ctrl_r_sends_execute_and_e_opens_editor(server):
- pytest.importorskip("textual")
- from textual.widgets import Input
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path)
- sent: list[tuple] = []
- observed: dict = {}
- async def scenario():
- async with app.run_test() as pilot:
- await pilot.pause()
- app.client.send = lambda type_, **fields: sent.append((type_, fields))
- await pilot.press("ctrl+r")
- await pilot.pause()
- await pilot.press("a")
- await pilot.pause()
- await pilot.press("escape")
- await pilot.pause()
- await pilot.press("e")
- await pilot.pause()
- edit = app.query_one("#edit", Input)
- observed["edit_display"] = edit.display
- observed["edit_focused"] = edit.has_focus
- asyncio.run(scenario())
- assert ("command_execute", {}) in sent
- assert ("command_start", {}) in sent
- assert ("command_cancel", {}) in sent
- assert observed["edit_display"] is True
- assert observed["edit_focused"] is True
- def test_c_key_confirms_before_clearing(server):
- pytest.importorskip("textual")
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path)
- sent: list[tuple] = []
- async def scenario():
- async with app.run_test() as pilot:
- await pilot.pause()
- app.client.send = lambda type_, **fields: sent.append((type_, fields))
- await pilot.press("c")
- await pilot.pause()
- assert not any(type_ == "clear" for type_, _ in sent), "clear sent before confirmation"
- await pilot.press("y")
- await pilot.pause()
- asyncio.run(scenario())
- assert ("clear", {}) in sent
- def test_c_key_cancel_does_not_clear(server):
- pytest.importorskip("textual")
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path)
- sent: list[tuple] = []
- async def scenario():
- async with app.run_test() as pilot:
- await pilot.pause()
- app.client.send = lambda type_, **fields: sent.append((type_, fields))
- await pilot.press("c")
- await pilot.pause()
- await pilot.press("n")
- await pilot.pause()
- asyncio.run(scenario())
- assert not any(type_ == "clear" for type_, _ in sent)
- def test_c_key_escape_does_not_clear(server):
- pytest.importorskip("textual")
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path)
- sent: list[tuple] = []
- async def scenario():
- async with app.run_test() as pilot:
- await pilot.pause()
- app.client.send = lambda type_, **fields: sent.append((type_, fields))
- await pilot.press("c")
- await pilot.pause()
- await pilot.press("escape")
- await pilot.pause()
- asyncio.run(scenario())
- assert not any(type_ == "clear" for type_, _ in sent)
- def test_c_key_clears_rendered_output(server):
- """Pressing 'c' clears the conversation pane via the cleared event."""
- pytest.importorskip("textual")
- from textual.widgets import RichLog
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path)
- async def scenario():
- async with app.run_test() as pilot:
- await pilot.pause()
- app.state.ingest({"type": "chat_message", "role": "user", "text": "hello"})
- app.state.ingest(
- {
- "type": "tool_result",
- "tool": "get_datetime",
- "ok": True,
- "result": "2026-09-21",
- }
- )
- app._render()
- await pilot.pause()
- app.state.ingest({"type": "cleared"})
- app._render()
- await pilot.pause()
- log = app.query_one("#chat", RichLog)
- return list(log.lines)
- lines = asyncio.run(scenario())
- assert not any("hello" in str(line) for line in lines)
- assert not any("2026-09-21" in str(line) for line in lines)
- def test_tui_renders_command_output(server):
- """A tool result must actually reach a widget - otherwise 'nothing happens'."""
- pytest.importorskip("textual")
- from textual.widgets import RichLog
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path)
- async def scenario():
- async with app.run_test() as pilot:
- await pilot.pause()
- app.state.ingest(
- {
- "type": "tool_result",
- "tool": "get_datetime",
- "ok": True,
- "result": "2026-09-21T01:00:00",
- }
- )
- app._render()
- await pilot.pause()
- log = app.query_one("#chat", RichLog)
- return list(log.lines)
- lines = asyncio.run(scenario())
- assert lines, "command output was not rendered anywhere"
- def test_tui_clears_chat_pane_on_chat_exit(server):
- """Leaving chat must drop the rendered conversation but keep command output."""
- pytest.importorskip("textual")
- from textual.widgets import RichLog
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path)
- async def scenario():
- async with app.run_test() as pilot:
- await pilot.pause()
- app.state.ingest(
- {"type": "tool_result", "tool": "get_datetime", "ok": True, "result": "kept"}
- )
- app.state.ingest({"type": "mode", "mode": "chat"})
- app.state.ingest({"type": "chat_message", "role": "user", "text": "gone"})
- app._render()
- await pilot.pause()
- app.state.ingest({"type": "mode", "mode": "idle"})
- app._render()
- await pilot.pause()
- return [str(line) for line in app.query_one("#chat", RichLog).lines]
- lines = asyncio.run(scenario())
- assert not any("gone" in line for line in lines)
- assert any("kept" in line for line in lines)
- def test_tui_mounts_subscribes_and_ingests_broadcasts(server):
- pytest.importorskip("textual")
- from stt_daemon.tui.app import AssistantTUI
- path, service = server
- app = AssistantTUI(path)
- async def scenario():
- async with app.run_test() as pilot:
- await pilot.pause()
- service.arm() # broadcast to the subscribed viewer
- for _ in range(60):
- await pilot.pause()
- if app.state.mode == "command_armed":
- break
- await asyncio.sleep(0.05)
- asyncio.run(scenario())
- assert app.state.mode == "command_armed"
- def test_render_after_unmount_is_safe(server):
- """A timer tick can land during teardown; rendering must not raise."""
- pytest.importorskip("textual")
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path)
- async def scenario():
- async with app.run_test() as pilot:
- await pilot.pause()
- asyncio.run(scenario())
- app._render() # must not raise NoMatches
- def test_tui_records_connect_failure(tmp_path):
- pytest.importorskip("textual")
- from stt_daemon.tui.app import AssistantTUI
- app = AssistantTUI(str(tmp_path / "missing.sock"))
- async def scenario():
- async with app.run_test() as pilot:
- await pilot.pause()
- asyncio.run(scenario())
- assert app.connect_error is not None
- assert "cannot connect" in app.connect_error
- def test_border_titles_show_queue_and_inference_depths(server):
- """Border titles for transcript and chat show live segment and inference queue counts."""
- pytest.importorskip("textual")
- from textual.widgets import RichLog
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path)
- async def scenario():
- async with app.run_test() as pilot:
- await pilot.pause()
- app.state.ingest(
- {
- "type": "heartbeat",
- "queue_depth": 5,
- "inference_depth": 3,
- "inference_in_flight": False,
- }
- )
- app._render()
- await pilot.pause()
- transcript_title = app.query_one("#transcript", RichLog).border_title
- chat_title = app.query_one("#chat", RichLog).border_title
- assert "[seg 5]" in transcript_title
- assert "[inf 3]" in chat_title
- assert "busy" not in chat_title
- # Test with inference in flight
- app.state.ingest(
- {
- "type": "heartbeat",
- "inference_depth": 7,
- "inference_in_flight": True,
- }
- )
- app._render()
- await pilot.pause()
- chat_title = app.query_one("#chat", RichLog).border_title
- assert "[inf 7" in chat_title
- assert "busy" in chat_title
- asyncio.run(scenario())
- def test_main_exit_code_reflects_connect_failure(monkeypatch, tmp_path, capsys):
- pytest.importorskip("textual")
- from stt_daemon.tui import app as tui_app
- monkeypatch.setattr(
- tui_app.AssistantTUI,
- "run",
- lambda self: setattr(self, "connect_error", "boom"),
- )
- monkeypatch.setattr(sys, "argv", ["stt-tui", "--socket", str(tmp_path / "x.sock")])
- assert tui_app.main() == 1
- assert "boom" in capsys.readouterr().err
- monkeypatch.setattr(tui_app.AssistantTUI, "run", lambda self: None)
- assert tui_app.main() == 0
- def test_triage_disabled_composes_no_triage_widget(server):
- """Default (triage_enabled=False): no #triage, no #main — tree as today."""
- pytest.importorskip("textual")
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path)
- async def scenario():
- async with app.run_test() as pilot:
- await pilot.pause()
- assert not app.query("#triage")
- assert not app.query("#main")
- # The rest of the tree is intact.
- app.query_one("#transcript")
- app.query_one("#status")
- app.query_one("#command")
- app.query_one("#chat")
- app.query_one("#edit")
- asyncio.run(scenario())
- def test_triage_enabled_composes_split(server):
- """triage_enabled=True: #main wraps #triage (left) + #chat (right); transcript stays outside."""
- pytest.importorskip("textual")
- from textual.containers import Horizontal
- from textual.widgets import RichLog
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path, triage_enabled=True)
- async def scenario():
- async with app.run_test() as pilot:
- await pilot.pause()
- main = app.query_one("#main", Horizontal)
- transcript = app.query_one("#transcript", RichLog)
- triage = app.query_one("#triage", RichLog)
- chat = app.query_one("#chat", RichLog)
- assert triage.parent is main
- assert chat.parent is main
- assert transcript.parent is not main
- assert triage.border_title == "triage"
- # triage sits LEFT of chat inside #main.
- assert main.children.index(triage) < main.children.index(chat)
- # Outside widgets stay outside #main.
- assert app.query_one("#status").parent is not main
- assert transcript.parent is not main
- asyncio.run(scenario())
- def test_triage_enabled_layout_proportions(server):
- """Rendered proportions: #triage ~1/3 and #chat ~2/3 of #main; #transcript ~25% of screen."""
- pytest.importorskip("textual")
- from textual.containers import Horizontal
- from textual.widgets import RichLog
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path, triage_enabled=True)
- async def scenario(size=None):
- kwargs = {"size": size} if size else {}
- async with app.run_test(**kwargs) as pilot:
- await pilot.pause()
- main = app.query_one("#main", Horizontal)
- transcript = app.query_one("#transcript", RichLog)
- triage = app.query_one("#triage", RichLog)
- chat = app.query_one("#chat", RichLog)
- return (
- main.size.width,
- triage.size.width,
- chat.size.width,
- app.screen.size.height,
- transcript.region.height,
- )
- main_w, triage_w, chat_w, screen_h, transcript_h = asyncio.run(scenario())
- print(
- f"measured: main_w={main_w} triage_w={triage_w} chat_w={chat_w} "
- f"screen_h={screen_h} transcript_h={transcript_h}"
- )
- # Content boxes exclude the 1-cell border on each side; allow ±2 px for
- # borders/rounding on the fractional split.
- assert abs(triage_w / main_w - 1 / 3) < 0.05, (main_w, triage_w, chat_w)
- assert abs(chat_w / main_w - 2 / 3) < 0.05, (main_w, triage_w, chat_w)
- # region.height includes the transcript's 2 border rows; at the small
- # default size rounding is coarse, so only demand a floor here. The
- # precise 25% check runs at a taller size below.
- assert transcript_h / screen_h >= 0.18, (screen_h, transcript_h)
- assert transcript_h >= 1, transcript_h
- app2 = AssistantTUI(path, triage_enabled=True)
- async def scenario_tall():
- async with app2.run_test(size=(80, 40)) as pilot:
- await pilot.pause()
- transcript2 = app2.query_one("#transcript", RichLog)
- return app2.screen.size.height, transcript2.region.height
- screen_h2, transcript_h2 = asyncio.run(scenario_tall())
- print(f"measured tall: screen_h={screen_h2} transcript_h={transcript_h2}")
- assert abs(transcript_h2 / screen_h2 - 0.25) < 0.06, (screen_h2, transcript_h2)
- def test_triage_enabled_renders_signals_on_epoch_change(server):
- """A triage event + status event must actually write rendered lines into #triage."""
- pytest.importorskip("textual")
- from textual.widgets import RichLog
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path, triage_enabled=True)
- async def scenario():
- async with app.run_test() as pilot:
- await pilot.pause()
- app._timer.stop() # no background drain: server status must not clobber state
- app._drain()
- app.state.ingest(
- {
- "type": "triage",
- "stage": "signals",
- "triage": {
- "order": ["category"],
- "items": {
- "category": {
- "kind": "choice",
- "choice": "question",
- "answer_confidence": 0.91,
- }
- },
- },
- }
- )
- app.state.ingest(
- {"type": "status", "triage_state": "ok", "triage_reason": ""}
- )
- app._render()
- await pilot.pause()
- return [str(line) for line in app.query_one("#triage", RichLog).lines]
- lines = asyncio.run(scenario())
- rendered = "\n".join(lines)
- assert "category: question (0.91)" in rendered
- def test_triage_enabled_loading_shows_reason_line_never_blank(server):
- """loading/unknown triage_state renders the reason line — never a blank panel."""
- pytest.importorskip("textual")
- from textual.widgets import RichLog
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path, triage_enabled=True)
- async def scenario():
- async with app.run_test() as pilot:
- await pilot.pause()
- app._timer.stop() # no background drain: server status must not clobber state
- app._drain()
- app.state.ingest(
- {
- "type": "status",
- "triage_state": "loading",
- "triage_reason": "model loading",
- }
- )
- app._render()
- await pilot.pause()
- loading_lines = [str(line) for line in app.query_one("#triage", RichLog).lines]
- # Unknown state falls back to a generic reason line, no crash.
- app.state.ingest(
- {"type": "status", "triage_state": "weird_state", "triage_reason": ""}
- )
- app._render()
- await pilot.pause()
- unknown_lines = [str(line) for line in app.query_one("#triage", RichLog).lines]
- return loading_lines, unknown_lines
- loading_lines, unknown_lines = asyncio.run(scenario())
- assert loading_lines, "triage panel was blank in loading state"
- assert any("triage: unavailable" in line for line in loading_lines)
- assert any("model loading" in line for line in loading_lines)
- assert unknown_lines, "triage panel was blank for unknown state"
- assert any("triage: unavailable" in line for line in unknown_lines)
- assert any("weird_state" in line for line in unknown_lines)
- def test_triage_epoch_change_clears_before_rewrite(server):
- """Epoch bump clears #triage first: no duplicated stale lines."""
- pytest.importorskip("textual")
- from textual.widgets import RichLog
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path, triage_enabled=True)
- async def scenario():
- async with app.run_test() as pilot:
- await pilot.pause()
- app._timer.stop() # no background drain: server status must not clobber state
- app._drain()
- app.state.ingest(
- {
- "type": "triage",
- "stage": "signals",
- "triage": {"order": ["q1"], "items": {"q1": {"kind": "choice", "choice": "a"}}},
- }
- )
- app.state.ingest({"type": "status", "triage_state": "ok"})
- app._render()
- await pilot.pause()
- # New epoch: cleared stage empties the payload, panel must not keep old lines.
- app.state.ingest({"type": "triage", "stage": "cleared"})
- app._render()
- await pilot.pause()
- return [str(line) for line in app.query_one("#triage", RichLog).lines]
- lines = asyncio.run(scenario())
- rendered = "\n".join(lines)
- assert "q1" not in rendered, "stale triage lines survived an epoch change"
- assert "triage: unavailable — no data" in rendered
- def test_triage_disabled_never_writes_triage(server):
- """Disabled: triage events are ingested but #triage is never touched (it doesn't exist)."""
- pytest.importorskip("textual")
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path)
- async def scenario():
- async with app.run_test() as pilot:
- await pilot.pause()
- app.state.ingest({"type": "triage", "stage": "signals", "triage": {}})
- app.state.ingest({"type": "status", "triage_state": "ok"})
- app._render() # must not raise NoMatches for #triage
- await pilot.pause()
- assert not app.query("#triage")
- asyncio.run(scenario())
- def test_default_ctor_has_no_triage_arg(server):
- """AssistantTUI(path) stays valid: triage_enabled defaults to False."""
- pytest.importorskip("textual")
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path)
- assert app.triage_enabled is False
- assert app._seen_triage_epoch == 0
- def test_maximize_triage_hides_chat_and_fills_screen(server):
- """Focus #triage, press m: #main maximized, chat hidden, triage fills screen."""
- pytest.importorskip("textual")
- from textual.widgets import RichLog
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path, triage_enabled=True)
- observed: dict = {}
- async def scenario():
- async with app.run_test(size=(80, 40)) as pilot:
- await pilot.pause()
- app.query_one("#triage", RichLog).focus()
- await pilot.pause()
- await pilot.press("m")
- await pilot.pause()
- observed["maximized_is_main"] = app.screen.maximized is app.query_one("#main")
- observed["triage_display"] = app.query_one("#triage", RichLog).display
- observed["chat_display"] = app.query_one("#chat", RichLog).display
- observed["region"] = app.query_one("#triage", RichLog).region
- observed["screen_size"] = app.screen.size
- asyncio.run(scenario())
- print("triage maximized: region=%s screen=%s" % (observed["region"], observed["screen_size"]))
- assert observed["maximized_is_main"] is True
- assert observed["triage_display"] is True
- assert observed["chat_display"] is False
- region = observed["region"]
- size = observed["screen_size"]
- assert abs(region.width - size.width) <= 2, (region, size)
- assert abs(region.height - size.height) <= 2, (region, size)
- def test_maximize_toggle_restores_layout(server):
- """Press m again: minimize restores both panes."""
- pytest.importorskip("textual")
- from textual.widgets import RichLog
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path, triage_enabled=True)
- observed: dict = {}
- async def scenario():
- async with app.run_test(size=(80, 40)) as pilot:
- await pilot.pause()
- app.query_one("#triage", RichLog).focus()
- await pilot.pause()
- await pilot.press("m")
- await pilot.pause()
- await pilot.press("m")
- await pilot.pause()
- observed["maximized"] = app.screen.maximized
- observed["triage_display"] = app.query_one("#triage", RichLog).display
- observed["chat_display"] = app.query_one("#chat", RichLog).display
- asyncio.run(scenario())
- assert observed["maximized"] is None
- assert observed["triage_display"] is True
- assert observed["chat_display"] is True
- def test_maximize_transcript_fills_screen(server):
- """Focus #transcript, press m: transcript maximized and fills the screen."""
- pytest.importorskip("textual")
- from textual.widgets import RichLog
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path, triage_enabled=True)
- observed: dict = {}
- async def scenario():
- async with app.run_test(size=(80, 40)) as pilot:
- await pilot.pause()
- app.query_one("#transcript", RichLog).focus()
- await pilot.pause()
- await pilot.press("m")
- await pilot.pause()
- observed["maximized_is_transcript"] = (
- app.screen.maximized is app.query_one("#transcript")
- )
- observed["region"] = app.query_one("#transcript", RichLog).region
- observed["screen_size"] = app.screen.size
- await pilot.press("m")
- await pilot.pause()
- observed["restored"] = app.screen.maximized
- asyncio.run(scenario())
- print("transcript maximized: region=%s screen=%s"
- % (observed["region"], observed["screen_size"]))
- assert observed["maximized_is_transcript"] is True
- region = observed["region"]
- size = observed["screen_size"]
- assert abs(region.width - size.width) <= 2, (region, size)
- assert abs(region.height - size.height) <= 2, (region, size)
- assert observed["restored"] is None
- def test_escape_while_maximized_restores_panes(server):
- """Escape minimizes (ESCAPE_TO_MINIMIZE) and _drain restores the hidden sibling."""
- pytest.importorskip("textual")
- from textual.widgets import RichLog
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path, triage_enabled=True)
- observed: dict = {}
- async def scenario():
- async with app.run_test(size=(80, 40)) as pilot:
- await pilot.pause()
- app.query_one("#triage", RichLog).focus()
- await pilot.pause()
- await pilot.press("m")
- await pilot.pause()
- assert app.screen.maximized is not None
- await pilot.press("escape")
- await pilot.pause()
- # Let the 0.1s _drain timer run so the restore path executes.
- for _ in range(10):
- await pilot.pause()
- await asyncio.sleep(0.05)
- if app.screen.maximized is None and app.query_one("#chat", RichLog).display:
- break
- observed["maximized"] = app.screen.maximized
- observed["triage_display"] = app.query_one("#triage", RichLog).display
- observed["chat_display"] = app.query_one("#chat", RichLog).display
- asyncio.run(scenario())
- assert observed["maximized"] is None
- assert observed["triage_display"] is True
- assert observed["chat_display"] is True
- def test_maximize_chat_when_triage_disabled(server):
- """triage_enabled=False: focus #chat, press m maximizes #main without crashing."""
- pytest.importorskip("textual")
- from textual.widgets import RichLog
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path)
- observed: dict = {}
- async def scenario():
- async with app.run_test(size=(80, 40)) as pilot:
- await pilot.pause()
- # Disabled: no #main/#triage; focusing #chat then m maximizes #chat itself.
- app.query_one("#chat", RichLog).focus()
- await pilot.pause()
- await pilot.press("m")
- await pilot.pause()
- observed["maximized"] = app.screen.maximized
- observed["chat"] = app.query_one("#chat", RichLog)
- observed["region"] = app.query_one("#chat", RichLog).region
- observed["screen_size"] = app.screen.size
- await pilot.press("m")
- await pilot.pause()
- observed["restored"] = app.screen.maximized
- asyncio.run(scenario())
- print("chat maximized (disabled): region=%s screen=%s"
- % (observed["region"], observed["screen_size"]))
- assert observed["maximized"] is observed["chat"]
- assert observed["restored"] is None
- def test_footer_hidden_while_maximized(server):
- """ALLOW_IN_MAXIMIZED_VIEW hides the Footer while a pane is maximized."""
- pytest.importorskip("textual")
- from textual.widgets import Footer, RichLog
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path, triage_enabled=True)
- observed: dict = {}
- async def scenario():
- async with app.run_test(size=(80, 40)) as pilot:
- await pilot.pause()
- app.query_one("#triage", RichLog).focus()
- await pilot.pause()
- await pilot.press("m")
- await pilot.pause()
- observed["maximized"] = app.screen.maximized
- observed["footer_region"] = app.query_one(Footer).region
- await pilot.press("m")
- await pilot.pause()
- asyncio.run(scenario())
- assert observed["maximized"] is not None
- footer_region = observed["footer_region"]
- assert footer_region.height == 0 or footer_region.width == 0, observed
- def test_question_mark_key_toggles_help_panel(server):
- pytest.importorskip("textual")
- from textual.widgets import HelpPanel
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path)
- observed: dict = {}
- async def scenario():
- async with app.run_test() as pilot:
- await pilot.pause()
- assert list(app.screen.query(HelpPanel)) == []
- await pilot.press("question_mark")
- await pilot.pause()
- observed["shown"] = list(app.screen.query(HelpPanel)) != []
- app.screen.query_one(HelpPanel) # must succeed while shown
- await pilot.press("question_mark")
- await pilot.pause()
- observed["hidden"] = list(app.screen.query(HelpPanel)) == []
- asyncio.run(scenario())
- assert observed["shown"] is True, "? did not mount HelpPanel"
- assert observed["hidden"] is True, "second ? did not remove HelpPanel"
- def test_app_bindings_list():
- """BINDINGS: 7 visible footer entries; hidden ones after; no h binding."""
- pytest.importorskip("textual")
- from stt_daemon.tui.app import AssistantTUI
- pairs = [(b.key, b.description) for b in AssistantTUI.BINDINGS]
- assert ("ctrl+r", "execute") in pairs
- assert ("m", "maximize pane") in pairs
- assert ("question_mark", "help") in pairs
- assert ("ctrl+s", "save screenshot") in pairs
- assert ("a", "arm command") in pairs
- assert ("escape", "cancel") in pairs
- assert ("e", "edit command") in pairs
- assert ("t", "chat toggle") in pairs
- assert ("c", "clear") in pairs
- assert ("q", "quit") in pairs
- assert ("ctrl+p", "instructions") in pairs
- assert not any(key == "h" for key, _ in pairs)
- assert not any(key == "ctrl+m" for key, _ in pairs)
- assert not any(key == "ctrl+h" for key, _ in pairs)
- assert not any(key == "enter" for key, _ in pairs)
- assert not any(key == "ctrl+k" for key, _ in pairs)
- by_key = {b.key: b for b in AssistantTUI.BINDINGS}
- assert by_key["question_mark"].show is True
- assert by_key["m"].show is False
- assert by_key["c"].show is False
- assert by_key["ctrl+s"].show is False
- assert by_key["ctrl+p"].show is False
- # Visible seven first, in footer order.
- visible = [(b.key, b.description) for b in AssistantTUI.BINDINGS if b.show]
- assert visible == [
- ("a", "arm command"),
- ("ctrl+r", "execute"),
- ("escape", "cancel"),
- ("e", "edit command"),
- ("t", "chat toggle"),
- ("question_mark", "help"),
- ("q", "quit"),
- ]
- assert by_key["ctrl+p"].priority is True
- def test_footer_shows_ordered_entries(server):
- """Left list is the 7 visible bindings (? for help); instructions only right-aligned."""
- pytest.importorskip("textual")
- from textual.widgets import Footer
- from textual.widgets._footer import FooterKey
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path)
- observed: dict = {}
- async def scenario():
- async with app.run_test() as pilot:
- await pilot.pause()
- shown = [
- (b.binding.key, b.binding.description)
- for b in app.screen.active_bindings.values()
- if b.binding.show
- ]
- observed["shown"] = shown
- footer = app.query_one(Footer)
- keys = list(footer.query(FooterKey))
- observed["keys"] = [
- (k.key, k.key_display, k.description, k.action, set(k.classes))
- for k in keys
- ]
- asyncio.run(scenario())
- expected = [
- ("a", "arm command"),
- ("ctrl+r", "execute"),
- ("escape", "cancel"),
- ("e", "edit command"),
- ("t", "chat toggle"),
- ("question_mark", "help"),
- ("q", "quit"),
- ]
- assert observed["shown"] == expected, observed["shown"]
- # Help advertises as "?" in the footer.
- assert app.get_key_display(
- next(b for b in AssistantTUI.BINDINGS if b.key == "question_mark")
- ) == "?"
- # The composed Footer: exactly one "instructions" entry and it is the
- # right-aligned command-palette key; no left duplicate exists.
- instructions = [k for k in observed["keys"] if k[2] == "instructions"]
- assert len(instructions) == 1, observed["keys"]
- key, key_display, _desc, action, classes = instructions[0]
- assert action == "command_palette", instructions
- assert "-command-palette" in classes, instructions
- assert key == "ctrl+p", instructions
- non_palette = [k for k in observed["keys"] if "-command-palette" not in k[4]]
- assert not any(k[2] == "instructions" for k in non_palette), observed["keys"]
- assert not any(k[3] == "command_palette" for k in non_palette), observed["keys"]
- def test_ctrl_p_opens_command_palette(server):
- """ctrl+p still opens the palette; its binding description is 'instructions'."""
- pytest.importorskip("textual")
- from textual.command import CommandPalette
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path)
- observed: dict = {}
- async def scenario():
- async with app.run_test() as pilot:
- await pilot.pause()
- await pilot.press("ctrl+p")
- await pilot.pause()
- await pilot.pause()
- observed["is_palette"] = isinstance(app.screen, CommandPalette)
- observed["stack"] = [type(s).__name__ for s in app.screen_stack]
- asyncio.run(scenario())
- palette_binding = next(b for b in AssistantTUI.BINDINGS if b.key == "ctrl+p")
- assert palette_binding.description == "instructions"
- assert observed["is_palette"], "ctrl+p did not open the command palette: %s" % (
- observed["stack"],
- )
- def test_question_mark_toggles_help_panel(server):
- """? toggles the help panel and is advertised in the left footer list."""
- pytest.importorskip("textual")
- from textual.widgets import HelpPanel
- from stt_daemon.tui.app import AssistantTUI
- path, _service = server
- app = AssistantTUI(path)
- observed: dict = {}
- async def scenario():
- async with app.run_test() as pilot:
- await pilot.pause()
- assert list(app.screen.query(HelpPanel)) == []
- await pilot.press("?")
- await pilot.pause()
- observed["shown"] = list(app.screen.query(HelpPanel)) != []
- await pilot.press("?")
- await pilot.pause()
- observed["hidden"] = list(app.screen.query(HelpPanel)) == []
- asyncio.run(scenario())
- assert observed["shown"] is True, "? did not mount HelpPanel"
- assert observed["hidden"] is True, "second ? did not remove HelpPanel"
- def test_plain_m_maximizes_triage_pane(server):
- pytest.importorskip("textual")
- from stt_daemon.tui.app import AssistantTUI
- from textual.widgets import RichLog
- path, _service = server
- app = AssistantTUI(path, triage_enabled=True)
- async def scenario():
- async with app.run_test(size=(80, 40)) as pilot:
- await pilot.pause()
- app.query_one("#triage", RichLog).focus()
- await pilot.pause()
- await pilot.press("m")
- await pilot.pause()
- assert app.screen.maximized is app.query_one("#main")
- await pilot.press("m")
- await pilot.pause()
- assert app.screen.maximized is None
- assert app.query_one("#triage", RichLog).display is True
- assert app.query_one("#chat", RichLog).display is True
- asyncio.run(scenario())
- def test_screenshot_saves_svg(server, tmp_path, monkeypatch):
- pytest.importorskip("textual")
- import glob
- from stt_daemon.tui.app import AssistantTUI
- monkeypatch.chdir(tmp_path)
- path, _service = server
- app = AssistantTUI(path)
- async def scenario():
- async with app.run_test() as pilot:
- await pilot.pause()
- await pilot.press("ctrl+s")
- await pilot.pause()
- await pilot.pause()
- asyncio.run(scenario())
- files = glob.glob(str(tmp_path / "stt-tui-*.svg"))
- assert files, "no stt-tui-*.svg found in %s" % tmp_path
- assert os.path.getsize(files[0]) > 0
|