test_tui_app.py 36 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059
  1. """Smoke test: the Textual app mounts, subscribes, and ingests broadcasts."""
  2. import asyncio
  3. import os
  4. import shutil
  5. import sys
  6. import tempfile
  7. import pytest
  8. from stt_daemon.assistant.control import ControlServer
  9. from stt_daemon.assistant.service import AssistantService
  10. from stt_daemon.config import Config
  11. class FakeSubscriber:
  12. def events(self):
  13. return iter(())
  14. def close(self):
  15. pass
  16. @pytest.fixture
  17. def server():
  18. directory = tempfile.mkdtemp(prefix="stta-")
  19. path = os.path.join(directory, "c.sock")
  20. cfg = Config()
  21. cfg.assistant.control_socket = path
  22. service = AssistantService(cfg, FakeSubscriber())
  23. control = ControlServer(cfg, service, path=path)
  24. service.set_sinks(emit=control.broadcast, relay=control.relay)
  25. control.start()
  26. yield path, service
  27. control.close()
  28. service.stop()
  29. shutil.rmtree(directory, ignore_errors=True)
  30. def test_ctrl_r_sends_execute_and_e_opens_editor(server):
  31. pytest.importorskip("textual")
  32. from textual.widgets import Input
  33. from stt_daemon.tui.app import AssistantTUI
  34. path, _service = server
  35. app = AssistantTUI(path)
  36. sent: list[tuple] = []
  37. observed: dict = {}
  38. async def scenario():
  39. async with app.run_test() as pilot:
  40. await pilot.pause()
  41. app.client.send = lambda type_, **fields: sent.append((type_, fields))
  42. await pilot.press("ctrl+r")
  43. await pilot.pause()
  44. await pilot.press("a")
  45. await pilot.pause()
  46. await pilot.press("escape")
  47. await pilot.pause()
  48. await pilot.press("e")
  49. await pilot.pause()
  50. edit = app.query_one("#edit", Input)
  51. observed["edit_display"] = edit.display
  52. observed["edit_focused"] = edit.has_focus
  53. asyncio.run(scenario())
  54. assert ("command_execute", {}) in sent
  55. assert ("command_start", {}) in sent
  56. assert ("command_cancel", {}) in sent
  57. assert observed["edit_display"] is True
  58. assert observed["edit_focused"] is True
  59. def test_c_key_confirms_before_clearing(server):
  60. pytest.importorskip("textual")
  61. from stt_daemon.tui.app import AssistantTUI
  62. path, _service = server
  63. app = AssistantTUI(path)
  64. sent: list[tuple] = []
  65. async def scenario():
  66. async with app.run_test() as pilot:
  67. await pilot.pause()
  68. app.client.send = lambda type_, **fields: sent.append((type_, fields))
  69. await pilot.press("c")
  70. await pilot.pause()
  71. assert not any(type_ == "clear" for type_, _ in sent), "clear sent before confirmation"
  72. await pilot.press("y")
  73. await pilot.pause()
  74. asyncio.run(scenario())
  75. assert ("clear", {}) in sent
  76. def test_c_key_cancel_does_not_clear(server):
  77. pytest.importorskip("textual")
  78. from stt_daemon.tui.app import AssistantTUI
  79. path, _service = server
  80. app = AssistantTUI(path)
  81. sent: list[tuple] = []
  82. async def scenario():
  83. async with app.run_test() as pilot:
  84. await pilot.pause()
  85. app.client.send = lambda type_, **fields: sent.append((type_, fields))
  86. await pilot.press("c")
  87. await pilot.pause()
  88. await pilot.press("n")
  89. await pilot.pause()
  90. asyncio.run(scenario())
  91. assert not any(type_ == "clear" for type_, _ in sent)
  92. def test_c_key_escape_does_not_clear(server):
  93. pytest.importorskip("textual")
  94. from stt_daemon.tui.app import AssistantTUI
  95. path, _service = server
  96. app = AssistantTUI(path)
  97. sent: list[tuple] = []
  98. async def scenario():
  99. async with app.run_test() as pilot:
  100. await pilot.pause()
  101. app.client.send = lambda type_, **fields: sent.append((type_, fields))
  102. await pilot.press("c")
  103. await pilot.pause()
  104. await pilot.press("escape")
  105. await pilot.pause()
  106. asyncio.run(scenario())
  107. assert not any(type_ == "clear" for type_, _ in sent)
  108. def test_c_key_clears_rendered_output(server):
  109. """Pressing 'c' clears the conversation pane via the cleared event."""
  110. pytest.importorskip("textual")
  111. from textual.widgets import RichLog
  112. from stt_daemon.tui.app import AssistantTUI
  113. path, _service = server
  114. app = AssistantTUI(path)
  115. async def scenario():
  116. async with app.run_test() as pilot:
  117. await pilot.pause()
  118. app.state.ingest({"type": "chat_message", "role": "user", "text": "hello"})
  119. app.state.ingest(
  120. {
  121. "type": "tool_result",
  122. "tool": "get_datetime",
  123. "ok": True,
  124. "result": "2026-09-21",
  125. }
  126. )
  127. app._render()
  128. await pilot.pause()
  129. app.state.ingest({"type": "cleared"})
  130. app._render()
  131. await pilot.pause()
  132. log = app.query_one("#chat", RichLog)
  133. return list(log.lines)
  134. lines = asyncio.run(scenario())
  135. assert not any("hello" in str(line) for line in lines)
  136. assert not any("2026-09-21" in str(line) for line in lines)
  137. def test_tui_renders_command_output(server):
  138. """A tool result must actually reach a widget - otherwise 'nothing happens'."""
  139. pytest.importorskip("textual")
  140. from textual.widgets import RichLog
  141. from stt_daemon.tui.app import AssistantTUI
  142. path, _service = server
  143. app = AssistantTUI(path)
  144. async def scenario():
  145. async with app.run_test() as pilot:
  146. await pilot.pause()
  147. app.state.ingest(
  148. {
  149. "type": "tool_result",
  150. "tool": "get_datetime",
  151. "ok": True,
  152. "result": "2026-09-21T01:00:00",
  153. }
  154. )
  155. app._render()
  156. await pilot.pause()
  157. log = app.query_one("#chat", RichLog)
  158. return list(log.lines)
  159. lines = asyncio.run(scenario())
  160. assert lines, "command output was not rendered anywhere"
  161. def test_tui_clears_chat_pane_on_chat_exit(server):
  162. """Leaving chat must drop the rendered conversation but keep command output."""
  163. pytest.importorskip("textual")
  164. from textual.widgets import RichLog
  165. from stt_daemon.tui.app import AssistantTUI
  166. path, _service = server
  167. app = AssistantTUI(path)
  168. async def scenario():
  169. async with app.run_test() as pilot:
  170. await pilot.pause()
  171. app.state.ingest(
  172. {"type": "tool_result", "tool": "get_datetime", "ok": True, "result": "kept"}
  173. )
  174. app.state.ingest({"type": "mode", "mode": "chat"})
  175. app.state.ingest({"type": "chat_message", "role": "user", "text": "gone"})
  176. app._render()
  177. await pilot.pause()
  178. app.state.ingest({"type": "mode", "mode": "idle"})
  179. app._render()
  180. await pilot.pause()
  181. return [str(line) for line in app.query_one("#chat", RichLog).lines]
  182. lines = asyncio.run(scenario())
  183. assert not any("gone" in line for line in lines)
  184. assert any("kept" in line for line in lines)
  185. def test_tui_mounts_subscribes_and_ingests_broadcasts(server):
  186. pytest.importorskip("textual")
  187. from stt_daemon.tui.app import AssistantTUI
  188. path, service = server
  189. app = AssistantTUI(path)
  190. async def scenario():
  191. async with app.run_test() as pilot:
  192. await pilot.pause()
  193. service.arm() # broadcast to the subscribed viewer
  194. for _ in range(60):
  195. await pilot.pause()
  196. if app.state.mode == "command_armed":
  197. break
  198. await asyncio.sleep(0.05)
  199. asyncio.run(scenario())
  200. assert app.state.mode == "command_armed"
  201. def test_render_after_unmount_is_safe(server):
  202. """A timer tick can land during teardown; rendering must not raise."""
  203. pytest.importorskip("textual")
  204. from stt_daemon.tui.app import AssistantTUI
  205. path, _service = server
  206. app = AssistantTUI(path)
  207. async def scenario():
  208. async with app.run_test() as pilot:
  209. await pilot.pause()
  210. asyncio.run(scenario())
  211. app._render() # must not raise NoMatches
  212. def test_tui_records_connect_failure(tmp_path):
  213. pytest.importorskip("textual")
  214. from stt_daemon.tui.app import AssistantTUI
  215. app = AssistantTUI(str(tmp_path / "missing.sock"))
  216. async def scenario():
  217. async with app.run_test() as pilot:
  218. await pilot.pause()
  219. asyncio.run(scenario())
  220. assert app.connect_error is not None
  221. assert "cannot connect" in app.connect_error
  222. def test_border_titles_show_queue_and_inference_depths(server):
  223. """Border titles for transcript and chat show live segment and inference queue counts."""
  224. pytest.importorskip("textual")
  225. from textual.widgets import RichLog
  226. from stt_daemon.tui.app import AssistantTUI
  227. path, _service = server
  228. app = AssistantTUI(path)
  229. async def scenario():
  230. async with app.run_test() as pilot:
  231. await pilot.pause()
  232. app.state.ingest(
  233. {
  234. "type": "heartbeat",
  235. "queue_depth": 5,
  236. "inference_depth": 3,
  237. "inference_in_flight": False,
  238. }
  239. )
  240. app._render()
  241. await pilot.pause()
  242. transcript_title = app.query_one("#transcript", RichLog).border_title
  243. chat_title = app.query_one("#chat", RichLog).border_title
  244. assert "[seg 5]" in transcript_title
  245. assert "[inf 3]" in chat_title
  246. assert "busy" not in chat_title
  247. # Test with inference in flight
  248. app.state.ingest(
  249. {
  250. "type": "heartbeat",
  251. "inference_depth": 7,
  252. "inference_in_flight": True,
  253. }
  254. )
  255. app._render()
  256. await pilot.pause()
  257. chat_title = app.query_one("#chat", RichLog).border_title
  258. assert "[inf 7" in chat_title
  259. assert "busy" in chat_title
  260. asyncio.run(scenario())
  261. def test_main_exit_code_reflects_connect_failure(monkeypatch, tmp_path, capsys):
  262. pytest.importorskip("textual")
  263. from stt_daemon.tui import app as tui_app
  264. monkeypatch.setattr(
  265. tui_app.AssistantTUI,
  266. "run",
  267. lambda self: setattr(self, "connect_error", "boom"),
  268. )
  269. monkeypatch.setattr(sys, "argv", ["stt-tui", "--socket", str(tmp_path / "x.sock")])
  270. assert tui_app.main() == 1
  271. assert "boom" in capsys.readouterr().err
  272. monkeypatch.setattr(tui_app.AssistantTUI, "run", lambda self: None)
  273. assert tui_app.main() == 0
  274. def test_triage_disabled_composes_no_triage_widget(server):
  275. """Default (triage_enabled=False): no #triage, no #main — tree as today."""
  276. pytest.importorskip("textual")
  277. from stt_daemon.tui.app import AssistantTUI
  278. path, _service = server
  279. app = AssistantTUI(path)
  280. async def scenario():
  281. async with app.run_test() as pilot:
  282. await pilot.pause()
  283. assert not app.query("#triage")
  284. assert not app.query("#main")
  285. # The rest of the tree is intact.
  286. app.query_one("#transcript")
  287. app.query_one("#status")
  288. app.query_one("#command")
  289. app.query_one("#chat")
  290. app.query_one("#edit")
  291. asyncio.run(scenario())
  292. def test_triage_enabled_composes_split(server):
  293. """triage_enabled=True: #main wraps #triage (left) + #chat (right); transcript stays outside."""
  294. pytest.importorskip("textual")
  295. from textual.containers import Horizontal
  296. from textual.widgets import RichLog
  297. from stt_daemon.tui.app import AssistantTUI
  298. path, _service = server
  299. app = AssistantTUI(path, triage_enabled=True)
  300. async def scenario():
  301. async with app.run_test() as pilot:
  302. await pilot.pause()
  303. main = app.query_one("#main", Horizontal)
  304. transcript = app.query_one("#transcript", RichLog)
  305. triage = app.query_one("#triage", RichLog)
  306. chat = app.query_one("#chat", RichLog)
  307. assert triage.parent is main
  308. assert chat.parent is main
  309. assert transcript.parent is not main
  310. assert triage.border_title == "triage"
  311. # triage sits LEFT of chat inside #main.
  312. assert main.children.index(triage) < main.children.index(chat)
  313. # Outside widgets stay outside #main.
  314. assert app.query_one("#status").parent is not main
  315. assert transcript.parent is not main
  316. asyncio.run(scenario())
  317. def test_triage_enabled_layout_proportions(server):
  318. """Rendered proportions: #triage ~1/3 and #chat ~2/3 of #main; #transcript ~25% of screen."""
  319. pytest.importorskip("textual")
  320. from textual.containers import Horizontal
  321. from textual.widgets import RichLog
  322. from stt_daemon.tui.app import AssistantTUI
  323. path, _service = server
  324. app = AssistantTUI(path, triage_enabled=True)
  325. async def scenario(size=None):
  326. kwargs = {"size": size} if size else {}
  327. async with app.run_test(**kwargs) as pilot:
  328. await pilot.pause()
  329. main = app.query_one("#main", Horizontal)
  330. transcript = app.query_one("#transcript", RichLog)
  331. triage = app.query_one("#triage", RichLog)
  332. chat = app.query_one("#chat", RichLog)
  333. return (
  334. main.size.width,
  335. triage.size.width,
  336. chat.size.width,
  337. app.screen.size.height,
  338. transcript.region.height,
  339. )
  340. main_w, triage_w, chat_w, screen_h, transcript_h = asyncio.run(scenario())
  341. print(
  342. f"measured: main_w={main_w} triage_w={triage_w} chat_w={chat_w} "
  343. f"screen_h={screen_h} transcript_h={transcript_h}"
  344. )
  345. # Content boxes exclude the 1-cell border on each side; allow ±2 px for
  346. # borders/rounding on the fractional split.
  347. assert abs(triage_w / main_w - 1 / 3) < 0.05, (main_w, triage_w, chat_w)
  348. assert abs(chat_w / main_w - 2 / 3) < 0.05, (main_w, triage_w, chat_w)
  349. # region.height includes the transcript's 2 border rows; at the small
  350. # default size rounding is coarse, so only demand a floor here. The
  351. # precise 25% check runs at a taller size below.
  352. assert transcript_h / screen_h >= 0.18, (screen_h, transcript_h)
  353. assert transcript_h >= 1, transcript_h
  354. app2 = AssistantTUI(path, triage_enabled=True)
  355. async def scenario_tall():
  356. async with app2.run_test(size=(80, 40)) as pilot:
  357. await pilot.pause()
  358. transcript2 = app2.query_one("#transcript", RichLog)
  359. return app2.screen.size.height, transcript2.region.height
  360. screen_h2, transcript_h2 = asyncio.run(scenario_tall())
  361. print(f"measured tall: screen_h={screen_h2} transcript_h={transcript_h2}")
  362. assert abs(transcript_h2 / screen_h2 - 0.25) < 0.06, (screen_h2, transcript_h2)
  363. def test_triage_enabled_renders_signals_on_epoch_change(server):
  364. """A triage event + status event must actually write rendered lines into #triage."""
  365. pytest.importorskip("textual")
  366. from textual.widgets import RichLog
  367. from stt_daemon.tui.app import AssistantTUI
  368. path, _service = server
  369. app = AssistantTUI(path, triage_enabled=True)
  370. async def scenario():
  371. async with app.run_test() as pilot:
  372. await pilot.pause()
  373. app._timer.stop() # no background drain: server status must not clobber state
  374. app._drain()
  375. app.state.ingest(
  376. {
  377. "type": "triage",
  378. "stage": "signals",
  379. "triage": {
  380. "order": ["category"],
  381. "items": {
  382. "category": {
  383. "kind": "choice",
  384. "choice": "question",
  385. "answer_confidence": 0.91,
  386. }
  387. },
  388. },
  389. }
  390. )
  391. app.state.ingest(
  392. {"type": "status", "triage_state": "ok", "triage_reason": ""}
  393. )
  394. app._render()
  395. await pilot.pause()
  396. return [str(line) for line in app.query_one("#triage", RichLog).lines]
  397. lines = asyncio.run(scenario())
  398. rendered = "\n".join(lines)
  399. assert "category: question (0.91)" in rendered
  400. def test_triage_enabled_loading_shows_reason_line_never_blank(server):
  401. """loading/unknown triage_state renders the reason line — never a blank panel."""
  402. pytest.importorskip("textual")
  403. from textual.widgets import RichLog
  404. from stt_daemon.tui.app import AssistantTUI
  405. path, _service = server
  406. app = AssistantTUI(path, triage_enabled=True)
  407. async def scenario():
  408. async with app.run_test() as pilot:
  409. await pilot.pause()
  410. app._timer.stop() # no background drain: server status must not clobber state
  411. app._drain()
  412. app.state.ingest(
  413. {
  414. "type": "status",
  415. "triage_state": "loading",
  416. "triage_reason": "model loading",
  417. }
  418. )
  419. app._render()
  420. await pilot.pause()
  421. loading_lines = [str(line) for line in app.query_one("#triage", RichLog).lines]
  422. # Unknown state falls back to a generic reason line, no crash.
  423. app.state.ingest(
  424. {"type": "status", "triage_state": "weird_state", "triage_reason": ""}
  425. )
  426. app._render()
  427. await pilot.pause()
  428. unknown_lines = [str(line) for line in app.query_one("#triage", RichLog).lines]
  429. return loading_lines, unknown_lines
  430. loading_lines, unknown_lines = asyncio.run(scenario())
  431. assert loading_lines, "triage panel was blank in loading state"
  432. assert any("triage: unavailable" in line for line in loading_lines)
  433. assert any("model loading" in line for line in loading_lines)
  434. assert unknown_lines, "triage panel was blank for unknown state"
  435. assert any("triage: unavailable" in line for line in unknown_lines)
  436. assert any("weird_state" in line for line in unknown_lines)
  437. def test_triage_epoch_change_clears_before_rewrite(server):
  438. """Epoch bump clears #triage first: no duplicated stale lines."""
  439. pytest.importorskip("textual")
  440. from textual.widgets import RichLog
  441. from stt_daemon.tui.app import AssistantTUI
  442. path, _service = server
  443. app = AssistantTUI(path, triage_enabled=True)
  444. async def scenario():
  445. async with app.run_test() as pilot:
  446. await pilot.pause()
  447. app._timer.stop() # no background drain: server status must not clobber state
  448. app._drain()
  449. app.state.ingest(
  450. {
  451. "type": "triage",
  452. "stage": "signals",
  453. "triage": {"order": ["q1"], "items": {"q1": {"kind": "choice", "choice": "a"}}},
  454. }
  455. )
  456. app.state.ingest({"type": "status", "triage_state": "ok"})
  457. app._render()
  458. await pilot.pause()
  459. # New epoch: cleared stage empties the payload, panel must not keep old lines.
  460. app.state.ingest({"type": "triage", "stage": "cleared"})
  461. app._render()
  462. await pilot.pause()
  463. return [str(line) for line in app.query_one("#triage", RichLog).lines]
  464. lines = asyncio.run(scenario())
  465. rendered = "\n".join(lines)
  466. assert "q1" not in rendered, "stale triage lines survived an epoch change"
  467. assert "triage: unavailable — no data" in rendered
  468. def test_triage_disabled_never_writes_triage(server):
  469. """Disabled: triage events are ingested but #triage is never touched (it doesn't exist)."""
  470. pytest.importorskip("textual")
  471. from stt_daemon.tui.app import AssistantTUI
  472. path, _service = server
  473. app = AssistantTUI(path)
  474. async def scenario():
  475. async with app.run_test() as pilot:
  476. await pilot.pause()
  477. app.state.ingest({"type": "triage", "stage": "signals", "triage": {}})
  478. app.state.ingest({"type": "status", "triage_state": "ok"})
  479. app._render() # must not raise NoMatches for #triage
  480. await pilot.pause()
  481. assert not app.query("#triage")
  482. asyncio.run(scenario())
  483. def test_default_ctor_has_no_triage_arg(server):
  484. """AssistantTUI(path) stays valid: triage_enabled defaults to False."""
  485. pytest.importorskip("textual")
  486. from stt_daemon.tui.app import AssistantTUI
  487. path, _service = server
  488. app = AssistantTUI(path)
  489. assert app.triage_enabled is False
  490. assert app._seen_triage_epoch == 0
  491. def test_maximize_triage_hides_chat_and_fills_screen(server):
  492. """Focus #triage, press m: #main maximized, chat hidden, triage fills screen."""
  493. pytest.importorskip("textual")
  494. from textual.widgets import RichLog
  495. from stt_daemon.tui.app import AssistantTUI
  496. path, _service = server
  497. app = AssistantTUI(path, triage_enabled=True)
  498. observed: dict = {}
  499. async def scenario():
  500. async with app.run_test(size=(80, 40)) as pilot:
  501. await pilot.pause()
  502. app.query_one("#triage", RichLog).focus()
  503. await pilot.pause()
  504. await pilot.press("m")
  505. await pilot.pause()
  506. observed["maximized_is_main"] = app.screen.maximized is app.query_one("#main")
  507. observed["triage_display"] = app.query_one("#triage", RichLog).display
  508. observed["chat_display"] = app.query_one("#chat", RichLog).display
  509. observed["region"] = app.query_one("#triage", RichLog).region
  510. observed["screen_size"] = app.screen.size
  511. asyncio.run(scenario())
  512. print("triage maximized: region=%s screen=%s" % (observed["region"], observed["screen_size"]))
  513. assert observed["maximized_is_main"] is True
  514. assert observed["triage_display"] is True
  515. assert observed["chat_display"] is False
  516. region = observed["region"]
  517. size = observed["screen_size"]
  518. assert abs(region.width - size.width) <= 2, (region, size)
  519. assert abs(region.height - size.height) <= 2, (region, size)
  520. def test_maximize_toggle_restores_layout(server):
  521. """Press m again: minimize restores both panes."""
  522. pytest.importorskip("textual")
  523. from textual.widgets import RichLog
  524. from stt_daemon.tui.app import AssistantTUI
  525. path, _service = server
  526. app = AssistantTUI(path, triage_enabled=True)
  527. observed: dict = {}
  528. async def scenario():
  529. async with app.run_test(size=(80, 40)) as pilot:
  530. await pilot.pause()
  531. app.query_one("#triage", RichLog).focus()
  532. await pilot.pause()
  533. await pilot.press("m")
  534. await pilot.pause()
  535. await pilot.press("m")
  536. await pilot.pause()
  537. observed["maximized"] = app.screen.maximized
  538. observed["triage_display"] = app.query_one("#triage", RichLog).display
  539. observed["chat_display"] = app.query_one("#chat", RichLog).display
  540. asyncio.run(scenario())
  541. assert observed["maximized"] is None
  542. assert observed["triage_display"] is True
  543. assert observed["chat_display"] is True
  544. def test_maximize_transcript_fills_screen(server):
  545. """Focus #transcript, press m: transcript maximized and fills the screen."""
  546. pytest.importorskip("textual")
  547. from textual.widgets import RichLog
  548. from stt_daemon.tui.app import AssistantTUI
  549. path, _service = server
  550. app = AssistantTUI(path, triage_enabled=True)
  551. observed: dict = {}
  552. async def scenario():
  553. async with app.run_test(size=(80, 40)) as pilot:
  554. await pilot.pause()
  555. app.query_one("#transcript", RichLog).focus()
  556. await pilot.pause()
  557. await pilot.press("m")
  558. await pilot.pause()
  559. observed["maximized_is_transcript"] = (
  560. app.screen.maximized is app.query_one("#transcript")
  561. )
  562. observed["region"] = app.query_one("#transcript", RichLog).region
  563. observed["screen_size"] = app.screen.size
  564. await pilot.press("m")
  565. await pilot.pause()
  566. observed["restored"] = app.screen.maximized
  567. asyncio.run(scenario())
  568. print("transcript maximized: region=%s screen=%s"
  569. % (observed["region"], observed["screen_size"]))
  570. assert observed["maximized_is_transcript"] is True
  571. region = observed["region"]
  572. size = observed["screen_size"]
  573. assert abs(region.width - size.width) <= 2, (region, size)
  574. assert abs(region.height - size.height) <= 2, (region, size)
  575. assert observed["restored"] is None
  576. def test_escape_while_maximized_restores_panes(server):
  577. """Escape minimizes (ESCAPE_TO_MINIMIZE) and _drain restores the hidden sibling."""
  578. pytest.importorskip("textual")
  579. from textual.widgets import RichLog
  580. from stt_daemon.tui.app import AssistantTUI
  581. path, _service = server
  582. app = AssistantTUI(path, triage_enabled=True)
  583. observed: dict = {}
  584. async def scenario():
  585. async with app.run_test(size=(80, 40)) as pilot:
  586. await pilot.pause()
  587. app.query_one("#triage", RichLog).focus()
  588. await pilot.pause()
  589. await pilot.press("m")
  590. await pilot.pause()
  591. assert app.screen.maximized is not None
  592. await pilot.press("escape")
  593. await pilot.pause()
  594. # Let the 0.1s _drain timer run so the restore path executes.
  595. for _ in range(10):
  596. await pilot.pause()
  597. await asyncio.sleep(0.05)
  598. if app.screen.maximized is None and app.query_one("#chat", RichLog).display:
  599. break
  600. observed["maximized"] = app.screen.maximized
  601. observed["triage_display"] = app.query_one("#triage", RichLog).display
  602. observed["chat_display"] = app.query_one("#chat", RichLog).display
  603. asyncio.run(scenario())
  604. assert observed["maximized"] is None
  605. assert observed["triage_display"] is True
  606. assert observed["chat_display"] is True
  607. def test_maximize_chat_when_triage_disabled(server):
  608. """triage_enabled=False: focus #chat, press m maximizes #main without crashing."""
  609. pytest.importorskip("textual")
  610. from textual.widgets import RichLog
  611. from stt_daemon.tui.app import AssistantTUI
  612. path, _service = server
  613. app = AssistantTUI(path)
  614. observed: dict = {}
  615. async def scenario():
  616. async with app.run_test(size=(80, 40)) as pilot:
  617. await pilot.pause()
  618. # Disabled: no #main/#triage; focusing #chat then m maximizes #chat itself.
  619. app.query_one("#chat", RichLog).focus()
  620. await pilot.pause()
  621. await pilot.press("m")
  622. await pilot.pause()
  623. observed["maximized"] = app.screen.maximized
  624. observed["chat"] = app.query_one("#chat", RichLog)
  625. observed["region"] = app.query_one("#chat", RichLog).region
  626. observed["screen_size"] = app.screen.size
  627. await pilot.press("m")
  628. await pilot.pause()
  629. observed["restored"] = app.screen.maximized
  630. asyncio.run(scenario())
  631. print("chat maximized (disabled): region=%s screen=%s"
  632. % (observed["region"], observed["screen_size"]))
  633. assert observed["maximized"] is observed["chat"]
  634. assert observed["restored"] is None
  635. def test_footer_hidden_while_maximized(server):
  636. """ALLOW_IN_MAXIMIZED_VIEW hides the Footer while a pane is maximized."""
  637. pytest.importorskip("textual")
  638. from textual.widgets import Footer, RichLog
  639. from stt_daemon.tui.app import AssistantTUI
  640. path, _service = server
  641. app = AssistantTUI(path, triage_enabled=True)
  642. observed: dict = {}
  643. async def scenario():
  644. async with app.run_test(size=(80, 40)) as pilot:
  645. await pilot.pause()
  646. app.query_one("#triage", RichLog).focus()
  647. await pilot.pause()
  648. await pilot.press("m")
  649. await pilot.pause()
  650. observed["maximized"] = app.screen.maximized
  651. observed["footer_region"] = app.query_one(Footer).region
  652. await pilot.press("m")
  653. await pilot.pause()
  654. asyncio.run(scenario())
  655. assert observed["maximized"] is not None
  656. footer_region = observed["footer_region"]
  657. assert footer_region.height == 0 or footer_region.width == 0, observed
  658. def test_question_mark_key_toggles_help_panel(server):
  659. pytest.importorskip("textual")
  660. from textual.widgets import HelpPanel
  661. from stt_daemon.tui.app import AssistantTUI
  662. path, _service = server
  663. app = AssistantTUI(path)
  664. observed: dict = {}
  665. async def scenario():
  666. async with app.run_test() as pilot:
  667. await pilot.pause()
  668. assert list(app.screen.query(HelpPanel)) == []
  669. await pilot.press("question_mark")
  670. await pilot.pause()
  671. observed["shown"] = list(app.screen.query(HelpPanel)) != []
  672. app.screen.query_one(HelpPanel) # must succeed while shown
  673. await pilot.press("question_mark")
  674. await pilot.pause()
  675. observed["hidden"] = list(app.screen.query(HelpPanel)) == []
  676. asyncio.run(scenario())
  677. assert observed["shown"] is True, "? did not mount HelpPanel"
  678. assert observed["hidden"] is True, "second ? did not remove HelpPanel"
  679. def test_app_bindings_list():
  680. """BINDINGS: 7 visible footer entries; hidden ones after; no h binding."""
  681. pytest.importorskip("textual")
  682. from stt_daemon.tui.app import AssistantTUI
  683. pairs = [(b.key, b.description) for b in AssistantTUI.BINDINGS]
  684. assert ("ctrl+r", "execute") in pairs
  685. assert ("m", "maximize pane") in pairs
  686. assert ("question_mark", "help") in pairs
  687. assert ("ctrl+s", "save screenshot") in pairs
  688. assert ("a", "arm command") in pairs
  689. assert ("escape", "cancel") in pairs
  690. assert ("e", "edit command") in pairs
  691. assert ("t", "chat toggle") in pairs
  692. assert ("c", "clear") in pairs
  693. assert ("q", "quit") in pairs
  694. assert ("ctrl+p", "instructions") in pairs
  695. assert not any(key == "h" for key, _ in pairs)
  696. assert not any(key == "ctrl+m" for key, _ in pairs)
  697. assert not any(key == "ctrl+h" for key, _ in pairs)
  698. assert not any(key == "enter" for key, _ in pairs)
  699. assert not any(key == "ctrl+k" for key, _ in pairs)
  700. by_key = {b.key: b for b in AssistantTUI.BINDINGS}
  701. assert by_key["question_mark"].show is True
  702. assert by_key["m"].show is False
  703. assert by_key["c"].show is False
  704. assert by_key["ctrl+s"].show is False
  705. assert by_key["ctrl+p"].show is False
  706. # Visible seven first, in footer order.
  707. visible = [(b.key, b.description) for b in AssistantTUI.BINDINGS if b.show]
  708. assert visible == [
  709. ("a", "arm command"),
  710. ("ctrl+r", "execute"),
  711. ("escape", "cancel"),
  712. ("e", "edit command"),
  713. ("t", "chat toggle"),
  714. ("question_mark", "help"),
  715. ("q", "quit"),
  716. ]
  717. assert by_key["ctrl+p"].priority is True
  718. def test_footer_shows_ordered_entries(server):
  719. """Left list is the 7 visible bindings (? for help); instructions only right-aligned."""
  720. pytest.importorskip("textual")
  721. from textual.widgets import Footer
  722. from textual.widgets._footer import FooterKey
  723. from stt_daemon.tui.app import AssistantTUI
  724. path, _service = server
  725. app = AssistantTUI(path)
  726. observed: dict = {}
  727. async def scenario():
  728. async with app.run_test() as pilot:
  729. await pilot.pause()
  730. shown = [
  731. (b.binding.key, b.binding.description)
  732. for b in app.screen.active_bindings.values()
  733. if b.binding.show
  734. ]
  735. observed["shown"] = shown
  736. footer = app.query_one(Footer)
  737. keys = list(footer.query(FooterKey))
  738. observed["keys"] = [
  739. (k.key, k.key_display, k.description, k.action, set(k.classes))
  740. for k in keys
  741. ]
  742. asyncio.run(scenario())
  743. expected = [
  744. ("a", "arm command"),
  745. ("ctrl+r", "execute"),
  746. ("escape", "cancel"),
  747. ("e", "edit command"),
  748. ("t", "chat toggle"),
  749. ("question_mark", "help"),
  750. ("q", "quit"),
  751. ]
  752. assert observed["shown"] == expected, observed["shown"]
  753. # Help advertises as "?" in the footer.
  754. assert app.get_key_display(
  755. next(b for b in AssistantTUI.BINDINGS if b.key == "question_mark")
  756. ) == "?"
  757. # The composed Footer: exactly one "instructions" entry and it is the
  758. # right-aligned command-palette key; no left duplicate exists.
  759. instructions = [k for k in observed["keys"] if k[2] == "instructions"]
  760. assert len(instructions) == 1, observed["keys"]
  761. key, key_display, _desc, action, classes = instructions[0]
  762. assert action == "command_palette", instructions
  763. assert "-command-palette" in classes, instructions
  764. assert key == "ctrl+p", instructions
  765. non_palette = [k for k in observed["keys"] if "-command-palette" not in k[4]]
  766. assert not any(k[2] == "instructions" for k in non_palette), observed["keys"]
  767. assert not any(k[3] == "command_palette" for k in non_palette), observed["keys"]
  768. def test_ctrl_p_opens_command_palette(server):
  769. """ctrl+p still opens the palette; its binding description is 'instructions'."""
  770. pytest.importorskip("textual")
  771. from textual.command import CommandPalette
  772. from stt_daemon.tui.app import AssistantTUI
  773. path, _service = server
  774. app = AssistantTUI(path)
  775. observed: dict = {}
  776. async def scenario():
  777. async with app.run_test() as pilot:
  778. await pilot.pause()
  779. await pilot.press("ctrl+p")
  780. await pilot.pause()
  781. await pilot.pause()
  782. observed["is_palette"] = isinstance(app.screen, CommandPalette)
  783. observed["stack"] = [type(s).__name__ for s in app.screen_stack]
  784. asyncio.run(scenario())
  785. palette_binding = next(b for b in AssistantTUI.BINDINGS if b.key == "ctrl+p")
  786. assert palette_binding.description == "instructions"
  787. assert observed["is_palette"], "ctrl+p did not open the command palette: %s" % (
  788. observed["stack"],
  789. )
  790. def test_question_mark_toggles_help_panel(server):
  791. """? toggles the help panel and is advertised in the left footer list."""
  792. pytest.importorskip("textual")
  793. from textual.widgets import HelpPanel
  794. from stt_daemon.tui.app import AssistantTUI
  795. path, _service = server
  796. app = AssistantTUI(path)
  797. observed: dict = {}
  798. async def scenario():
  799. async with app.run_test() as pilot:
  800. await pilot.pause()
  801. assert list(app.screen.query(HelpPanel)) == []
  802. await pilot.press("?")
  803. await pilot.pause()
  804. observed["shown"] = list(app.screen.query(HelpPanel)) != []
  805. await pilot.press("?")
  806. await pilot.pause()
  807. observed["hidden"] = list(app.screen.query(HelpPanel)) == []
  808. asyncio.run(scenario())
  809. assert observed["shown"] is True, "? did not mount HelpPanel"
  810. assert observed["hidden"] is True, "second ? did not remove HelpPanel"
  811. def test_plain_m_maximizes_triage_pane(server):
  812. pytest.importorskip("textual")
  813. from stt_daemon.tui.app import AssistantTUI
  814. from textual.widgets import RichLog
  815. path, _service = server
  816. app = AssistantTUI(path, triage_enabled=True)
  817. async def scenario():
  818. async with app.run_test(size=(80, 40)) as pilot:
  819. await pilot.pause()
  820. app.query_one("#triage", RichLog).focus()
  821. await pilot.pause()
  822. await pilot.press("m")
  823. await pilot.pause()
  824. assert app.screen.maximized is app.query_one("#main")
  825. await pilot.press("m")
  826. await pilot.pause()
  827. assert app.screen.maximized is None
  828. assert app.query_one("#triage", RichLog).display is True
  829. assert app.query_one("#chat", RichLog).display is True
  830. asyncio.run(scenario())
  831. def test_screenshot_saves_svg(server, tmp_path, monkeypatch):
  832. pytest.importorskip("textual")
  833. import glob
  834. from stt_daemon.tui.app import AssistantTUI
  835. monkeypatch.chdir(tmp_path)
  836. path, _service = server
  837. app = AssistantTUI(path)
  838. async def scenario():
  839. async with app.run_test() as pilot:
  840. await pilot.pause()
  841. await pilot.press("ctrl+s")
  842. await pilot.pause()
  843. await pilot.pause()
  844. asyncio.run(scenario())
  845. files = glob.glob(str(tmp_path / "stt-tui-*.svg"))
  846. assert files, "no stt-tui-*.svg found in %s" % tmp_path
  847. assert os.path.getsize(files[0]) > 0