vh
c17af18351
Volva's code-spec review (thread 01KS4CP6ZZ1F) surfaced four code-vs-
contract drift findings on the TDD-passing implementation. All four
addressed here; no contract amendments required.
1. _iter_events fell off the end of aiter_sse() normally on clean EOF
before any Done/Error/Cancelled. Per INV-001 the iterator MUST NOT
raise StopAsyncIteration before a terminal event unless the HTTP
connection drops, in which case it raises SseConnectionDropped.
Clean EOF before terminal is the same semantic — the stream ended
without delivering its contracted invariant. Fix: track terminal_seen
inside _iter_events; after the async-for completes, if not seen,
raise SseConnectionDropped(last_seen_sse_id=...). Two new tests:
test_clean_eof_before_terminal (one text then EOF) and
test_zero_event_eof (empty stream — last_seen_sse_id is None).
2. SseConnectFailed and CancelFailed both store .body without
truncation; ERROR_ROUTING specifies resp.read()[:1024]. Fix
truncates in each exception's __init__ before storing. New test
test_connect_failed_body_truncated (503 + 5000-byte body → 1024)
and test_cancel_failed_truncates_body (same shape on cancel).
3. _parse_sse_id PRE-001 specifies `assert isinstance(raw, str)`.
Previous code called raw.split(":") directly, which raises an
incidental AttributeError on non-str inputs — not the contracted
precondition path. Fix adds the assert. New test
test_non_string_input covers int and None.
4. Cancel ERROR_ROUTING said httpx.HTTPStatusError other status →
CancelFailed, but no test exercised the branch. test_cancel_failed_
truncates_body covers this (above) — single test double-covers
findings 2 and 4.
43 tests GREEN (42 sse_client + boundary smoke); ruff clean.
Meta-note from Volva: TDD caught the main happy/adversarial SSE shape,
resume header/body, turn-id flip, and cancel races. The remaining
misses were "negative space" cases (clean premature EOF, exception
payload truncation, untested generic cancel branch). Calibration
evidence that cross-model review pulls weight on what same-model
TDD's hypothesis-space doesn't probe.
697 lines
28 KiB
Python
697 lines
28 KiB
Python
"""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)
|