test_sse_parser.py (5367B)
1 """T026 — Tests for the SSE (Server-Sent Events) parser.""" 2 3 import pytest 4 from app.sse_parser import parse_sse_events 5 6 7 # ── helpers ────────────────────────────────────────────────────────────── 8 9 def _sse(data: str) -> list[dict]: 10 """Parse an SSE text block into event dicts.""" 11 return list(parse_sse_events(data)) 12 13 14 # ── tests: each SSE event type ────────────────────────────────────────── 15 16 17 def test_parse_run_started() -> None: 18 raw = 'event: run.started\ndata: {"run_id":"r1"}\n\n' 19 events = _sse(raw) 20 assert events == [{"event": "run.started", "data": {"run_id": "r1"}}] 21 22 23 def test_parse_message_started() -> None: 24 raw = 'event: message.started\ndata: {"message_id":"m1"}\n\n' 25 events = _sse(raw) 26 assert events == [{"event": "message.started", "data": {"message_id": "m1"}}] 27 28 29 def test_parse_assistant_delta() -> None: 30 raw = 'event: assistant.delta\ndata: {"delta":"Hello"}\n\n' 31 events = _sse(raw) 32 assert events == [{"event": "assistant.delta", "data": {"delta": "Hello"}}] 33 34 35 def test_parse_tool_started() -> None: 36 raw = 'event: tool.started\ndata: {"tool_id":"t1"}\n\n' 37 events = _sse(raw) 38 assert events == [{"event": "tool.started", "data": {"tool_id": "t1"}}] 39 40 41 def test_parse_tool_completed() -> None: 42 raw = 'event: tool.completed\ndata: {"tool_id":"t1"}\n\n' 43 events = _sse(raw) 44 assert events == [{"event": "tool.completed", "data": {"tool_id": "t1"}}] 45 46 47 def test_parse_tool_failed() -> None: 48 raw = 'event: tool.failed\ndata: {"error":"oops"}\n\n' 49 events = _sse(raw) 50 assert events == [{"event": "tool.failed", "data": {"error": "oops"}}] 51 52 53 def test_parse_tool_progress() -> None: 54 raw = 'event: tool.progress\ndata: {"progress":50}\n\n' 55 events = _sse(raw) 56 assert events == [{"event": "tool.progress", "data": {"progress": 50}}] 57 58 59 def test_parse_assistant_completed() -> None: 60 raw = 'event: assistant.completed\ndata: {"message_id":"m1"}\n\n' 61 events = _sse(raw) 62 assert events == [{"event": "assistant.completed", "data": {"message_id": "m1"}}] 63 64 65 def test_parse_run_completed() -> None: 66 raw = 'event: run.completed\ndata: {"run_id":"r1"}\n\n' 67 events = _sse(raw) 68 assert events == [{"event": "run.completed", "data": {"run_id": "r1"}}] 69 70 71 # ── tests: edge cases ─────────────────────────────────────────────────── 72 73 74 def test_handles_malformed_json_data() -> None: 75 """Malformed JSON data lines are skipped gracefully.""" 76 raw = 'event: run.started\ndata: not-json\n\n' 77 events = _sse(raw) 78 assert events == [] 79 80 81 def test_handles_empty_data_line() -> None: 82 """Empty data lines produce no events.""" 83 raw = "data: \n\n" 84 events = _sse(raw) 85 assert events == [] 86 87 88 def test_skips_keepalive_comments() -> None: 89 """Lines starting with ':' are SSE keepalive comments — skip them.""" 90 raw = ( 91 ": keepalive\n\n" 92 'event: run.started\ndata: {"run_id":"r1"}\n\n' 93 ) 94 events = _sse(raw) 95 assert len(events) == 1 96 assert events[0]["event"] == "run.started" 97 98 99 def test_multiple_events_in_stream() -> None: 100 """Multiple events in one block are parsed in order.""" 101 raw = ( 102 'event: run.started\ndata: {"run_id":"r1"}\n\n' 103 'event: assistant.delta\ndata: {"delta":"Hi"}\n\n' 104 'event: run.completed\ndata: {"run_id":"r1"}\n\n' 105 ) 106 events = _sse(raw) 107 assert [e["event"] for e in events] == [ 108 "run.started", "assistant.delta", "run.completed" 109 ] 110 111 112 def test_event_type_unknown_still_parsed() -> None: 113 """Unknown event types are still returned (don't filter).""" 114 raw = 'event: mystery.event\ndata: {"x":1}\n\n' 115 events = _sse(raw) 116 assert events == [{"event": "mystery.event", "data": {"x": 1}}] 117 118 class TestSSEParserEdgeCases: 119 """Edge cases for SSE event parsing.""" 120 121 def test_empty_stream(self) -> None: 122 from app.sse_parser import parse_sse_events 123 assert list(parse_sse_events("")) == [] 124 125 def test_keepalive_comment_only(self) -> None: 126 from app.sse_parser import parse_sse_events 127 assert list(parse_sse_events(": keepalive\n\n")) == [] 128 129 def test_multiple_events_in_stream(self) -> None: 130 from app.sse_parser import parse_sse_events 131 stream = 'event: run.started\ndata: {"run_id": "r1"}\n\nevent: run.completed\ndata: {"session_id": "s1"}\n\n' 132 events = list(parse_sse_events(stream)) 133 assert len(events) == 2 134 assert events[0]["event"] == "run.started" 135 assert events[1]["event"] == "run.completed" 136 137 def test_malformed_json_skips_event(self) -> None: 138 from app.sse_parser import parse_sse_events 139 assert list(parse_sse_events("event: test\ndata: not-json\n\n")) == [] 140 141 def test_data_only_skipped(self) -> None: 142 from app.sse_parser import parse_sse_events 143 assert list(parse_sse_events('data: {"key": "val"}\n\n')) == [] 144 145 def test_event_type_without_data_skipped(self) -> None: 146 from app.sse_parser import parse_sse_events 147 assert list(parse_sse_events("event: test\n\n")) == []