vh/ratatoskr
1
0
Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
ffd22fb587ab19fa8a90799962cf89ad18b3a2f0
ratatoskr/tests/test_sse_client.py
T
vh c713208585 feat(sse_client,cli,tui): implement issue #7 — empty-data skip + MalformedSseData 
Bundles initial TDD impl + Volva-code-review F1/F3 amendments.

sse_client.py:
- New MalformedSseData(raw) exception; truncates raw to 200 chars at
  __init__ (mirrors MalformedSseId.raw[:64] precedent).
- _iter_events gains `if sse.data == '': continue` BEFORE
  _parse_sse_id. Empty-data frames are silently skipped per issue #7
  INV-001 (keepalive semantics). Empty-data + bad-id is still a
  keepalive; intentional ordering, don't reorder.
- _iter_events json.loads(sse.data) now wrapped — JSONDecodeError →
  MalformedSseData(raw=sse.data).

cli.py:
- Imports MalformedSseData; _run_turn ERROR_ROUTING gains the case →
  stderr `[malformed_sse_data] raw={exc.raw!r}` + exit 22 (protocol-
  failure bucket, same as MalformedSseId/TurnIdFlip).

tui.py:
- Imports MalformedSseData; _stream_turn_worker ERROR_ROUTING gains
  the case → transcript label; finally block restores state→idle
  per INV-008 (mid-session errors don't exit the app).

Tests (6 new):
- test_sse_client.py: empty_data_skipped (tracer — 4 frames in, 3
  events out), malformed_data_raises, whitespace_data_raises,
  malformed_data_truncation, AND empty_data_skip_preserves_last_seen_sse_id
  (F1 from Volva code-review — drop-after-empty probes internal
  last_sse_id non-advancement via SseConnectionDropped.last_seen_sse_id).
- test_cli.py: malformed_sse_data (tightened to assert exact
  `[malformed_sse_data] raw='not-json'` shape per F3),
  malformed_sse_data_truncation (5000-char payload — verifies
  truncation carries through presenter rendering, F3).
- test_tui.py: malformed_sse_data_returns_to_idle (state→idle per
  INV-008; app does NOT exit).

Smoke validation (2026-05-22): the original crashing prompt
("what about system 1 and system 2 framing?") now completes cleanly
end-to-end. mimir streamed 3193 tokens (50 seconds, 374980-token
context), `[done] turn_id=96 duration_ms=50436`. Empty-data frames
somewhere in the stream silently skipped; no crash.

172/172 tests GREEN; ruff clean; all 5 issue contracts (#1, #3, #4,
#5, #7) drift-check clean.

Persistent-memory updated per the commit-along rule: status reflects
v0+#7 milestone; new dated decisions for #5/#6/#7 filing + #7
implementation; foot-gun entry for unguarded json.loads(sse.data).
2026-05-22 16:41:38 -07:00

839 lines
35 KiB
Python
Raw Blame History

"""Tests for ratatoskr.sse_client per docs/contracts/issues/1.contract.md."""
import httpx
import pytest
import respx
from ratatoskr.sse_client import (
CancelAlreadyCompleted,
Cancelled,
CancelResult,
CancelTurnNotFound,
Done,
Error,
InvalidLastEventId,
MalformedSseId,
ResumeBufferExpired,
ResumeTurnFinished,
SseConnectFailed,
SseId,
Text,
TurnIdFlip,
_parse_sse_id,
cancel_turn,
reconnect_turn,
stream_turn,
)
_DONE_42_6 = {
"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,
},
}
def _sse_chunk(sse_id: str, body: dict[str, object]) -> bytes:
"""Compose one SSE event in wire format. Trailing blank line per spec."""
import json
return f"id: {sse_id}\ndata: {json.dumps(body)}\n\n".encode()
class TestParseSseId:
def test_happy_simple(self) -> None:
"""happy_simple [happy,tracer]: '42:3' -> SseId(turn_id=42, seq=3)."""
assert _parse_sse_id("42:3") == SseId(turn_id=42, seq=3)
def test_happy_seq_one(self) -> None:
"""happy_seq_one: smallest valid id per spec — first event of first turn."""
assert _parse_sse_id("1:1") == SseId(turn_id=1, seq=1)
def test_empty(self) -> None:
"""empty [adversarial]: '' -> ValueError."""
with pytest.raises(ValueError):
_parse_sse_id("")
def test_no_colon(self) -> None:
"""no_colon [adversarial]: '42' -> ValueError with '42' in the message."""
with pytest.raises(ValueError, match="42"):
_parse_sse_id("42")
def test_too_many_colons(self) -> None:
"""too_many_colons [adversarial]: '42:3:7' -> ValueError."""
with pytest.raises(ValueError):
_parse_sse_id("42:3:7")
def test_alpha_turn_id(self) -> None:
"""alpha_turn_id [adversarial]: 'foo:3' -> ValueError."""
with pytest.raises(ValueError):
_parse_sse_id("foo:3")
def test_alpha_seq(self) -> None:
"""alpha_seq [adversarial]: '42:bar' -> ValueError."""
with pytest.raises(ValueError):
_parse_sse_id("42:bar")
def test_zero_turn_id(self) -> None:
"""zero_turn_id [adversarial]: '0:3' -> ValueError (turn_id ≥ 1)."""
with pytest.raises(ValueError):
_parse_sse_id("0:3")
def test_zero_seq(self) -> None:
"""zero_seq [adversarial]: '1:0' -> ValueError (seq ≥ 1, spec line 705)."""
with pytest.raises(ValueError):
_parse_sse_id("1:0")
def test_negative_turn_id(self) -> None:
"""negative_turn_id [adversarial]: '-1:3' -> ValueError."""
with pytest.raises(ValueError):
_parse_sse_id("-1:3")
def test_negative_seq(self) -> None:
"""negative_seq [adversarial]: '42:-1' -> ValueError."""
with pytest.raises(ValueError):
_parse_sse_id("42:-1")
def test_trailing_whitespace(self) -> None:
"""trailing_whitespace [adversarial]: '42:3 ' -> ValueError (strict; no strip)."""
with pytest.raises(ValueError):
_parse_sse_id("42:3 ")
def test_truncation(self) -> None:
"""truncation [security]: 5000-char no-colon -> ValueError msg contains only raw[:64]."""
raw = "a" * 5000
with pytest.raises(ValueError) as exc_info:
_parse_sse_id(raw)
assert raw not in str(exc_info.value), "full 5000-char input must not appear in message"
assert raw[:64] in str(exc_info.value), "first 64 chars must appear in message"
def test_non_string_input(self) -> None:
"""PRE-001 hard: raw is a string -- isinstance check before parse."""
with pytest.raises(AssertionError):
_parse_sse_id(42) # type: ignore[arg-type]
with pytest.raises(AssertionError):
_parse_sse_id(None) # type: ignore[arg-type]
class TestStreamTurn:
@respx.mock
async def test_happy_one_text_done(self) -> None:
"""happy_one_text_done: text then done; same turn_id; iter ends after done."""
stream = _sse_chunk("42:1", {"type": "text", "content": "hello"}) + _sse_chunk(
"42:2",
{
"type": "done",
"phase": "succeeded",
"response": "hello",
"model": "glm5-turbo",
"duration_ms": 100,
"usage": {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0,
"cached_input_tokens": 0,
},
},
)
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(
200, headers={"content-type": "text/event-stream"}, content=stream
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
events = [e async for e in stream_turn(client, "s1", "hi")]
assert len(events) == 2
assert isinstance(events[0], Text)
assert events[0].content == "hello"
assert events[0].sse_id == SseId(42, 1)
assert isinstance(events[1], Done)
assert events[1].sse_id == SseId(42, 2)
assert events[0].sse_id.turn_id == events[1].sse_id.turn_id
@respx.mock
async def test_full_event_vocab(self) -> None:
"""full_event_vocab: one of each variant; all carry parsed sse_id."""
stream = b"".join(
[
_sse_chunk(
"42:1",
{"type": "worker_phase", "phase": "BuildingPrompt", "turn_id": 42},
),
_sse_chunk("42:2", {"type": "thinking", "content": "thinking out loud"}),
_sse_chunk("42:3", {"type": "text", "content": "hello"}),
_sse_chunk(
"42:4",
{
"type": "text_boundary",
"kind": "sentence",
"char_offset": 5,
"ts": "2026-05-21T00:00:00Z",
},
),
_sse_chunk(
"42:5",
{"type": "tool_start", "name": "search", "arguments": {"q": "x"}},
),
_sse_chunk(
"42:6",
{
"type": "tool_result",
"name": "search",
"result": {"n": 1},
"duration_ms": 12,
},
),
_sse_chunk("42:7", {"type": "text", "content": " world"}),
_sse_chunk(
"42:8",
{
"type": "done",
"phase": "succeeded",
"response": "hello world",
"model": "m",
"duration_ms": 50,
"usage": {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0,
"cached_input_tokens": 0,
},
},
),
]
)
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(
200, headers={"content-type": "text/event-stream"}, content=stream
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
events = [e async for e in stream_turn(client, "s1", "hi")]
types_seen = [type(e).__name__ for e in events]
assert types_seen == [
"WorkerPhase",
"Thinking",
"Text",
"TextBoundary",
"ToolStart",
"ToolResult",
"Text",
"Done",
]
# INV-002: every event carries parsed sse_id with turn_id, seq >= 1
for e in events:
assert e.sse_id.turn_id == 42
assert e.sse_id.seq >= 1
@respx.mock
async def test_error_terminal(self) -> None:
"""error_terminal: text then error; iteration ends; error_code populated."""
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/s1/messages").mock(
return_value=httpx.Response(
200, headers={"content-type": "text/event-stream"}, content=stream
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
events = [e async for e in stream_turn(client, "s1", "hi")]
assert len(events) == 2
assert isinstance(events[1], Error)
assert events[1].error_code == "llm_output_invalid"
assert events[1].message == "boom"
@respx.mock
async def test_cancelled_terminal(self) -> None:
"""cancelled_terminal: cancelled with phase=cancelled, turn_id; iteration ends."""
stream = _sse_chunk(
"42:1",
{
"type": "cancelled",
"phase": "cancelled",
"turn_id": 42,
"reason": "user_cancel",
"partial_message_id": None,
},
)
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(
200, headers={"content-type": "text/event-stream"}, content=stream
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
events = [e async for e in stream_turn(client, "s1", "hi")]
assert len(events) == 1
assert isinstance(events[0], Cancelled)
assert events[0].turn_id == 42
assert events[0].reason == "user_cancel"
@respx.mock
async def test_session_not_found(self) -> None:
"""session_not_found: 404 -> SseConnectFailed(status=404)."""
respx.post("https://w.example/sessions/missing/messages").mock(
return_value=httpx.Response(404, json={"error": "session_not_found"})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SseConnectFailed) as exc_info:
_ = [e async for e in stream_turn(client, "missing", "hi")]
assert exc_info.value.status == 404
@respx.mock
async def test_malformed_id_no_seq(self) -> None:
"""malformed_id_no_seq: id `42` (missing :seq) -> MalformedSseId; no event yielded."""
stream = b"id: 42\ndata: {\"type\":\"text\",\"content\":\"x\"}\n\n"
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(
200, headers={"content-type": "text/event-stream"}, content=stream
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
events: list[object] = []
with pytest.raises(MalformedSseId):
async for e in stream_turn(client, "s1", "hi"):
events.append(e)
assert events == []
@respx.mock
async def test_malformed_id_alpha(self) -> None:
"""malformed_id_alpha: id `foo:bar` -> MalformedSseId."""
stream = b"id: foo:bar\ndata: {\"type\":\"text\",\"content\":\"x\"}\n\n"
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(
200, headers={"content-type": "text/event-stream"}, content=stream
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(MalformedSseId):
_ = [e async for e in stream_turn(client, "s1", "hi")]
@respx.mock
async def test_turn_id_flip(self) -> None:
"""turn_id_flip: 42:1 then 99:2 -> TurnIdFlip; only first yielded."""
stream = _sse_chunk("42:1", {"type": "text", "content": "a"}) + _sse_chunk(
"99:2", {"type": "text", "content": "b"}
)
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(
200, headers={"content-type": "text/event-stream"}, content=stream
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
yielded: list[Text] = []
with pytest.raises(TurnIdFlip) as exc_info:
async for e in stream_turn(client, "s1", "hi"):
yielded.append(e) # type: ignore[arg-type]
assert len(yielded) == 1
assert exc_info.value.established == 42
assert exc_info.value.got == 99
@respx.mock
async def test_connection_drop(self) -> None:
"""connection_drop: RemoteProtocolError after one text -> SseConnectionDropped((42,1))."""
from ratatoskr.sse_client import SseConnectionDropped
class _DropAfter(httpx.AsyncByteStream):
def __init__(self, chunks: list[bytes]) -> None:
self._chunks = chunks
async def __aiter__(self): # type: ignore[no-untyped-def]
for c in self._chunks:
yield c
raise httpx.RemoteProtocolError("simulated mid-stream drop")
async def aclose(self) -> None:
return None
first = _sse_chunk("42:1", {"type": "text", "content": "x"})
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(
200,
headers={"content-type": "text/event-stream"},
stream=_DropAfter([first]),
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
yielded: list[Text] = []
with pytest.raises(SseConnectionDropped) as exc_info:
async for e in stream_turn(client, "s1", "hi"):
yielded.append(e) # type: ignore[arg-type]
assert len(yielded) == 1
assert exc_info.value.last_seen_sse_id == SseId(42, 1)
@respx.mock
async def test_clean_eof_before_terminal(self) -> None:
"""INV-001: clean EOF without Done/Error/Cancelled -> SseConnectionDropped."""
from ratatoskr.sse_client import SseConnectionDropped
# One text event, then stream ends cleanly (no terminal).
stream = _sse_chunk("42:1", {"type": "text", "content": "x"})
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(
200, headers={"content-type": "text/event-stream"}, content=stream
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
yielded: list[Text] = []
with pytest.raises(SseConnectionDropped) as exc_info:
async for e in stream_turn(client, "s1", "hi"):
yielded.append(e) # type: ignore[arg-type]
assert len(yielded) == 1
assert exc_info.value.last_seen_sse_id == SseId(42, 1)
@respx.mock
async def test_zero_event_eof(self) -> None:
"""INV-001: empty stream (no events at all) -> SseConnectionDropped(None)."""
from ratatoskr.sse_client import SseConnectionDropped
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(
200, headers={"content-type": "text/event-stream"}, content=b""
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SseConnectionDropped) as exc_info:
_ = [e async for e in stream_turn(client, "s1", "hi")]
assert exc_info.value.last_seen_sse_id is None
@respx.mock
async def test_connect_failed_body_truncated(self) -> None:
"""ERROR_ROUTING: SseConnectFailed.body is truncated to <= 1024 bytes."""
big_body = b"x" * 5000
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(503, content=big_body)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SseConnectFailed) as exc_info:
_ = [e async for e in stream_turn(client, "s1", "hi")]
assert len(exc_info.value.body) <= 1024
assert exc_info.value.body == big_body[:1024]
@respx.mock
async def test_no_text_aggregation(self) -> None:
"""no_text_aggregation: consumer yields each text event separately; no concat."""
stream = (
_sse_chunk("42:1", {"type": "text", "content": "hello "})
+ _sse_chunk("42:2", {"type": "text", "content": "world"})
+ _sse_chunk(
"42:3",
{
"type": "done",
"phase": "succeeded",
"response": "hello world",
"model": "m",
"duration_ms": 1,
"usage": {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0,
"cached_input_tokens": 0,
},
},
)
)
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(
200, headers={"content-type": "text/event-stream"}, content=stream
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
events = [e async for e in stream_turn(client, "s1", "hi")]
texts = [e for e in events if isinstance(e, Text)]
assert [t.content for t in texts] == ["hello ", "world"]
class TestReconnectTurn:
@respx.mock
async def test_happy_resume_from_seq_3(self) -> None:
"""happy_resume_from_seq_3 [scenario,tracer]: replay 42:4,42:5 then live 42:6 done."""
stream = (
_sse_chunk("42:4", {"type": "text", "content": "d"})
+ _sse_chunk("42:5", {"type": "text", "content": "e"})
+ _sse_chunk("42:6", _DONE_42_6)
)
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(
200, headers={"content-type": "text/event-stream"}, content=stream
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
events = [e async for e in reconnect_turn(client, "s1", "original", "42:3")]
assert len(events) == 3
assert [e.sse_id.seq for e in events] == [4, 5, 6]
assert all(e.sse_id.turn_id == 42 for e in events)
assert isinstance(events[-1], Done)
@respx.mock
async def test_header_verbatim(self) -> None:
"""header_verbatim: outbound carries Last-Event-ID: 42:3 exactly."""
stream = _sse_chunk("42:4", _DONE_42_6)
route = respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(
200, headers={"content-type": "text/event-stream"}, content=stream
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
_ = [e async for e in reconnect_turn(client, "s1", "x", "42:3")]
req = route.calls[0].request
assert req.headers["Last-Event-ID"] == "42:3"
@respx.mock
async def test_body_threads_content(self) -> None:
"""body_threads_content: outbound JSON body is {'content': <content>} byte-for-byte."""
import json as _json
stream = _sse_chunk("42:4", _DONE_42_6)
route = respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(
200, headers={"content-type": "text/event-stream"}, content=stream
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
_ = [
e async for e in reconnect_turn(client, "s1", "hello mimir", "42:3")
]
body = _json.loads(route.calls[0].request.content)
assert body == {"content": "hello mimir"}
@respx.mock
async def test_buffer_expired(self) -> None:
"""buffer_expired: 412 -> ResumeBufferExpired(turn_id=42, buffered_from_seq=10)."""
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(
412,
json={"error": "buffer_expired", "buffered_from_seq": 10, "turn_id": 42},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(ResumeBufferExpired) as exc_info:
_ = [e async for e in reconnect_turn(client, "s1", "x", "42:3")]
assert exc_info.value.turn_id == 42
assert exc_info.value.buffered_from_seq == 10
@respx.mock
async def test_turn_finished(self) -> None:
"""turn_finished: 410 -> ResumeTurnFinished(turn_id=42)."""
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(
410, json={"error": "turn_finished", "turn_id": 42}
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(ResumeTurnFinished) as exc_info:
_ = [e async for e in reconnect_turn(client, "s1", "x", "42:3")]
assert exc_info.value.turn_id == 42
@respx.mock
async def test_invalid_last_event_id_server(self) -> None:
"""invalid_last_event_id_server: 400 -> InvalidLastEventId."""
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(400, json={"error": "invalid_last_event_id"})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(InvalidLastEventId):
_ = [e async for e in reconnect_turn(client, "s1", "x", "42:3")]
@respx.mock
async def test_malformed_input(self) -> None:
"""malformed_input: caller passes '42' (no seq) -> ValueError; no HTTP issued."""
route = respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(200, content=b"")
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(ValueError):
_ = [e async for e in reconnect_turn(client, "s1", "x", "42")]
assert route.call_count == 0
@respx.mock
async def test_turn_id_flip_on_first_event(self) -> None:
"""turn_id_flip_on_first_event: server's first event 99:4 -> TurnIdFlip pre-yield."""
stream = _sse_chunk("99:4", {"type": "text", "content": "wrong turn"})
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(
200, headers={"content-type": "text/event-stream"}, content=stream
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
yielded: list[Text] = []
with pytest.raises(TurnIdFlip) as exc_info:
async for e in reconnect_turn(client, "s1", "x", "42:3"):
yielded.append(e) # type: ignore[arg-type]
assert yielded == []
assert exc_info.value.established == 42
assert exc_info.value.got == 99
class TestCancelTurn:
@respx.mock
async def test_happy_cancel(self) -> None:
"""happy_cancel [happy,tracer]: 200 -> CancelResult(42, True, None, None)."""
respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
return_value=httpx.Response(
200,
json={
"turn_id": 42,
"cancelled": True,
"reason": None,
"partial_message_id": None,
},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
result = await cancel_turn(client, "s1", 42)
assert result == CancelResult(
turn_id=42, cancelled=True, reason=None, partial_message_id=None
)
@respx.mock
async def test_persist_partial_query(self) -> None:
"""persist_partial_query [trace]: persist_partial=True -> URL has ?persist_partial=true."""
route = respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
return_value=httpx.Response(
200,
json={
"turn_id": 42,
"cancelled": True,
"reason": None,
"partial_message_id": None,
},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
await cancel_turn(client, "s1", 42, persist_partial=True)
assert "persist_partial=true" in str(route.calls[0].request.url)
@respx.mock
async def test_default_no_persist(self) -> None:
"""default_no_persist [trace]: default call -> URL has no persist_partial param."""
route = respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
return_value=httpx.Response(
200,
json={
"turn_id": 42,
"cancelled": True,
"reason": None,
"partial_message_id": None,
},
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
await cancel_turn(client, "s1", 42)
assert "persist_partial" not in str(route.calls[0].request.url)
@respx.mock
async def test_cancel_already_completed(self) -> None:
"""cancel_already_completed [error]: 409 -> CancelAlreadyCompleted(turn_id=42)."""
respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
return_value=httpx.Response(409, json={"error": "turn_already_completed"})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(CancelAlreadyCompleted) as exc_info:
await cancel_turn(client, "s1", 42)
assert exc_info.value.turn_id == 42
@respx.mock
async def test_cancel_turn_not_found(self) -> None:
"""cancel_turn_not_found [error]: 404 -> CancelTurnNotFound(turn_id=42)."""
respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
return_value=httpx.Response(404, json={"error": "turn_not_found"})
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(CancelTurnNotFound) as exc_info:
await cancel_turn(client, "s1", 42)
assert exc_info.value.turn_id == 42
@respx.mock
async def test_cancel_failed_truncates_body(self) -> None:
"""ERROR_ROUTING: 5xx -> CancelFailed; body truncated to <= 1024 bytes."""
from ratatoskr.sse_client import CancelFailed
big_body = b"x" * 5000
respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
return_value=httpx.Response(503, content=big_body)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(CancelFailed) as exc_info:
await cancel_turn(client, "s1", 42)
assert exc_info.value.status == 503
assert len(exc_info.value.body) <= 1024
assert exc_info.value.body == big_body[:1024]
@respx.mock
async def test_cancel_idempotent(self) -> None:
"""cancel_idempotent: first 200 -> happy; second 409 -> CancelAlreadyCompleted."""
respx.post("https://w.example/sessions/s1/turns/42/cancel").mock(
side_effect=[
httpx.Response(
200,
json={
"turn_id": 42,
"cancelled": True,
"reason": None,
"partial_message_id": None,
},
),
httpx.Response(409, json={"error": "turn_already_completed"}),
]
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
first = await cancel_turn(client, "s1", 42)
assert first.cancelled is True
with pytest.raises(CancelAlreadyCompleted):
await cancel_turn(client, "s1", 42)
# ============================================================================
# Issue #7: empty-data skip + MalformedSseData raise
# ============================================================================
def _sse_empty_chunk(sse_id: str) -> bytes:
"""SSE frame with id but empty data (server-emitted keepalive shape)."""
return f"id: {sse_id}\ndata:\n\n".encode()
def _sse_raw_chunk(sse_id: str, raw_data: str) -> bytes:
"""SSE frame with id + arbitrary raw data (for testing malformed JSON)."""
return f"id: {sse_id}\ndata: {raw_data}\n\n".encode()
class TestEmptyDataSkipped:
@respx.mock
async def test_empty_data_skipped(self) -> None:
"""empty_data_skipped [trace]: 4 frames in, 3 events out; skip preserves last_sse_id."""
from ratatoskr.sse_client import Done as _Done
from ratatoskr.sse_client import Text as _Text
stream = (
_sse_chunk("42:1", {"type": "text", "content": "first"})
+ _sse_empty_chunk("42:2") # ← skipped silently
+ _sse_chunk("42:3", {"type": "text", "content": "second"})
+ _sse_chunk("42:4", _DONE_42_6)
)
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(
200, headers={"content-type": "text/event-stream"}, content=stream
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
events = [e async for e in stream_turn(client, "s1", "hi")]
# Exactly 3 events: Text, Text, Done — empty-data frame at 42:2 is invisible
assert len(events) == 3
assert isinstance(events[0], _Text) and events[0].sse_id == SseId(42, 1)
assert isinstance(events[1], _Text) and events[1].sse_id == SseId(42, 3)
assert isinstance(events[2], _Done) and events[2].sse_id == SseId(42, 4)
# Per INV-003: skip MUST NOT advance through 42:2. The second Text's sse_id
# is (42, 3) — directly verifies the skip didn't bookkeep 42:2.
assert events[1].sse_id.seq == 3, "skip advanced through 42:2"
@respx.mock
async def test_empty_data_skip_preserves_last_seen_sse_id(self) -> None:
"""empty_skip_does_not_advance [trace]: drop-after-empty → last_seen is last real event."""
from ratatoskr.sse_client import SseConnectionDropped
# Stream: text(42:1), empty(42:2), then drop. Per INV-003, the skipped
# 42:2 must NOT advance internal last_sse_id. If the consumer caught a
# drop, SseConnectionDropped.last_seen_sse_id should be (42, 1) — the
# last *real* event — NOT (42, 2).
first = _sse_chunk("42:1", {"type": "text", "content": "x"})
empty = _sse_empty_chunk("42:2")
class _DropAfterEmpty(httpx.AsyncByteStream):
async def __aiter__(self): # type: ignore[no-untyped-def]
yield first
yield empty
raise httpx.RemoteProtocolError("drop after skip")
async def aclose(self) -> None:
return None
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(
200,
headers={"content-type": "text/event-stream"},
stream=_DropAfterEmpty(),
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(SseConnectionDropped) as exc_info:
async for _ in stream_turn(client, "s1", "hi"):
pass
assert exc_info.value.last_seen_sse_id == SseId(42, 1), (
f"skip advanced last_sse_id through 42:2; got {exc_info.value.last_seen_sse_id}"
)
@respx.mock
async def test_malformed_data_raises(self) -> None:
"""malformed_data_raises [error]: text + bad-JSON → yields Text then MalformedSseData."""
from ratatoskr.sse_client import MalformedSseData
from ratatoskr.sse_client import Text as _Text
stream = (
_sse_chunk("42:1", {"type": "text", "content": "hi"})
+ _sse_raw_chunk("42:2", "not-json")
)
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(
200, headers={"content-type": "text/event-stream"}, content=stream
)
)
yielded: list = []
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(MalformedSseData) as exc_info:
async for e in stream_turn(client, "s1", "hi"):
yielded.append(e)
assert len(yielded) == 1
assert isinstance(yielded[0], _Text)
assert exc_info.value.raw == "not-json"
@respx.mock
async def test_whitespace_data_raises(self) -> None:
"""whitespace_data_raises [adv]: single-space data → MalformedSseData (NOT skipped)."""
from ratatoskr.sse_client import MalformedSseData
stream = _sse_raw_chunk("42:1", " ") # single space — non-empty, JSON-invalid
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(
200, headers={"content-type": "text/event-stream"}, content=stream
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(MalformedSseData):
async for _ in stream_turn(client, "s1", "hi"):
pass
@respx.mock
async def test_malformed_data_truncation(self) -> None:
"""malformed_data_truncation [security]: 5000-char bad data → raw truncated to 200."""
from ratatoskr.sse_client import MalformedSseData
huge_bad = "x" * 5000 # not JSON; very long
stream = _sse_raw_chunk("42:1", huge_bad)
respx.post("https://w.example/sessions/s1/messages").mock(
return_value=httpx.Response(
200, headers={"content-type": "text/event-stream"}, content=stream
)
)
async with httpx.AsyncClient(base_url="https://w.example") as client:
with pytest.raises(MalformedSseData) as exc_info:
async for _ in stream_turn(client, "s1", "hi"):
pass
assert len(exc_info.value.raw) == 200
assert exc_info.value.raw == "x" * 200
# Exception message also only contains the truncated form
assert "x" * 5000 not in str(exc_info.value)
Reference in New Issue View Git Blame Copy Permalink