vh
942e33898c
Volva code-review surfaced 8 findings against the TDD-passing
TUI shell. All 8 addressed.
Drift fixes (code):
- Primary: INV-002 + INV-003 require visible Footer-area rendering
of session-identity + Ctrl-C state hint. Implementation stored
the strings in `self.sub_title` (which lands in the Header, not
Footer) and `self.hint` (a plain attribute, never rendered). Fixed
by adding two `Static` widgets (id="identity" and id="hint") in
compose; the `_set_hint()` helper mirrors state into the widget on
every state transition. Same-model TDD missed this because tests
asserted internal state, not visible widget content.
- Reverted `_stream_turn_worker(content, log)` to single-param
`(content)` per the contract FN signature. The widened signature
was a TDD-time workaround for a NoMatches-during-worker
execution; root cause was test timing (added `await pilot.pause()`
before the polling loop in `_submit_and_wait`).
- Restored `exclusive=True` on `self.run_worker(...)` per the
contract STEP 6 spec.
- Added missing `isinstance(args, ParsedArgs)` PRE assertion to
`run_tui`. Required hoisting `from ratatoskr.cli import
ParsedArgs` out of TYPE_CHECKING — runtime import is fine (no
circular dependency: cli lazy-imports tui inside main; tui
imports cli unconditionally at module load).
- Added missing union-type PRE assertion to `_render_event_to_log`.
Contract amendments (precision):
- COMPOSE shape: RichLog `markup=False, highlight=False` (was True,
True). Explanatory comment in-line: bracketed labels like
[cancel_failed] would otherwise be interpreted+stripped as Rich
style spans; the post-Done Markdown rendering still works via
Markdown() Renderable.
- INV-002 reworded: identity rendered via dedicated
Static(id="identity") widget composed adjacent to Footer (Textual's
built-in Footer renders BINDINGS descriptions; a sibling Static
carries custom content in the same visual region).
- on_mount POST-003 amended to allow `agent_id is None` when
--session is used without --agent (matches INV-002 carve-out;
GET /sessions/{id} agent lookup is out of scope for this shell).
- run_tui happy_returns_zero_on_quit test description clarified:
App.run() is sync and can't be driven by Pilot, so run_tui's
wrapping behavior is tested via monkeypatch; the piloted Ctrl-D
exit path is covered separately by TestActionQuit.
Test fixes:
- footer_identity_visible_first_frame, footer_hint_flips_to_cancel,
streaming_first_ctrl_c_cancels: now query the Static(#identity) /
Static(#hint) widgets via `widget.render()` instead of asserting
on `app.sub_title` / `app.hint` internal state. The internal
state still exists (mirror), but the load-bearing assertion is
on visible widget content.
Meta-note from Volva: "TDD pass caught most stream/session/error
mechanics, but tested internal state where the contract required
visible Footer behavior, so same-model TDD would plausibly miss the
primary drift." Calibration shape continues across all four issues:
the post-TDD cross-model review consistently catches assert-boundary
+ observability-shape gaps the test-author's hypotheses don't cover
(#1: 4 findings, #2: 3, #3: 5, #4: 8).
164/164 tests GREEN; ruff clean; both contract drift checks clean.
998 lines
36 KiB
Python
998 lines
36 KiB
Python
"""Tests for ratatoskr.tui per docs/contracts/issues/4.contract.md."""
|
||
|
||
from unittest.mock import MagicMock
|
||
|
||
import httpx
|
||
import pytest
|
||
import respx
|
||
from textual.widgets import RichLog
|
||
|
||
from ratatoskr.cli import ParsedArgs
|
||
from ratatoskr.sse_client import (
|
||
Cancelled,
|
||
Done,
|
||
Error,
|
||
SseId,
|
||
Text,
|
||
TextBoundary,
|
||
Thinking,
|
||
ToolResult,
|
||
ToolStart,
|
||
WorkerPhase,
|
||
)
|
||
from ratatoskr.tui import RatatoskrApp, _cancel_via_sse, _render_event_to_log
|
||
|
||
_CANCEL_OK_RESP = {"turn_id": 42, "cancelled": True, "reason": None, "partial_message_id": None}
|
||
_CREATE_OK_RESP = {
|
||
"session_id": "s-new12345",
|
||
"agent_id": "mimir",
|
||
"message_count": 0,
|
||
"created_at": "2026-05-21T00:00:00+00:00",
|
||
"last_active": "2026-05-21T00:00:00+00:00",
|
||
"metadata": {},
|
||
}
|
||
|
||
|
||
def _args_new(**overrides) -> ParsedArgs:
|
||
base = dict(
|
||
send_content=None,
|
||
session_id=None,
|
||
new=True,
|
||
agent_id="mimir",
|
||
api_key="k",
|
||
server_url="https://w.example",
|
||
raw=False,
|
||
)
|
||
base.update(overrides)
|
||
return ParsedArgs(**base)
|
||
|
||
|
||
def _args_existing(session_id: str = "s-1existing", **overrides) -> ParsedArgs:
|
||
base = dict(
|
||
send_content=None,
|
||
session_id=session_id,
|
||
new=False,
|
||
agent_id=None,
|
||
api_key="k",
|
||
server_url="https://w.example",
|
||
raw=False,
|
||
)
|
||
base.update(overrides)
|
||
return ParsedArgs(**base)
|
||
|
||
|
||
def _spy_writes(monkeypatch) -> list:
|
||
"""Patch RichLog.write to record every arg into a list (returned)."""
|
||
writes: list = []
|
||
original = RichLog.write
|
||
|
||
def spy(self, content, **kw):
|
||
writes.append(content)
|
||
return original(self, content, **kw)
|
||
|
||
monkeypatch.setattr(RichLog, "write", spy)
|
||
return writes
|
||
|
||
|
||
SID = SseId(42, 5)
|
||
|
||
|
||
class TestRenderEventToLog:
|
||
def test_text_renders_raw_delta(self) -> None:
|
||
"""text_renders_raw_delta [happy,tracer]: Text → log.write('hello')."""
|
||
log = MagicMock()
|
||
_render_event_to_log(Text(sse_id=SID, content="hello"), log=log, raw=False)
|
||
log.write.assert_called_once_with("hello")
|
||
|
||
def test_done_renders_label_only(self) -> None:
|
||
"""done_renders_label_only: …"""
|
||
log = MagicMock()
|
||
evt = Done(
|
||
sse_id=SID,
|
||
phase="completed",
|
||
response="hi there",
|
||
model="glm5-turbo",
|
||
duration_ms=1234,
|
||
usage={"prompt": 1, "completion": 2},
|
||
)
|
||
_render_event_to_log(evt, log=log, raw=False)
|
||
log.write.assert_called_once()
|
||
line = log.write.call_args[0][0]
|
||
assert line.startswith("[done]")
|
||
assert "turn_id=42" in line
|
||
assert "model=glm5-turbo" in line
|
||
# POST-003: the Done line is labels only; markdown render is the caller's job
|
||
assert "hi there" not in line
|
||
|
||
def test_error_renders_label(self) -> None:
|
||
"""error_renders_label: Error → log line starts with [error]."""
|
||
log = MagicMock()
|
||
evt = Error(sse_id=SID, phase="failed", message="boom", error_code="llm_output_invalid")
|
||
_render_event_to_log(evt, log=log, raw=False)
|
||
line = log.write.call_args[0][0]
|
||
assert line.startswith("[error]")
|
||
assert "turn_id=42" in line
|
||
assert "code=llm_output_invalid" in line
|
||
|
||
def test_cancelled_renders_label(self) -> None:
|
||
"""cancelled_renders_label: Cancelled → log line starts with [cancelled]."""
|
||
log = MagicMock()
|
||
evt = Cancelled(
|
||
sse_id=SID, phase="cancelled", turn_id=42, reason="user", partial_message_id=7
|
||
)
|
||
_render_event_to_log(evt, log=log, raw=False)
|
||
line = log.write.call_args[0][0]
|
||
assert line.startswith("[cancelled]")
|
||
assert "reason='user'" in line
|
||
assert "partial_message_id=7" in line
|
||
|
||
def test_worker_phase_renders_label(self) -> None:
|
||
"""worker_phase_renders_label: WorkerPhase → log line starts with [worker_phase]."""
|
||
log = MagicMock()
|
||
evt = WorkerPhase(sse_id=SID, phase="streaming", turn_id=42)
|
||
_render_event_to_log(evt, log=log, raw=False)
|
||
line = log.write.call_args[0][0]
|
||
assert line.startswith("[worker_phase]")
|
||
assert "phase=streaming" in line
|
||
|
||
def test_thinking_truncated(self) -> None:
|
||
"""thinking_truncated [trace]: …"""
|
||
log = MagicMock()
|
||
_render_event_to_log(Thinking(sse_id=SID, content="a" * 500), log=log, raw=False)
|
||
line = log.write.call_args[0][0]
|
||
assert line.startswith("[thinking]")
|
||
assert "a" * 500 not in line
|
||
assert "a" * 200 in line
|
||
|
||
def test_tool_start_renders_label(self) -> None:
|
||
"""tool_start_renders_label: ToolStart → [tool_start] name=... args=..."""
|
||
log = MagicMock()
|
||
evt = ToolStart(sse_id=SID, name="read_file", arguments={"path": "/x"})
|
||
_render_event_to_log(evt, log=log, raw=False)
|
||
line = log.write.call_args[0][0]
|
||
assert line.startswith("[tool_start] name=read_file args=")
|
||
|
||
def test_tool_result_truncated(self) -> None:
|
||
"""tool_result_truncated [trace]: …"""
|
||
log = MagicMock()
|
||
evt = ToolResult(sse_id=SID, name="x", result="b" * 500, duration_ms=42)
|
||
_render_event_to_log(evt, log=log, raw=False)
|
||
line = log.write.call_args[0][0]
|
||
assert line.startswith("[tool_result]")
|
||
# The whole repr-portion of the result is truncated to 200; the full 500-b
|
||
# string can never fit in line whole.
|
||
assert "b" * 500 not in line
|
||
|
||
def test_text_boundary_renders_label(self) -> None:
|
||
"""text_boundary_renders_label: TextBoundary → [text_boundary] kind=... char_offset=..."""
|
||
log = MagicMock()
|
||
evt = TextBoundary(sse_id=SID, kind="sentence", char_offset=128, ts="2026-05-21T00:00:00Z")
|
||
_render_event_to_log(evt, log=log, raw=False)
|
||
line = log.write.call_args[0][0]
|
||
assert line.startswith("[text_boundary]")
|
||
assert "kind=sentence" in line
|
||
assert "char_offset=128" in line
|
||
|
||
|
||
class TestCancelViaSse:
|
||
@respx.mock
|
||
async def test_happy_cancel(self) -> None:
|
||
"""happy_cancel [happy,tracer]: 200 OK → returns None; log has no [cancel_failed]."""
|
||
respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
|
||
return_value=httpx.Response(200, json=_CANCEL_OK_RESP)
|
||
)
|
||
log = MagicMock()
|
||
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
||
result = await _cancel_via_sse(client, "s-1", 42, log=log)
|
||
assert result is None
|
||
log.write.assert_not_called()
|
||
|
||
@respx.mock
|
||
async def test_cancel_failed_500(self) -> None:
|
||
"""cancel_failed_500 [error]: …"""
|
||
respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
|
||
return_value=httpx.Response(500, content=b"boom")
|
||
)
|
||
log = MagicMock()
|
||
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
||
await _cancel_via_sse(client, "s-1", 42, log=log)
|
||
line = log.write.call_args[0][0]
|
||
assert "[cancel_failed]" in line
|
||
assert "CancelFailed" in line
|
||
|
||
@respx.mock
|
||
async def test_cancel_already_completed(self) -> None:
|
||
"""cancel_already_completed [scenario]: 409 → '[cancel_failed] CancelAlreadyCompleted:'."""
|
||
respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
|
||
return_value=httpx.Response(409)
|
||
)
|
||
log = MagicMock()
|
||
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
||
await _cancel_via_sse(client, "s-1", 42, log=log)
|
||
line = log.write.call_args[0][0]
|
||
assert "[cancel_failed]" in line
|
||
assert "CancelAlreadyCompleted" in line
|
||
|
||
@respx.mock
|
||
async def test_transport_error_swallowed(self) -> None:
|
||
"""transport_error_swallowed [error]: …"""
|
||
respx.post("https://w.example/sessions/s-1/turns/42/cancel").mock(
|
||
side_effect=httpx.ConnectError("network down")
|
||
)
|
||
log = MagicMock()
|
||
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
||
await _cancel_via_sse(client, "s-1", 42, log=log)
|
||
line = log.write.call_args[0][0]
|
||
assert "[cancel_failed]" in line
|
||
assert "ConnectError" in line
|
||
|
||
|
||
class TestAppMount:
|
||
@respx.mock
|
||
async def test_happy_new_session_mount(self) -> None:
|
||
"""happy_new_session_mount [happy,tracer]: …"""
|
||
respx.post("https://w.example/sessions").mock(
|
||
return_value=httpx.Response(201, json=_CREATE_OK_RESP)
|
||
)
|
||
app = RatatoskrApp(_args_new())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
assert app.session_id == "s-new12345"
|
||
assert app.agent_id == "mimir"
|
||
assert app.state == "idle"
|
||
# INV-002: session identity visible — agent_id + last 8 of session_id
|
||
assert "mimir" in (app.sub_title or "")
|
||
assert app.session_id[-8:] in (app.sub_title or "")
|
||
|
||
@respx.mock
|
||
async def test_happy_existing_session_mount(self) -> None:
|
||
"""happy_existing_session_mount: …"""
|
||
sessions_route = respx.post("https://w.example/sessions").mock(
|
||
return_value=httpx.Response(201, json=_CREATE_OK_RESP)
|
||
)
|
||
app = RatatoskrApp(_args_existing(session_id="s-existing-tail8x"))
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
assert app.session_id == "s-existing-tail8x"
|
||
assert app.state == "idle"
|
||
assert sessions_route.call_count == 0
|
||
# INV-002 carve-out: agent unknown → <unknown> · …<tail>
|
||
assert "<unknown>" in (app.sub_title or "")
|
||
assert app.session_id[-8:] in (app.sub_title or "")
|
||
|
||
@respx.mock
|
||
async def test_agent_not_found_on_mount(
|
||
self, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""agent_not_found_on_mount [error]: …"""
|
||
respx.post("https://w.example/sessions").mock(
|
||
return_value=httpx.Response(404, json={"error": "unknown_agent_id"})
|
||
)
|
||
writes = _spy_writes(monkeypatch)
|
||
app = RatatoskrApp(_args_new())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
assert app.return_value == 12
|
||
assert any("[agent_not_found]" in str(w) for w in writes)
|
||
|
||
@respx.mock
|
||
async def test_session_api_failed_on_mount(
|
||
self, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""session_api_failed_on_mount [error]: …"""
|
||
respx.post("https://w.example/sessions").mock(
|
||
return_value=httpx.Response(500, content=b"server error")
|
||
)
|
||
writes = _spy_writes(monkeypatch)
|
||
app = RatatoskrApp(_args_new())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
assert app.return_value == 20
|
||
assert any("[session_api_failed]" in str(w) and "status=500" in str(w) for w in writes)
|
||
|
||
@respx.mock
|
||
async def test_network_error_on_mount(
|
||
self, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""network_error_on_mount [error]: …"""
|
||
respx.post("https://w.example/sessions").mock(side_effect=httpx.ConnectError("down"))
|
||
writes = _spy_writes(monkeypatch)
|
||
app = RatatoskrApp(_args_new())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
assert app.return_value == 21
|
||
assert any("[network_error]" in str(w) for w in writes)
|
||
|
||
@respx.mock
|
||
async def test_footer_identity_visible_first_frame(self) -> None:
|
||
"""footer_identity_visible_first_frame [trace]: …"""
|
||
from textual.widgets import Static
|
||
|
||
respx.post("https://w.example/sessions").mock(
|
||
return_value=httpx.Response(201, json=_CREATE_OK_RESP)
|
||
)
|
||
app = RatatoskrApp(_args_new())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
identity_widget = app.query_one("#identity", Static)
|
||
rendered = str(identity_widget.render())
|
||
assert "mimir" in rendered
|
||
assert "·" in rendered
|
||
assert app.session_id[-8:] in rendered
|
||
|
||
@respx.mock
|
||
async def test_client_open_after_mount(self) -> None:
|
||
"""client_open_after_mount [trace]: post-mount self.client is open."""
|
||
respx.post("https://w.example/sessions").mock(
|
||
return_value=httpx.Response(201, json=_CREATE_OK_RESP)
|
||
)
|
||
app = RatatoskrApp(_args_new())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
assert app.client is not None
|
||
assert app.client.is_closed is False
|
||
|
||
|
||
class TestAppUnmount:
|
||
@respx.mock
|
||
async def test_unmount_closes_client(self) -> None:
|
||
"""unmount_closes_client [happy,tracer]: …"""
|
||
app = RatatoskrApp(_args_existing())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
client_ref = app.client
|
||
assert client_ref is not None and not client_ref.is_closed
|
||
await pilot.press("ctrl+d")
|
||
await pilot.pause()
|
||
assert client_ref.is_closed
|
||
|
||
|
||
import asyncio # noqa: E402
|
||
|
||
from textual.widgets import Input # noqa: E402
|
||
|
||
|
||
async def _noop_worker(self, content: str) -> None:
|
||
"""Fake _stream_turn_worker that never completes (lets state stay 'streaming')."""
|
||
await asyncio.Future() # await forever; cancelled when test exits
|
||
|
||
|
||
class TestOnInputSubmitted:
|
||
@respx.mock
|
||
async def test_happy_submit_echoes_and_spawns(
|
||
self, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""happy_submit_echoes_and_spawns [happy,tracer]: …"""
|
||
monkeypatch.setattr(RatatoskrApp, "_stream_turn_worker", _noop_worker)
|
||
writes = _spy_writes(monkeypatch)
|
||
app = RatatoskrApp(_args_existing())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
inp = app.query_one("#prompt", Input)
|
||
inp.value = "hello"
|
||
await inp.action_submit()
|
||
await pilot.pause()
|
||
assert any("❯ hello" in str(w) for w in writes) # noqa: RUF001
|
||
assert inp.value == ""
|
||
assert app.state == "streaming"
|
||
assert app.stream_worker is not None
|
||
|
||
@respx.mock
|
||
async def test_empty_submit_no_op(
|
||
self, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""empty_submit_no_op [trace]: '' + Enter → no change; no worker spawned."""
|
||
monkeypatch.setattr(RatatoskrApp, "_stream_turn_worker", _noop_worker)
|
||
app = RatatoskrApp(_args_existing())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
inp = app.query_one("#prompt", Input)
|
||
inp.value = ""
|
||
await inp.action_submit()
|
||
await pilot.pause()
|
||
assert app.state == "idle"
|
||
assert app.stream_worker is None
|
||
|
||
@respx.mock
|
||
async def test_submit_during_streaming_shows_busy_notice(
|
||
self, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""submit_during_streaming_shows_busy_notice [adversarial]: …"""
|
||
monkeypatch.setattr(RatatoskrApp, "_stream_turn_worker", _noop_worker)
|
||
writes = _spy_writes(monkeypatch)
|
||
app = RatatoskrApp(_args_existing())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
inp = app.query_one("#prompt", Input)
|
||
# First submit: enters streaming
|
||
inp.value = "first"
|
||
await inp.action_submit()
|
||
await pilot.pause()
|
||
first_worker = app.stream_worker
|
||
assert app.state == "streaming"
|
||
# Second submit while streaming → busy notice; no new worker
|
||
writes.clear()
|
||
inp.value = "second"
|
||
await inp.action_submit()
|
||
await pilot.pause()
|
||
assert any("[busy] turn in flight; input ignored" in str(w) for w in writes)
|
||
assert app.stream_worker is first_worker # unchanged
|
||
assert app.state == "streaming"
|
||
assert inp.value == ""
|
||
|
||
@respx.mock
|
||
async def test_submit_during_cancelling_shows_busy_notice(
|
||
self, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""submit_during_cancelling_shows_busy_notice [adversarial]: …"""
|
||
monkeypatch.setattr(RatatoskrApp, "_stream_turn_worker", _noop_worker)
|
||
writes = _spy_writes(monkeypatch)
|
||
app = RatatoskrApp(_args_existing())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
app.state = "cancelling" # bypass the natural transition for the test
|
||
inp = app.query_one("#prompt", Input)
|
||
inp.value = "x"
|
||
await inp.action_submit()
|
||
await pilot.pause()
|
||
assert any("[busy]" in str(w) for w in writes)
|
||
assert app.state == "cancelling"
|
||
|
||
@respx.mock
|
||
async def test_footer_hint_flips_to_cancel(
|
||
self, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""footer_hint_flips_to_cancel [trace]: hint widget shows 'Ctrl-C to cancel'."""
|
||
from textual.widgets import Static
|
||
|
||
monkeypatch.setattr(RatatoskrApp, "_stream_turn_worker", _noop_worker)
|
||
app = RatatoskrApp(_args_existing())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
hint_widget = app.query_one("#hint", Static)
|
||
assert str(hint_widget.render()) == RatatoskrApp.HINT_IDLE
|
||
inp = app.query_one("#prompt", Input)
|
||
inp.value = "hi"
|
||
await inp.action_submit()
|
||
await pilot.pause()
|
||
assert str(hint_widget.render()) == RatatoskrApp.HINT_STREAMING
|
||
|
||
|
||
import json # noqa: E402
|
||
|
||
|
||
def _sse_chunk(sse_id: str, body: dict) -> bytes:
|
||
return f"id: {sse_id}\ndata: {json.dumps(body)}\n\n".encode()
|
||
|
||
|
||
_DONE_BODY = {
|
||
"type": "done",
|
||
"phase": "succeeded",
|
||
"response": "hello",
|
||
"model": "m",
|
||
"duration_ms": 1,
|
||
"usage": {
|
||
"prompt_tokens": 0,
|
||
"completion_tokens": 0,
|
||
"total_tokens": 0,
|
||
"cached_input_tokens": 0,
|
||
},
|
||
}
|
||
_CANCELLED_STREAM_BODY = {
|
||
"type": "cancelled",
|
||
"phase": "cancelled",
|
||
"turn_id": 42,
|
||
"reason": "user_cancel",
|
||
"partial_message_id": None,
|
||
}
|
||
|
||
|
||
def _sse_resp(body: bytes | httpx.AsyncByteStream) -> httpx.Response:
|
||
headers = {"content-type": "text/event-stream"}
|
||
if isinstance(body, bytes):
|
||
return httpx.Response(200, headers=headers, content=body)
|
||
return httpx.Response(200, headers=headers, stream=body)
|
||
|
||
|
||
async def _submit_and_wait(app: RatatoskrApp, pilot, content: str) -> None:
|
||
"""Type content into the input and submit; wait for worker to finish."""
|
||
inp = app.query_one("#prompt", Input)
|
||
inp.value = content
|
||
await inp.action_submit()
|
||
await pilot.pause() # let the Input.Submitted message dispatch
|
||
# Poll until the worker resolves (state returns to idle)
|
||
for _ in range(100):
|
||
if app.state == "idle" and app.stream_worker is not None:
|
||
return
|
||
await pilot.pause(0.02)
|
||
|
||
|
||
class TestStreamTurnWorker:
|
||
@respx.mock
|
||
async def test_happy_text_done_renders_markdown(
|
||
self, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""happy_text_done_renders_markdown [happy,tracer]: …"""
|
||
stream = (
|
||
|
||
_sse_chunk("42:1", {"type": "text", "content": "hello"})
|
||
|
||
+ _sse_chunk("42:2", _DONE_BODY)
|
||
|
||
)
|
||
respx.post(
|
||
|
||
"https://w.example/sessions/s-1existing/messages"
|
||
|
||
).mock(return_value=_sse_resp(stream))
|
||
|
||
writes = _spy_writes(monkeypatch)
|
||
|
||
app = RatatoskrApp(_args_existing())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
await _submit_and_wait(app, pilot, "hi")
|
||
assert app.state == "idle"
|
||
# Streamed delta + done label + rule + markdown render
|
||
assert any(w == "hello" for w in writes)
|
||
assert any("[done]" in str(w) for w in writes)
|
||
# The post-Done markdown render uses rich Rule + Markdown — non-string writes
|
||
from rich.markdown import Markdown
|
||
assert any(isinstance(w, Markdown) for w in writes)
|
||
|
||
@respx.mock
|
||
async def test_raw_flag_skips_markdown_render(
|
||
self, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""raw_flag_skips_markdown_render [trace]: …"""
|
||
stream = (
|
||
|
||
_sse_chunk("42:1", {"type": "text", "content": "hi"})
|
||
|
||
+ _sse_chunk("42:2", _DONE_BODY)
|
||
|
||
)
|
||
respx.post(
|
||
|
||
"https://w.example/sessions/s-1existing/messages"
|
||
|
||
).mock(return_value=_sse_resp(stream))
|
||
writes = _spy_writes(monkeypatch)
|
||
app = RatatoskrApp(_args_existing(raw=True))
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
await _submit_and_wait(app, pilot, "x")
|
||
from rich.markdown import Markdown
|
||
assert not any(isinstance(w, Markdown) for w in writes)
|
||
|
||
@respx.mock
|
||
async def test_error_terminal_returns_to_idle(
|
||
self, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""error_terminal_returns_to_idle [happy]: …"""
|
||
stream = (
|
||
|
||
_sse_chunk("42:1", {"type": "text", "content": "x"})
|
||
|
||
+ _sse_chunk(
|
||
"42:2",
|
||
{
|
||
"type": "error",
|
||
"phase": "failed",
|
||
"error_code": "llm_output_invalid",
|
||
"message": "boom",
|
||
},
|
||
)
|
||
|
||
)
|
||
respx.post(
|
||
|
||
"https://w.example/sessions/s-1existing/messages"
|
||
|
||
).mock(return_value=_sse_resp(stream))
|
||
writes = _spy_writes(monkeypatch)
|
||
app = RatatoskrApp(_args_existing())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
await _submit_and_wait(app, pilot, "x")
|
||
assert app.state == "idle"
|
||
assert any("[error]" in str(w) for w in writes)
|
||
|
||
@respx.mock
|
||
async def test_cancelled_terminal_returns_to_idle(
|
||
self, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""cancelled_terminal_returns_to_idle [happy]: …"""
|
||
stream = (
|
||
|
||
_sse_chunk("42:1", {"type": "text", "content": "x"})
|
||
|
||
+ _sse_chunk("42:2", _CANCELLED_STREAM_BODY)
|
||
|
||
)
|
||
respx.post(
|
||
|
||
"https://w.example/sessions/s-1existing/messages"
|
||
|
||
).mock(return_value=_sse_resp(stream))
|
||
writes = _spy_writes(monkeypatch)
|
||
app = RatatoskrApp(_args_existing())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
await _submit_and_wait(app, pilot, "x")
|
||
assert app.state == "idle"
|
||
assert any("[cancelled]" in str(w) for w in writes)
|
||
|
||
@respx.mock
|
||
async def test_active_turn_id_set_on_first_event(
|
||
self, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""active_turn_id_set_on_first_event [trace]: …"""
|
||
# Use a gated stream: yield first event, then hold, so we can inspect mid-stream
|
||
first = _sse_chunk("42:1", {"type": "text", "content": "x"})
|
||
gate = asyncio.Event()
|
||
|
||
class _GatedAfterFirst(httpx.AsyncByteStream):
|
||
async def __aiter__(self):
|
||
yield first
|
||
await gate.wait()
|
||
yield _sse_chunk("42:2", _DONE_BODY)
|
||
|
||
async def aclose(self) -> None:
|
||
return None
|
||
|
||
respx.post("https://w.example/sessions/s-1existing/messages").mock(
|
||
return_value=_sse_resp(_GatedAfterFirst())
|
||
)
|
||
|
||
app = RatatoskrApp(_args_existing())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
inp = app.query_one("#prompt", Input)
|
||
inp.value = "x"
|
||
await inp.action_submit()
|
||
# Wait for first event to be processed (active_turn_id set)
|
||
for _ in range(50):
|
||
if app.active_turn_id is not None:
|
||
break
|
||
await pilot.pause(0.02)
|
||
assert app.active_turn_id == 42
|
||
# Release the gate so the worker can finish and the app can shut down cleanly
|
||
gate.set()
|
||
for _ in range(50):
|
||
if app.state == "idle":
|
||
break
|
||
await pilot.pause(0.02)
|
||
|
||
@respx.mock
|
||
async def test_sse_connect_failed_returns_to_idle(
|
||
self, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""sse_connect_failed_returns_to_idle [error]: …"""
|
||
respx.post("https://w.example/sessions/s-1existing/messages").mock(
|
||
return_value=httpx.Response(404, json={"error": "session_not_found"})
|
||
)
|
||
writes = _spy_writes(monkeypatch)
|
||
app = RatatoskrApp(_args_existing())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
await _submit_and_wait(app, pilot, "x")
|
||
assert app.state == "idle"
|
||
assert any("[sse_connect_failed]" in str(w) for w in writes)
|
||
assert app.return_value is None # app NOT exited per INV-008
|
||
|
||
@respx.mock
|
||
async def test_connection_dropped_returns_to_idle(
|
||
self, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""connection_dropped_returns_to_idle [error]: …"""
|
||
|
||
class _DropAfter(httpx.AsyncByteStream):
|
||
async def __aiter__(self):
|
||
yield _sse_chunk("42:1", {"type": "text", "content": "x"})
|
||
raise httpx.RemoteProtocolError("drop")
|
||
|
||
async def aclose(self) -> None:
|
||
return None
|
||
|
||
respx.post("https://w.example/sessions/s-1existing/messages").mock(
|
||
return_value=_sse_resp(_DropAfter())
|
||
)
|
||
writes = _spy_writes(monkeypatch)
|
||
app = RatatoskrApp(_args_existing())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
await _submit_and_wait(app, pilot, "x")
|
||
assert app.state == "idle"
|
||
assert any("[connection_dropped]" in str(w) for w in writes)
|
||
|
||
@respx.mock
|
||
async def test_rendered_event_per_event(
|
||
self, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""rendered_event_per_event [trace]: …"""
|
||
chunks = (
|
||
_sse_chunk("42:1", {"type": "worker_phase", "phase": "streaming", "turn_id": 42})
|
||
+ _sse_chunk("42:2", {"type": "text", "content": "hi"})
|
||
+ _sse_chunk("42:3", _DONE_BODY)
|
||
)
|
||
respx.post(
|
||
|
||
"https://w.example/sessions/s-1existing/messages"
|
||
|
||
).mock(return_value=_sse_resp(chunks))
|
||
|
||
from ratatoskr import tui as tui_mod
|
||
|
||
call_count = 0
|
||
original = tui_mod._render_event_to_log
|
||
|
||
def spy(event, *, log, raw):
|
||
nonlocal call_count
|
||
call_count += 1
|
||
return original(event, log=log, raw=raw)
|
||
|
||
monkeypatch.setattr(tui_mod, "_render_event_to_log", spy)
|
||
app = RatatoskrApp(_args_existing())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
await _submit_and_wait(app, pilot, "x")
|
||
assert call_count == 3
|
||
|
||
|
||
class TestActionInterrupt:
|
||
@respx.mock
|
||
async def test_idle_ctrl_c_exits_zero(self) -> None:
|
||
"""idle_ctrl_c_exits_zero [happy,tracer]: state=idle; ctrl+c → exit(0)."""
|
||
app = RatatoskrApp(_args_existing())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
assert app.state == "idle"
|
||
await pilot.press("ctrl+c")
|
||
await pilot.pause()
|
||
assert app.return_value == 0
|
||
|
||
@respx.mock
|
||
async def test_streaming_first_ctrl_c_cancels(
|
||
self, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""streaming_first_ctrl_c_cancels [scenario,tracer]: …"""
|
||
# Stream that yields one text event (sets active_turn_id) then waits forever
|
||
first_chunk = _sse_chunk("42:1", {"type": "text", "content": "x"})
|
||
gate = asyncio.Event()
|
||
|
||
class _GatedAfterFirst(httpx.AsyncByteStream):
|
||
async def __aiter__(self):
|
||
yield first_chunk
|
||
await gate.wait()
|
||
|
||
async def aclose(self) -> None:
|
||
return None
|
||
|
||
respx.post("https://w.example/sessions/s-1existing/messages").mock(
|
||
return_value=_sse_resp(_GatedAfterFirst())
|
||
)
|
||
cancel_observed = asyncio.Event()
|
||
|
||
def cancel_handler(req: httpx.Request) -> httpx.Response:
|
||
cancel_observed.set()
|
||
return httpx.Response(200, json=_CANCEL_OK_RESP)
|
||
|
||
cancel_route = respx.post("https://w.example/sessions/s-1existing/turns/42/cancel").mock(
|
||
side_effect=cancel_handler
|
||
)
|
||
|
||
app = RatatoskrApp(_args_existing())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
inp = app.query_one("#prompt", Input)
|
||
inp.value = "go"
|
||
await inp.action_submit()
|
||
await pilot.pause()
|
||
# Wait for active_turn_id to be set (first event consumed)
|
||
for _ in range(50):
|
||
if app.active_turn_id == 42:
|
||
break
|
||
await pilot.pause(0.02)
|
||
assert app.active_turn_id == 42
|
||
assert app.state == "streaming"
|
||
await pilot.press("ctrl+c")
|
||
# Wait for cancel POST to land
|
||
for _ in range(50):
|
||
if cancel_observed.is_set():
|
||
break
|
||
await pilot.pause(0.02)
|
||
assert cancel_route.call_count == 1
|
||
assert app.state == "cancelling"
|
||
from textual.widgets import Static
|
||
|
||
hint_widget = app.query_one("#hint", Static)
|
||
assert str(hint_widget.render()) == RatatoskrApp.HINT_CANCELLING
|
||
# Release the gate so the stream worker can finish cleanly during teardown
|
||
gate.set()
|
||
|
||
@respx.mock
|
||
async def test_streaming_no_turn_id_force_exits(
|
||
self, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""streaming_no_turn_id_force_exits [scenario]: …"""
|
||
cancel_route = respx.post("https://w.example/sessions/s-1existing/turns/0/cancel").mock(
|
||
return_value=httpx.Response(200, json=_CANCEL_OK_RESP)
|
||
)
|
||
# Stream that hangs forever (no events to set active_turn_id)
|
||
gate = asyncio.Event()
|
||
|
||
class _NeverYields(httpx.AsyncByteStream):
|
||
async def __aiter__(self):
|
||
await gate.wait()
|
||
if False:
|
||
yield b""
|
||
|
||
async def aclose(self) -> None:
|
||
return None
|
||
|
||
respx.post("https://w.example/sessions/s-1existing/messages").mock(
|
||
return_value=_sse_resp(_NeverYields())
|
||
)
|
||
|
||
app = RatatoskrApp(_args_existing())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
inp = app.query_one("#prompt", Input)
|
||
inp.value = "go"
|
||
await inp.action_submit()
|
||
await pilot.pause()
|
||
assert app.state == "streaming"
|
||
assert app.active_turn_id is None
|
||
await pilot.press("ctrl+c")
|
||
await pilot.pause()
|
||
gate.set() # let the gated stream resolve so teardown is clean
|
||
assert app.return_value == 3
|
||
assert cancel_route.call_count == 0
|
||
|
||
@respx.mock
|
||
async def test_cancelling_second_ctrl_c_force_exits(self) -> None:
|
||
"""cancelling_second_ctrl_c_force_exits [scenario]: …"""
|
||
app = RatatoskrApp(_args_existing())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
app.state = "cancelling" # bypass the natural transition for the test
|
||
await pilot.press("ctrl+c")
|
||
await pilot.pause()
|
||
assert app.return_value == 3
|
||
|
||
@respx.mock
|
||
async def test_cancel_failed_swallowed(self) -> None:
|
||
"""cancel_failed_swallowed [scenario]: …"""
|
||
first_chunk = _sse_chunk("42:1", {"type": "text", "content": "x"})
|
||
stream_gate = asyncio.Event()
|
||
|
||
class _GatedAfterFirst(httpx.AsyncByteStream):
|
||
async def __aiter__(self):
|
||
yield first_chunk
|
||
await stream_gate.wait()
|
||
|
||
async def aclose(self) -> None:
|
||
return None
|
||
|
||
respx.post("https://w.example/sessions/s-1existing/messages").mock(
|
||
return_value=_sse_resp(_GatedAfterFirst())
|
||
)
|
||
cancel_observed = asyncio.Event()
|
||
|
||
def cancel_handler(req: httpx.Request) -> httpx.Response:
|
||
cancel_observed.set()
|
||
return httpx.Response(500, content=b"boom")
|
||
|
||
respx.post("https://w.example/sessions/s-1existing/turns/42/cancel").mock(
|
||
side_effect=cancel_handler
|
||
)
|
||
|
||
app = RatatoskrApp(_args_existing())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
inp = app.query_one("#prompt", Input)
|
||
inp.value = "go"
|
||
await inp.action_submit()
|
||
await pilot.pause()
|
||
for _ in range(50):
|
||
if app.active_turn_id == 42:
|
||
break
|
||
await pilot.pause(0.02)
|
||
await pilot.press("ctrl+c")
|
||
for _ in range(50):
|
||
if cancel_observed.is_set():
|
||
break
|
||
await pilot.pause(0.02)
|
||
# Give _cancel_via_sse time to write the [cancel_failed] line
|
||
await pilot.pause(0.05)
|
||
log = app.query_one("#transcript", RichLog)
|
||
rendered = "\n".join(str(strip.text) for strip in log.lines)
|
||
assert "[cancel_failed]" in rendered
|
||
assert app.state == "cancelling"
|
||
stream_gate.set() # let stream finish for teardown
|
||
|
||
|
||
class TestActionQuit:
|
||
@respx.mock
|
||
async def test_idle_ctrl_d_exits_zero(self) -> None:
|
||
"""idle_ctrl_d_exits_zero [happy,tracer]: state=idle; ctrl+d → exit(0)."""
|
||
app = RatatoskrApp(_args_existing())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
await pilot.press("ctrl+d")
|
||
await pilot.pause()
|
||
assert app.return_value == 0
|
||
|
||
@respx.mock
|
||
async def test_streaming_ctrl_d_force_exits(
|
||
self, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""streaming_ctrl_d_force_exits [scenario]: …"""
|
||
cancel_route = respx.post("https://w.example/sessions/s-1existing/turns/42/cancel").mock(
|
||
return_value=httpx.Response(200, json=_CANCEL_OK_RESP)
|
||
)
|
||
first_chunk = _sse_chunk("42:1", {"type": "text", "content": "x"})
|
||
gate = asyncio.Event()
|
||
|
||
class _GatedAfterFirst(httpx.AsyncByteStream):
|
||
async def __aiter__(self):
|
||
yield first_chunk
|
||
await gate.wait()
|
||
|
||
async def aclose(self) -> None:
|
||
return None
|
||
|
||
respx.post("https://w.example/sessions/s-1existing/messages").mock(
|
||
return_value=_sse_resp(_GatedAfterFirst())
|
||
)
|
||
app = RatatoskrApp(_args_existing())
|
||
async with app.run_test() as pilot:
|
||
await pilot.pause()
|
||
inp = app.query_one("#prompt", Input)
|
||
inp.value = "go"
|
||
await inp.action_submit()
|
||
await pilot.pause()
|
||
for _ in range(50):
|
||
if app.active_turn_id == 42:
|
||
break
|
||
await pilot.pause(0.02)
|
||
await pilot.press("ctrl+d")
|
||
await pilot.pause()
|
||
gate.set()
|
||
assert app.return_value == 0
|
||
assert cancel_route.call_count == 0
|
||
|
||
|
||
from ratatoskr.tui import run_tui # noqa: E402
|
||
|
||
|
||
class TestRunTui:
|
||
def test_happy_returns_zero_on_quit(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""happy_returns_zero_on_quit [happy,tracer]: run_tui propagates App.run() exit code."""
|
||
captured: list[ParsedArgs] = []
|
||
|
||
def fake_run(self) -> int:
|
||
captured.append(self.args)
|
||
return 0
|
||
|
||
monkeypatch.setattr(RatatoskrApp, "run", fake_run)
|
||
rc = run_tui(_args_existing())
|
||
assert rc == 0
|
||
assert len(captured) == 1
|
||
assert captured[0].send_content is None
|
||
|
||
def test_precondition_send_content_none(self) -> None:
|
||
"""precondition_send_content_none [adversarial]: …"""
|
||
bad_args = ParsedArgs(
|
||
send_content="x", # PRE-001 violation
|
||
session_id="s-1",
|
||
new=False,
|
||
agent_id=None,
|
||
api_key="k",
|
||
server_url="https://w.example",
|
||
raw=False,
|
||
)
|
||
with pytest.raises(AssertionError):
|
||
run_tui(bad_args)
|