"""Smoke tests for the Navi Code TUI."""
from __future__ import annotations
from pathlib import Path
import pytest
from clients.terminal.tui.events import WsEvent
from clients.terminal.tui.tui_app import NaviCodeTui
@pytest.fixture(autouse=True)
def tmp_state_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
"""Override the state dir so tests never touch ~/.navi_code."""
from clients.terminal import config
original = config.settings.state_dir
config.settings.state_dir = tmp_path
import clients.terminal.tui.settings as settings_module
settings_module._tui_settings = None
yield tmp_path
config.settings.state_dir = original
settings_module._tui_settings = None
@pytest.fixture(autouse=True)
def mock_tui_api(monkeypatch: pytest.MonkeyPatch) -> None:
"""Mock the backend REST api so TUI tests never hit a live server.
Every test mounts NaviCodeTui, whose _startup worker calls api.create_session /
api.get_session. Without a server these raise and _ctx.session_id stays None,
which breaks any test that asserts on the session. Tests that assert on
stop_session still monkeypatch it per-test with their own call-tracking list.
"""
import clients.terminal.api as api_module
def fake_create_session(profile_id: str | None = None) -> dict:
return {"session_id": "test-session", "profile_id": profile_id or "navi_code"}
def fake_get_session(session_id: str) -> dict:
return {"session_id": session_id, "profile_id": "navi_code"}
def fake_list_sessions() -> list[dict]:
return []
def fake_get_profile_model(profile_id: str) -> str | None:
return "configured-model"
monkeypatch.setattr(api_module, "create_session", fake_create_session)
monkeypatch.setattr(api_module, "get_session", fake_get_session)
monkeypatch.setattr(api_module, "list_sessions", fake_list_sessions)
monkeypatch.setattr(api_module, "get_profile_model", fake_get_profile_model)
@pytest.mark.anyio
async def test_tui_mounts_widgets() -> None:
"""The TUI app mounts chat, status, and input widgets."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
assert pilot.app.query_one("ChatPanel") is not None
assert pilot.app.query_one("StatusPanel") is not None
assert pilot.app.query_one("InputBox") is not None
@pytest.mark.anyio
async def test_user_message_appears_in_chat() -> None:
"""Typing a message and submitting adds it to the chat panel."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
input_box = pilot.app.query_one("InputBox")
input_box._input.text = "hello"
await pilot.press("enter")
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
assert any(
item.kind == "user_message" and item.content == "hello" for item in chat._model.items
)
@pytest.mark.anyio
async def test_input_text_is_rendered_with_visible_color() -> None:
"""Typed text in the input field is rendered with a visible foreground color."""
async with NaviCodeTui(new_session=True).run_test(size=(80, 24)) as pilot:
await pilot.pause()
input_box = pilot.app.query_one("InputBox")
input_box._input.focus()
await pilot.press("h", "i", " ", "n", "a", "v", "i")
await pilot.pause()
assert input_box._input.text == "hi navi"
strip = input_box._input.render_line(0)
segments = list(strip)
rendered_text = "".join(seg.text for seg in segments)
assert "hi navi" in rendered_text
# Ensure foreground and background are not identical so text is visible.
for seg in segments:
if "hi navi" in seg.text and seg.style:
assert seg.style.color != seg.style.bgcolor
@pytest.mark.anyio
async def test_enter_submits_multiline_text() -> None:
"""Enter submits the full multi-line buffer (newlines preserved)."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
input_box = pilot.app.query_one("InputBox")
input_box._input.text = "line one\nline two"
input_box._input.focus()
await pilot.press("enter")
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
submitted = [
item.content
for item in chat._model.items
if item.kind == "user_message"
]
assert submitted == ["line one\nline two"]
# Input is cleared after submit.
assert input_box._input.text == ""
@pytest.mark.anyio
async def test_input_grows_with_multiline_content() -> None:
"""The input field height grows with content (multi-line), up to a cap."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
input_box = pilot.app.query_one("InputBox")
single = input_box._input.region.height
input_box._input.text = "a\nb\nc\nd\ne\nf"
await pilot.pause()
grown = input_box._input.region.height
assert grown > single
@pytest.mark.anyio
async def test_ws_event_renders_in_chat() -> None:
"""A synthetic WebSocket stream_delta is added to the assistant response."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.handle_ws_event({"type": "stream_start"})
chat.handle_ws_event({"type": "stream_delta", "delta": "hi"})
await pilot.pause()
assert chat._model._current_assistant is not None
assert chat._model._current_assistant.content == "hi"
@pytest.mark.anyio
async def test_stream_start_does_not_create_empty_assistant_item() -> None:
"""stream_start must not eagerly add an empty assistant bubble.
The answer bubble is created lazily on the first stream_delta so it lands at
the position where text arrives, not stranded at the top of the turn.
"""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.handle_ws_event({"type": "stream_start"})
await pilot.pause()
assert chat._model._current_assistant is None
assert not any(item.kind == "assistant_message" for item in chat._model.items)
@pytest.mark.anyio
async def test_final_answer_appears_after_tools() -> None:
"""The final assistant answer must render BELOW the tool cards, not above them.
Regression: stream_start used to create the assistant bubble at the top, so
the final answer (accumulated there) was scrolled out of view by scroll_end
while tools/thinking below stayed visible. Now each text segment opens its
own bubble at the point it arrives, so the final answer lands at the bottom.
"""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.handle_ws_event({"type": "stream_start"})
chat.handle_ws_event({"type": "thinking_delta", "delta": "hmm"})
chat.handle_ws_event({"type": "thinking_end"})
chat.handle_ws_event({"type": "stream_delta", "delta": "I'll list X. "})
chat.handle_ws_event({"type": "tool_started", "tool": "filesystem", "args": {}})
chat.handle_ws_event(
{"type": "tool_call", "tool": "filesystem", "args": {}, "result": "a\nb", "success": True}
)
chat.handle_ws_event({"type": "thinking_delta", "delta": "now answer"})
chat.handle_ws_event({"type": "thinking_end"})
chat.handle_ws_event({"type": "stream_delta", "delta": "X contains a and b."})
chat.handle_ws_event({"type": "stream_end", "content": "X contains a and b."})
await pilot.pause()
items = chat._model.items
# The final answer is the last assistant_message (a turn_meta duration
# line may follow it, which is expected and fine).
last_answer = max(i for i, it in enumerate(items) if it.kind == "assistant_message")
assert items[last_answer].content == "X contains a and b."
# It must come after the last tool_call.
last_tool = max(i for i, it in enumerate(items) if it.kind == "tool_call")
assert last_answer > last_tool
# No empty assistant bubbles.
assert all(it.content for it in items if it.kind == "assistant_message")
@pytest.mark.anyio
async def test_empty_assistant_item_purged_on_stream_end() -> None:
"""stream_end drops assistant/thinking bubbles that never received any text."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.handle_ws_event({"type": "stream_start"})
# An empty delta creates a bubble with no content.
chat.handle_ws_event({"type": "stream_delta", "delta": ""})
chat.handle_ws_event({"type": "stream_end", "content": ""})
await pilot.pause()
assert not any(
item.kind in ("assistant_message", "thinking_block") and not item.content
for item in chat._model.items
)
@pytest.mark.anyio
async def test_planning_status_rolls_in_place() -> None:
"""Successive non-subagent planning_status events update one line, not stack."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.handle_ws_event({"type": "planning_status", "phase": 1, "label": "Analysis", "is_subagent": False})
chat.handle_ws_event({"type": "planning_status", "phase": 2, "label": "Execution plan", "is_subagent": False})
chat.handle_ws_event({"type": "planning_status", "phase": 3, "label": "Plan review", "is_subagent": False})
await pilot.pause()
planning = [it for it in chat._model.items if it.kind == "planning_status"]
assert len(planning) == 1
assert planning[0].content == "Plan review"
assert planning[0].meta["phase"] == 3
@pytest.mark.anyio
async def test_plan_ready_consumes_planning_indicator_and_adds_card() -> None:
"""plan_ready removes the pending planning line and appends a plan card."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.handle_ws_event({"type": "planning_status", "phase": 1, "label": "Analysis", "is_subagent": False})
chat.handle_ws_event(
{"type": "plan_ready", "plan": "## Plan\n**Steps:**\n1. Do thing\n", "is_subagent": False}
)
await pilot.pause()
items = chat._model.items
assert not any(it.kind == "planning_status" for it in items)
plans = [it for it in items if it.kind == "plan_ready"]
assert len(plans) == 1
assert "## Plan" in plans[0].content
assert items[-1].kind == "plan_ready"
@pytest.mark.anyio
async def test_subagent_planning_appended_separately() -> None:
"""Subagent planning does not merge into the parent's rolling indicator."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.handle_ws_event({"type": "planning_status", "phase": 1, "label": "Analysis", "is_subagent": False})
chat.handle_ws_event({"type": "planning_status", "phase": 1, "label": "sub plan", "is_subagent": True})
await pilot.pause()
planning = [it for it in chat._model.items if it.kind == "planning_status"]
assert len(planning) == 2
assert planning[0].content == "Analysis"
assert planning[0].meta["is_subagent"] is False
assert planning[1].content == "sub plan"
assert planning[1].meta["is_subagent"] is True
@pytest.mark.anyio
async def test_turn_thinking_creates_subagent_thinking_block() -> None:
"""turn_thinking events become thinking_block items flagged is_subagent."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.handle_ws_event(
{"type": "turn_thinking", "thinking": "let me consider…", "is_subagent": True}
)
await pilot.pause()
blocks = [it for it in chat._model.items if it.kind == "thinking_block"]
assert len(blocks) == 1
assert blocks[0].content == "let me consider…"
assert blocks[0].meta["is_subagent"] is True
@pytest.mark.anyio
async def test_unknown_slash_command_shows_error() -> None:
"""Unknown slash command produces an error item in chat."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
input_box = pilot.app.query_one("InputBox")
input_box._input.text = "/notacommand"
await pilot.press("enter")
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
assert any(item.kind == "error" for item in chat._model.items)
@pytest.mark.anyio
async def test_chat_panel_clear_resets_conversation() -> None:
"""ChatPanel.clear() removes all items and current assistant state."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.handle_ws_event({"type": "stream_start"})
chat.handle_ws_event({"type": "stream_delta", "delta": "hi"})
await pilot.pause()
assert chat._model.items
chat.clear()
assert not chat._model.items
assert chat._model._current_assistant is None
@pytest.mark.anyio
async def test_status_panel_shows_backend_and_theme() -> None:
"""Status panel displays backend URL and current theme."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
status = pilot.app.query_one("StatusPanel")
backend_text = str(status._backend.render())
theme_text = str(status._theme.render())
assert "Backend:" in backend_text
assert "Theme: gnexus-dark" in theme_text
@pytest.mark.anyio
async def test_attach_shows_configured_model() -> None:
"""On attach the status panel shows the profile's configured model (pre-request)."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
status = pilot.app.query_one("StatusPanel")
model_text = str(status._model.render())
assert "configured-model" in model_text
@pytest.mark.anyio
async def test_model_info_updates_status_panel() -> None:
"""A model_info WS event updates the status panel to the resolved model."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
app = pilot.app
app.on_ws_event(WsEvent({"type": "model_info", "model": "resolved-by-backend"}))
await pilot.pause()
status = app.query_one("StatusPanel")
model_text = str(status._model.render())
assert "resolved-by-backend" in model_text
@pytest.mark.anyio
async def test_model_info_not_forwarded_to_chat() -> None:
"""model_info is a status update, not a chat item."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
app = pilot.app
app.on_ws_event(WsEvent({"type": "model_info", "model": "some-model"}))
await pilot.pause()
chat = app.query_one("ChatPanel")
assert not any(
getattr(item, "kind", None) == "model_info" for item in chat._model.items
)
@pytest.mark.anyio
async def test_todo_updated_event_renders_in_status_panel() -> None:
"""A todo_updated WS event updates the TodoList in the side panel."""
from clients.terminal.tui.widgets.todo_list import TodoList
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
app = pilot.app
tasks = [
{"index": 1, "text": "first step", "status": "done", "validation": "ok"},
{"index": 2, "text": "second step", "status": "in_progress", "validation": ""},
]
app.on_ws_event(WsEvent({"type": "todo_updated", "session_id": "x", "tasks": tasks}))
await pilot.pause()
todo = app.query_one(TodoList)
from rich.text import Text as _Text
plain = _Text.from_markup(todo._build_content()).plain
assert "1/2 done" in plain
assert "first step" in plain
assert "second step" in plain
@pytest.mark.anyio
async def test_todo_updated_not_forwarded_to_chat() -> None:
"""todo_updated is a side-panel update, not a chat item."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
app = pilot.app
app.on_ws_event(
WsEvent({"type": "todo_updated", "session_id": "x", "tasks": [{"index": 1, "text": "t", "status": "pending", "validation": ""}]})
)
await pilot.pause()
chat = app.query_one("ChatPanel")
assert not any(
getattr(item, "kind", None) == "todo_updated" for item in chat._model.items
)
@pytest.mark.anyio
async def test_attach_session_seeds_todo_from_api(monkeypatch) -> None:
"""attach_session fetches the existing todo via api.get_todos so a resumed
session shows its plan before the first todo_updated WS event."""
from clients.terminal.tui.widgets.todo_list import TodoList
fetched: list[str] = []
def fake_get_todos(session_id: str) -> dict:
fetched.append(session_id)
return {
"session_id": session_id,
"tasks": [{"index": 1, "text": "seeded step", "status": "pending", "validation": ""}],
}
import clients.terminal.api as api_module
monkeypatch.setattr(api_module, "get_todos", fake_get_todos)
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
# The startup worker attached a session, which should have fetched todos.
assert fetched, "attach_session did not call api.get_todos"
todo = pilot.app.query_one(TodoList)
assert "seeded step" in todo._build_content()
@pytest.mark.anyio
async def test_streaming_state_tracks_ws_events() -> None:
"""stream_start enters streaming mode; stream_end/stream_stopped/error leave it."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
app = pilot.app
input_box = app.query_one("InputBox")
app.on_ws_event(WsEvent({"type": "stream_start"}))
await pilot.pause()
assert app._streaming is True
assert "Esc to stop" in input_box._input.placeholder
app.on_ws_event(WsEvent({"type": "stream_delta", "delta": "hi"}))
await pilot.pause()
assert app._streaming is True
app.on_ws_event(WsEvent({"type": "stream_stopped"}))
await pilot.pause()
assert app._streaming is False
assert input_box._input.placeholder == "Ask anything..."
@pytest.mark.anyio
async def test_escape_stops_active_stream(monkeypatch: pytest.MonkeyPatch) -> None:
"""Pressing Esc while streaming calls api.stop_session for the current session."""
stopped: list[str] = []
async def fake_stop_session(session_id: str) -> dict:
stopped.append(session_id)
return {"ok": True}
import clients.terminal.tui.tui_app as tui_app_module
monkeypatch.setattr(tui_app_module.api, "stop_session", fake_stop_session)
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
app = pilot.app
# Simulate an active run for the resolved session.
app.on_ws_event(WsEvent({"type": "stream_start"}))
await pilot.pause()
assert app._streaming is True
current_session = app._ctx.session_id
assert current_session
await pilot.press("escape")
await pilot.pause(0.1)
assert stopped == [current_session]
@pytest.mark.anyio
async def test_stream_stopped_renders_status() -> None:
"""stream_stopped resets the assistant buffer and shows a status message."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.handle_ws_event({"type": "stream_start"})
chat.handle_ws_event({"type": "stream_delta", "delta": "partial"})
chat.handle_ws_event({"type": "stream_stopped"})
await pilot.pause()
assert chat._model._current_assistant is None
assert any(
item.kind == "status" and "stopped" in item.content.lower()
for item in chat._model.items
)
@pytest.mark.anyio
async def test_escape_does_nothing_when_not_streaming(monkeypatch: pytest.MonkeyPatch) -> None:
"""Esc without an active stream does not call stop_session."""
calls: list[str] = []
async def fake_stop_session(session_id: str) -> dict:
calls.append(session_id)
return {"ok": True}
import clients.terminal.tui.tui_app as tui_app_module
monkeypatch.setattr(tui_app_module.api, "stop_session", fake_stop_session)
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
await pilot.press("escape")
await pilot.pause(0.1)
assert calls == []
@pytest.mark.anyio
async def test_resolve_session_with_explicit_id_resumes_it(monkeypatch: pytest.MonkeyPatch) -> None:
"""An explicit session_id (--resume) resumes that exact session."""
import clients.terminal.api as api_module
seen: list[str] = []
monkeypatch.setattr(api_module, "get_session", lambda sid: seen.append(sid) or {
"session_id": sid,
"profile_id": "navi_code",
})
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
# _startup already attached a session (calling get_session once), so
# snapshot the recorder baseline before our explicit resume call.
baseline = len(seen)
result = await pilot.app._resolve_session("sess-explicit", None, False)
assert result == "sess-explicit"
assert seen[baseline:] == ["sess-explicit"]
@pytest.mark.anyio
async def test_resolve_session_strict_on_bad_explicit_id(monkeypatch: pytest.MonkeyPatch) -> None:
"""A bad --resume id surfaces an error and does NOT silently create a new session."""
import clients.terminal.api as api_module
def boom(_sid: str) -> dict:
raise RuntimeError("session not found")
created: list[str] = []
monkeypatch.setattr(api_module, "get_session", boom)
monkeypatch.setattr(api_module, "create_session", lambda *_a, **_k: created.append("x") or {
"session_id": "NEW",
"profile_id": "navi_code",
})
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
# _startup (force_new=True) already created a session; snapshot so we
# only assert about what our bad-resume call does.
created_before = len(created)
chat = pilot.app.query_one("ChatPanel")
before_errors = sum(1 for it in chat._model.items if it.kind == "error")
result = await pilot.app._resolve_session("sess-bad", None, False)
assert result is None
# No new session was created behind the user's back.
assert len(created) == created_before
# An error was surfaced to the chat panel.
after_errors = sum(1 for it in chat._model.items if it.kind == "error")
assert after_errors == before_errors + 1
@pytest.mark.anyio
async def test_activity_indicator_runs_during_turn() -> None:
"""stream_start starts the spinner; stream_end stops it."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
status_bar = pilot.app.query_one("StatusBar")
activity = status_bar._activity
assert activity._active is False
pilot.app.on_ws_event(WsEvent({"type": "stream_start"}))
await pilot.pause()
assert activity._active is True
assert activity._label == "thinking"
pilot.app.on_ws_event(WsEvent({"type": "stream_end", "content": ""}))
await pilot.pause()
assert activity._active is False
assert activity._label == ""
@pytest.mark.anyio
async def test_activity_indicator_phase_label_reflects_events() -> None:
"""The spinner label tracks the current stage as WS events arrive."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
status_bar = pilot.app.query_one("StatusBar")
activity = status_bar._activity
pilot.app.on_ws_event(WsEvent({"type": "stream_start"}))
await pilot.pause()
assert activity._label == "thinking"
pilot.app.on_ws_event(WsEvent({"type": "planning_status", "phase": 1, "label": "Analysis", "is_subagent": False}))
await pilot.pause()
assert activity._label == "planning"
pilot.app.on_ws_event(WsEvent({"type": "stream_delta", "delta": "hi"}))
await pilot.pause()
assert activity._label == "responding"
pilot.app.on_ws_event(WsEvent({"type": "tool_started", "tool": "filesystem", "args": {}}))
await pilot.pause()
assert activity._label == "running filesystem"
@pytest.mark.anyio
async def test_activity_indicator_stops_on_disconnect() -> None:
"""A dropped connection stops the spinner (no stream_end will arrive)."""
from clients.terminal.tui.events import ConnectionStatusChanged
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
status_bar = pilot.app.query_one("StatusBar")
activity = status_bar._activity
pilot.app.on_ws_event(WsEvent({"type": "stream_start"}))
await pilot.pause()
assert activity._active is True
pilot.app.post_message(ConnectionStatusChanged(connected=False, detail=""))
await pilot.pause()
assert activity._active is False
@pytest.mark.anyio
async def test_attach_session_loads_history(monkeypatch: pytest.MonkeyPatch) -> None:
"""Resuming/attaching a session replays its stored messages into the chat."""
import clients.terminal.api as api_module
import clients.terminal.tui.tui_app as tui_app_module
class FakeBridge:
def __init__(self, app, session_id, cwd=None) -> None:
self.client = None
async def start(self) -> None:
pass
async def stop(self) -> None:
pass
monkeypatch.setattr(tui_app_module, "WsBridge", FakeBridge)
messages = [
{"role": "user", "content": "earlier question", "is_display": True},
{"role": "assistant", "content": "earlier answer", "is_display": True},
{"role": "user", "content": "context-only", "is_display": False},
]
monkeypatch.setattr(api_module, "get_session", lambda sid: {
"session_id": sid,
"profile_id": "navi_code",
"messages": messages,
})
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
await pilot.app.attach_session("sess-resume")
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
contents = [it.content for it in chat._model.items]
assert "earlier question" in contents
assert "earlier answer" in contents
# The context-only message (is_display=False) was skipped.
assert "context-only" not in contents
@pytest.mark.anyio
async def test_stream_end_updates_context_fill() -> None:
"""A stream_end event carrying context_tokens/max_context_tokens seeds the gauge."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
ctx = pilot.app.query_one("StatusBar")._ctx_fill
# No data yet → placeholder.
assert ctx._used is None
pilot.app.on_ws_event(WsEvent({
"type": "stream_end", "content": "",
"context_tokens": 12000,
"max_context_tokens": 32000,
}))
await pilot.pause()
assert ctx._used == 12000
assert ctx._max == 32000
# 12000/32000 = 37.5%, which rounds to 38% (banker's rounding to even).
assert "38%" in str(ctx.render())
@pytest.mark.anyio
async def test_context_compressed_updates_fill() -> None:
"""context_compressed reports the post-compression token count."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
ctx = pilot.app.query_one("StatusBar")._ctx_fill
pilot.app.on_ws_event(WsEvent({
"type": "stream_end", "content": "",
"context_tokens": 30000,
"max_context_tokens": 32000,
}))
await pilot.pause()
assert ctx._used == 30000
pilot.app.on_ws_event(WsEvent({
"type": "context_compressed",
"context_tokens": 8000,
"max_context_tokens": 32000,
}))
await pilot.pause()
assert ctx._used == 8000
@pytest.mark.anyio
async def test_stream_stopped_does_not_touch_context() -> None:
"""stream_stopped carries no context fields; the gauge keeps its last value."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
ctx = pilot.app.query_one("StatusBar")._ctx_fill
pilot.app.on_ws_event(WsEvent({
"type": "stream_end", "content": "",
"context_tokens": 5000,
"max_context_tokens": 32000,
}))
await pilot.pause()
assert ctx._used == 5000
pilot.app.on_ws_event(WsEvent({"type": "stream_stopped"}))
await pilot.pause()
# Unchanged — stream_stopped has no token fields.
assert ctx._used == 5000
@pytest.mark.anyio
async def test_context_fill_keeps_last_on_none() -> None:
"""A context report missing token fields keeps the last known value."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
ctx = pilot.app.query_one("StatusBar")._ctx_fill
pilot.app.on_ws_event(WsEvent({
"type": "stream_end", "content": "",
"context_tokens": 10000,
"max_context_tokens": 32000,
}))
await pilot.pause()
pilot.app.on_ws_event(WsEvent({
"type": "stream_end", "content": "",
"context_tokens": None,
"max_context_tokens": None,
}))
await pilot.pause()
assert ctx._used == 10000
assert ctx._max == 32000
@pytest.mark.anyio
async def test_context_fill_color_thresholds() -> None:
"""The gauge color escalates as the window fills (dim → warning → error)."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
ctx = pilot.app.query_one("StatusBar")._ctx_fill
from clients.terminal.tui.themes import get_active_theme
t = get_active_theme()
# Under 70% → dim.
assert ctx._color_for(50) == t.text_dim.hex
# 70–89% → warning.
assert ctx._color_for(75) == t.warning.hex
# 90%+ → error.
assert ctx._color_for(95) == t.error.hex
@pytest.mark.anyio
async def test_context_fill_k_notation() -> None:
"""Token counts render in compact k/M notation."""
from clients.terminal.tui.widgets.status_bar import _humanize
assert _humanize(500) == "500"
assert _humanize(12000) == "12.0k"
assert _humanize(1_500_000) == "1.50M"
@pytest.mark.anyio
async def test_attach_session_seeds_context_fill(monkeypatch: pytest.MonkeyPatch) -> None:
"""On resume the gauge is seeded from get_session so it isn't blank pre-turn."""
import clients.terminal.api as api_module
import clients.terminal.tui.tui_app as tui_app_module
class FakeBridge:
def __init__(self, app, session_id, cwd=None) -> None:
self.client = None
async def start(self) -> None:
pass
async def stop(self) -> None:
pass
monkeypatch.setattr(tui_app_module, "WsBridge", FakeBridge)
monkeypatch.setattr(api_module, "get_session", lambda sid: {
"session_id": sid,
"profile_id": "navi_code",
"messages": [],
"context_token_count": 9000,
"max_context_tokens": 32000,
})
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
ctx = pilot.app.query_one("StatusBar")._ctx_fill
# _startup already ran attach_session (against the monkeypatched
# get_session), so the gauge is seeded — not blank — before any turn.
assert ctx._used == 9000
assert ctx._max == 32000
assert "28%" in str(ctx.render())
# An explicit re-attach re-seeds from the same session response.
await pilot.app.attach_session("sess-resume")
await pilot.pause()
assert ctx._used == 9000
assert "28%" in str(ctx.render())
# ─── elapsed timer + turn metadata ──────────────────────────────────────────
@pytest.mark.anyio
async def test_elapsed_timer_starts_on_stream_start() -> None:
"""stream_start starts the elapsed timer (it is active and showing 0s)."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
elapsed = pilot.app.query_one("StatusBar")._elapsed
assert elapsed._active is False
pilot.app.on_ws_event(WsEvent({"type": "stream_start"}))
await pilot.pause()
assert elapsed._active is True
assert elapsed._start_ts is not None
assert "0s" in str(elapsed.render())
@pytest.mark.anyio
async def test_elapsed_timer_stops_on_stream_end_synced_to_backend() -> None:
"""stream_end freezes the timer at the backend's authoritative elapsed."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
elapsed = pilot.app.query_one("StatusBar")._elapsed
pilot.app.on_ws_event(WsEvent({"type": "stream_start"}))
await pilot.pause()
pilot.app.on_ws_event(WsEvent({"type": "stream_end", "content": "", "elapsed_seconds": 136}))
await pilot.pause()
assert elapsed._active is False
assert "2m 16s" in str(elapsed.render())
@pytest.mark.anyio
async def test_help_command_has_no_literal_markup() -> None:
"""/help content must not contain raw ``[b]`` markup: StatusRenderer builds a
literal ``Text(...)`` which does not parse markup, so the tags would show as
literal square brackets. Headers are plain text."""
from clients.terminal.tui.commands.builtin import HelpCommand
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
await HelpCommand().execute(pilot.app._ctx, "")
await pilot.pause()
status = [it for it in chat._model.items if it.kind == "status"]
assert status
content = status[-1].content
assert "[b]" not in content
assert "[/b]" not in content
assert "Slash commands" in content
@pytest.mark.anyio
async def test_submit_while_reconnecting_enqueues_not_errors() -> None:
"""A message submitted while the bridge is reconnecting (connected=False) is
enqueued for later delivery, not dropped with a 'not connected' error — the
WsBridge input loop buffers until the socket is live again."""
class _FakeClient:
def __init__(self) -> None:
self.enqueued: list[str] = []
def enqueue(self, msg) -> None:
self.enqueued.append(msg)
class _FakeBridge:
connected = False
def __init__(self) -> None:
self.client = _FakeClient()
async def stop(self) -> None: # called from on_unmount on test teardown
return None
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
bridge = _FakeBridge()
pilot.app._bridge = bridge
input_box = pilot.app.query_one("InputBox")
input_box._input.text = "hello during reconnect"
await pilot.press("enter")
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
assert bridge.client.enqueued, "message was enqueued for delivery"
assert "hello during reconnect" in bridge.client.enqueued[0]
assert not any(it.kind == "error" for it in chat._model.items)
@pytest.mark.anyio
async def test_elapsed_timer_freezes_on_stream_stopped_without_backend() -> None:
"""stream_stopped has no elapsed_seconds; the timer freezes at its local value."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
elapsed = pilot.app.query_one("StatusBar")._elapsed
pilot.app.on_ws_event(WsEvent({"type": "stream_start"}))
await pilot.pause()
assert elapsed._active is True
pilot.app.on_ws_event(WsEvent({"type": "stream_stopped"}))
await pilot.pause()
assert elapsed._active is False
# Frozen at some duration (locally measured, near zero) — still a value.
assert str(elapsed.render()) != ""
@pytest.mark.anyio
async def test_stream_end_writes_turn_meta_into_chat() -> None:
"""stream_end appends a dim ⏱ duration line to the chat for the turn."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
pilot.app.on_ws_event(WsEvent({"type": "stream_start"}))
pilot.app.on_ws_event(WsEvent({"type": "stream_delta", "delta": "done"}))
pilot.app.on_ws_event(WsEvent({"type": "stream_end", "content": "done", "elapsed_seconds": 136}))
await pilot.pause()
metas = [it for it in chat._model.items if it.kind == "turn_meta"]
assert len(metas) == 1
assert metas[0].meta["elapsed_seconds"] == 136
# The metadata renders below the assistant answer.
assert chat._model.items[-1].kind == "turn_meta"
@pytest.mark.anyio
async def test_stream_stopped_does_not_write_turn_meta_into_chat() -> None:
"""An interrupted turn does not get a duration metadata line."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
pilot.app.on_ws_event(WsEvent({"type": "stream_start"}))
pilot.app.on_ws_event(WsEvent({"type": "stream_delta", "delta": "partial"}))
pilot.app.on_ws_event(WsEvent({"type": "stream_stopped"}))
await pilot.pause()
assert not any(it.kind == "turn_meta" for it in chat._model.items)
@pytest.mark.anyio
async def test_disconnect_stops_elapsed_timer() -> None:
"""A dropped connection stops the elapsed timer (no stream_end will arrive)."""
from clients.terminal.tui.events import ConnectionStatusChanged
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
elapsed = pilot.app.query_one("StatusBar")._elapsed
pilot.app.on_ws_event(WsEvent({"type": "stream_start"}))
await pilot.pause()
assert elapsed._active is True
pilot.app.post_message(ConnectionStatusChanged(connected=False, detail=""))
await pilot.pause()
assert elapsed._active is False
@pytest.mark.anyio
async def test_keyboard_scroll_bindings_move_chat_panel() -> None:
"""Ctrl+End jumps to the bottom and Ctrl+Shift+Up/Down scroll the chat panel
one line, while focus stays on the input box (the keys are not claimed by
TextArea, so they bubble to the app bindings)."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.clear()
for i in range(30):
chat.add_user_message(f"line {i} " + "x" * 40)
chat.handle_ws_event({"type": "stream_start"})
chat.handle_ws_event({"type": "stream_delta", "delta": "y" * 40})
chat.handle_ws_event({"type": "stream_end"})
await pilot.pause()
assert chat.max_scroll_y > 0
# Start at the top.
chat.scroll_to(y=0, animate=False)
await pilot.pause()
assert chat.scroll_y == 0
# Ctrl+End jumps to the bottom.
await pilot.press("ctrl+end")
await pilot.pause()
assert chat.scroll_y == chat.max_scroll_y
# Ctrl+Shift+Up scrolls one line up from the bottom.
bottom = chat.scroll_y
await pilot.press("ctrl+shift+up")
await pilot.pause()
assert chat.scroll_y < bottom
# Ctrl+Shift+Down scrolls back down one line.
up_y = chat.scroll_y
await pilot.press("ctrl+shift+down")
await pilot.pause()
assert chat.scroll_y > up_y
@pytest.mark.anyio
async def test_reading_mode_hides_chrome_around_chat() -> None:
"""Ctrl+R toggles reading mode: the right column, input prompt, and status
bar are hidden and the chat panel loses its outer border, so only the
message column remains — letting Shift+drag select just the conversation
for Ctrl+Shift+C (raw terminal copy) in terminals without OSC 52."""
async with NaviCodeTui(new_session=True).run_test(size=(80, 24)) as pilot:
await pilot.pause()
app = pilot.app
right = app.query_one("#tui-right-column")
input_box = app.query_one("InputBox")
status_bar = app.query_one("StatusBar")
chat = app.query_one("ChatPanel")
# Initially the full UI is shown.
assert not app.has_class("reading-mode")
assert right.styles.display != "none"
assert input_box.styles.display != "none"
assert status_bar.styles.display != "none"
# Ctrl+R enters reading mode.
await pilot.press("ctrl+r")
await pilot.pause()
assert app.has_class("reading-mode")
assert right.styles.display == "none"
assert input_box.styles.display == "none"
assert status_bar.styles.display == "none"
# The chat panel's outer border is dropped so it isn't part of the copy.
# ``border: none`` resolves to an empty border type ("") on each edge.
assert chat.styles.border_top[0] == ""
# Ctrl+R again restores the full UI.
await pilot.press("ctrl+r")
await pilot.pause()
assert not app.has_class("reading-mode")
assert right.styles.display != "none"
assert input_box.styles.display != "none"
assert status_bar.styles.display != "none"
