STT-Assistant/tests/test_tui_app.py

"""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_enter_sends_execute_and_e_opens_editor(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("enter")
            await pilot.pause()
            await pilot.press("a")
            await pilot.pause()
            await pilot.press("escape")
            await pilot.pause()

    asyncio.run(scenario())
    assert ("command_execute", {}) in sent
    assert ("command_start", {}) in sent
    assert ("command_cancel", {}) in sent


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