STT-Assistant/tests/test_tui_app.py

"""Smoke test: the Textual app mounts, subscribes, and ingests broadcasts."""

import asyncio
import os
import shutil
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_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_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"