NexusOS/tests/test_tui.py

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

import asyncio

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_escape_round_trip_helper():
    assert "[" in _escape("x[y]") or "\\[" in _escape("x[y]")