NexusOS/tests/test_tui.py

"""TUI helpers and headless App.run_test coverage."""
from __future__ import annotations

import asyncio
import threading

import pytest
from rich.text import Text

from nexusos_cli.tui_app import (
    _compact_status,
    _deny_tool_request,
    _escape,
    _status_line,
    format_assistant_line,
    format_user_line,
)


class _ApprovalResponse:
    def raise_for_status(self):
        return None


class _ApprovalClient:
    calls = []

    def __init__(self, **kwargs):
        self.kwargs = kwargs

    def __enter__(self):
        return self

    def __exit__(self, *args):
        return None

    def post(self, path, *, json):
        self.calls.append((path, json, self.kwargs))
        return _ApprovalResponse()


def test_status_line_mentions_services():
    snap = {
        "version": "1.0.0",
        "services": {
            "backend": {"running": True},
            "memory": {"running": False},
            "provider": {"reachable": True},
        },
        "api": {"online": True, "action_tool_policy": "ask"},
        "host": {"cpu_pct": 10.0},
        "toolchains": [{"lang": "python", "ready": True}],
    }
    line = _status_line(snap)
    assert "backend=UP" in line
    assert "memory=DOWN" in line
    assert "provider=UP" in line
    assert "tools=ask" in line
    assert "run=python" in line


def test_compact_status_handles_api_down():
    snap = {
        "host": {},
        "api": {"online": False},
        "recent_tools": [],
    }
    assert "api DOWN" in _compact_status(snap)


def test_escape_preserves_code_brackets_in_display():
    raw = "idx = arr[i] and rng = [a-z]+"
    plain = Text.from_markup(format_assistant_line(raw)).plain
    assert "arr[i]" in plain
    assert "[a-z]+" in plain
    # Unescaped markup would drop the bracket contents.
    assert plain != "nexus> idx = arr and rng = +"


def test_closing_tag_in_model_output_does_not_raise():
    raw = "close with [/] please"
    plain = Text.from_markup(format_assistant_line(raw)).plain
    assert "[/]" in plain


def test_user_line_escapes_markup():
    plain = Text.from_markup(format_user_line("use [bold] please")).plain
    assert "[bold]" in plain


def test_finish_stream_markup_does_not_wedge_busy():
    """A stray '[/]' used to raise before _busy=False and lock the TUI forever."""
    pytest.importorskip("textual")
    from nexusos_cli.tui_app import NexusTUI

    app = NexusTUI.build_app(api_url="http://127.0.0.1:9")

    async def _run():
        async with app.run_test():
            app._busy = True
            app._finish_stream("see [/] and arr[i]")
            assert app._busy is False
            assert app.history[-1]["content"] == "see [/] and arr[i]"

    asyncio.run(_run())


def test_stream_error_remains_visible_after_finish():
    pytest.importorskip("textual")
    from nexusos_cli.tui_app import NexusTUI

    app = NexusTUI.build_app(api_url="http://127.0.0.1:9")

    async def _run():
        async with app.run_test():
            app._busy = True
            app._show_error("[red]Backend not reachable[/]")
            app._finish_stream("")
            log = app.query_one("#log")
            assert any("Backend not reachable" in line.text for line in log.lines)
            assert app._busy is False

    asyncio.run(_run())


def test_tool_request_is_denied_with_stream_token():
    _ApprovalClient.calls.clear()
    names = _deny_tool_request(
        api_url="http://localhost:8000",
        conversation_id="conversation-1",
        payload='{"token":"secret","actions":[{"name":"run_snippet"}]}',
        client_factory=_ApprovalClient,
    )

    assert names == ["run_snippet"]
    path, body, client_kwargs = _ApprovalClient.calls[-1]
    assert path == "/chat/approve"
    assert body == {
        "conversation_id": "conversation-1",
        "token": "secret",
        "decisions": {"run_snippet": False},
    }
    assert client_kwargs["base_url"] == "http://localhost:8000"


def test_inflight_tool_denial_uses_original_conversation_id(monkeypatch):
    pytest.importorskip("textual")
    import nexusos_cli.tui_app as tui_app

    stream_started = threading.Event()
    release_stream = threading.Event()
    denied_for = []

    class _StreamResponse:
        status_code = 200

        async def __aenter__(self):
            return self

        async def __aexit__(self, *args):
            return None

        async def aiter_lines(self):
            stream_started.set()
            await asyncio.to_thread(release_stream.wait, 2)
            yield "event: tool_request"
            yield 'data: {"token":"secret","actions":[{"name":"run_snippet"}]}'
            yield ""
            yield "event: done"
            yield "data: {}"

    class _StreamClient:
        def __init__(self, **kwargs):
            pass

        async def __aenter__(self):
            return self

        async def __aexit__(self, *args):
            return None

        def stream(self, *args, **kwargs):
            return _StreamResponse()

    def _capture_denial(*, conversation_id, **kwargs):
        denied_for.append(conversation_id)
        return ["run_snippet"]

    monkeypatch.setattr(tui_app.httpx, "AsyncClient", _StreamClient)
    monkeypatch.setattr(tui_app, "_deny_tool_request", _capture_denial)
    app = tui_app.NexusTUI.build_app(api_url="http://127.0.0.1:9")

    async def _run():
        async with app.run_test():
            app._start_chat("run it")
            assert await asyncio.to_thread(stream_started.wait, 2)
            original_id = app.conversation_id
            app._handle_slash("/new")
            assert app.conversation_id is None
            release_stream.set()
            for _ in range(200):
                if not app._busy:
                    break
                await asyncio.sleep(0.01)
            assert app._busy is False
            assert denied_for == [original_id]

    asyncio.run(_run())


def test_interrupt_cancels_silent_stream_and_accepts_next_message(monkeypatch):
    pytest.importorskip("textual")
    import nexusos_cli.tui_app as tui_app

    first_stream_started = threading.Event()

    class _StreamResponse:
        status_code = 200

        def __init__(self, call_number):
            self.call_number = call_number

        async def __aenter__(self):
            return self

        async def __aexit__(self, *args):
            return None

        async def aiter_lines(self):
            if self.call_number == 1:
                first_stream_started.set()
                await asyncio.Event().wait()
            yield "data: \"READY\""
            yield ""
            yield "event: done"
            yield "data: {}"

    class _StreamClient:
        calls = 0

        def __init__(self, **kwargs):
            pass

        async def __aenter__(self):
            return self

        async def __aexit__(self, *args):
            return None

        def stream(self, *args, **kwargs):
            type(self).calls += 1
            return _StreamResponse(type(self).calls)

    monkeypatch.setattr(tui_app.httpx, "AsyncClient", _StreamClient)
    app = tui_app.NexusTUI.build_app(api_url="http://127.0.0.1:9")

    async def _wait_until_idle():
        for _ in range(100):
            if not app._busy:
                return
            await asyncio.sleep(0.01)
        pytest.fail("stream did not become idle within one second")

    async def _run():
        async with app.run_test():
            app._start_chat("first")
            assert await asyncio.to_thread(first_stream_started.wait, 2)
            app.action_interrupt()
            await _wait_until_idle()

            log = app.query_one("#log")
            assert not any("ReadTimeout" in line.text for line in log.lines)

            app._start_chat("second")
            await _wait_until_idle()
            assert app.history[-1] == {
                "role": "assistant",
                "content": "READY",
            }

    asyncio.run(_run())


def test_escape_round_trip_helper():
    assert "[" in _escape("x[y]") or "\\[" in _escape("x[y]")