vh
59602fe3ff
DEC-4 live smoke PASSED first (personal :8081, b127/b128): create → streamed turn that rendered (worker_phase/text/text_boundary/done with usage) → SIGINT cancel that round-tripped to a cancelled terminal. With both CLI + web on the adapter, the hand-rolled turn-stream family is fully orphaned — deleting it now. - sse_client.py (714 → 224): removed stream_turn / reconnect_turn / stream_turn_resilient / cancel_turn + the Event dataclasses (Text/Done/…/Event union) + CancelResult + the SSE parse helpers (_iter_events / _envelope_for_type / _parse_sse_id / _eager_failure_fields / _INT_RE). KEPT: the caller-semantic exceptions (the adapter raises them, DEC-2), SseId, AdminEvent, stream_admin_events (slice-6 admin surface). - sessions.py (677 → 608): removed list_sessions + get_session_tools (no surface users) + SessionPage. KEPT: create_session / get_session_messages (the --seed-first-message probe still uses them, slice-3) + all exceptions + SessionInfo. - tests: test_sse_client pruned to TestStreamAdminEvents; test_sessions dropped the list_sessions + get_session_tools classes. The deleted turn-stream behavior is now covered by test_wt.py + the CLI/web integration tests + the live smoke. Suite 490 green (570 − 80 deleted turn-stream tests); ruff clean on all touched files; no new mypy errors. Patch (internal cleanup; behavior preserved).
95 lines
4.2 KiB
Python
95 lines
4.2 KiB
Python
"""Tests for ratatoskr.sse_client — the admin-events stream (#11).
|
|
|
|
The turn-stream + Event-model tests retired with the worldtree-sdk cutover (#20);
|
|
the turn path is now covered by tests/test_wt.py + the CLI/web integration tests.
|
|
This module keeps the still-hand-rolled admin-events surface (slice-6)."""
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
|
|
from ratatoskr.sse_client import (
|
|
AdminEvent,
|
|
SseConnectFailed,
|
|
stream_admin_events,
|
|
)
|
|
|
|
|
|
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 TestStreamAdminEvents:
|
|
"""docs/conversation-api-spec.md § Admin Event Stream — stream_admin_events (#11)."""
|
|
|
|
@respx.mock
|
|
async def test_happy_multi_event_admin_bearer(self) -> None:
|
|
"""happy [happy,tracer]: yields AdminEvent envelopes; request uses the ADMIN bearer."""
|
|
env1 = {
|
|
"id": 41, "type": "session.created", "timestamp": "2026-05-06T10:00:00.000Z",
|
|
"data": {"session_id": "s1", "agent_id": "mimir", "user_id": None},
|
|
}
|
|
env2 = {
|
|
"id": 42, "type": "turn.started", "timestamp": "2026-05-06T10:00:01.000Z",
|
|
"data": {"session_id": "s1", "turn_id": 7, "agent_id": "mimir", "user_id": None},
|
|
}
|
|
stream = _sse_chunk("41", env1) + _sse_chunk("42", env2)
|
|
route = respx.get("https://w.example/admin/events").mock(
|
|
return_value=httpx.Response(
|
|
200, headers={"content-type": "text/event-stream"}, content=stream
|
|
)
|
|
)
|
|
async with httpx.AsyncClient(
|
|
base_url="https://w.example", headers={"Authorization": "Bearer consumer"}
|
|
) as client:
|
|
events = [e async for e in stream_admin_events(client, admin_key="admin-xyz")]
|
|
assert [e.type for e in events] == ["session.created", "turn.started"]
|
|
assert isinstance(events[0], AdminEvent)
|
|
assert events[0].id == 41
|
|
assert events[1].data["turn_id"] == 7
|
|
assert route.calls[0].request.headers["Authorization"] == "Bearer admin-xyz"
|
|
|
|
@respx.mock
|
|
async def test_last_event_id_header(self) -> None:
|
|
"""last_event_id_header [trace]: empty stream → []; Last-Event-ID header sent."""
|
|
route = respx.get("https://w.example/admin/events").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:
|
|
events = [e async for e in stream_admin_events(client, admin_key="k", last_event_id=99)]
|
|
assert events == []
|
|
assert route.calls[0].request.headers["Last-Event-ID"] == "99"
|
|
|
|
@respx.mock
|
|
async def test_403_scope_denied(self) -> None:
|
|
"""403 [error]: key lacks admin.events.read → SseConnectFailed(403)."""
|
|
respx.get("https://w.example/admin/events").mock(
|
|
return_value=httpx.Response(403, json={"error_code": "auth_scope_denied"})
|
|
)
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
with pytest.raises(SseConnectFailed) as exc:
|
|
_ = [e async for e in stream_admin_events(client, admin_key="k")]
|
|
assert exc.value.status == 403
|
|
|
|
@respx.mock
|
|
async def test_skips_malformed_frame(self) -> None:
|
|
"""skips_malformed [adversarial]: a bad-JSON frame is skipped, not fatal."""
|
|
good = _sse_chunk("41", {"id": 41, "type": "session.created", "data": {"session_id": "s1"}})
|
|
bad = b"id: 42\ndata: not-json\n\n"
|
|
good2 = _sse_chunk(
|
|
"43", {"id": 43, "type": "session.deleted", "data": {"session_id": "s1"}}
|
|
)
|
|
respx.get("https://w.example/admin/events").mock(
|
|
return_value=httpx.Response(
|
|
200, headers={"content-type": "text/event-stream"}, content=good + bad + good2
|
|
)
|
|
)
|
|
async with httpx.AsyncClient(base_url="https://w.example") as client:
|
|
events = [e async for e in stream_admin_events(client, admin_key="k")]
|
|
assert [e.type for e in events] == ["session.created", "session.deleted"]
|