31 lines
949 B
Python
31 lines
949 B
Python
"""Tests for SSE frame formatting utilities."""
|
|
|
|
import json
|
|
|
|
|
|
def _format_sse(event: str, data, *, event_id: str | None = None) -> str:
|
|
from app.gateway.services import format_sse
|
|
|
|
return format_sse(event, data, event_id=event_id)
|
|
|
|
|
|
def test_sse_end_event_data_null():
|
|
"""End event should have data: null."""
|
|
frame = _format_sse("end", None)
|
|
assert "data: null" in frame
|
|
|
|
|
|
def test_sse_metadata_event():
|
|
"""Metadata event should include run_id and attempt."""
|
|
frame = _format_sse("metadata", {"run_id": "abc", "attempt": 1}, event_id="123-0")
|
|
assert "event: metadata" in frame
|
|
assert "id: 123-0" in frame
|
|
|
|
|
|
def test_sse_error_format():
|
|
"""Error event should use message/name format."""
|
|
frame = _format_sse("error", {"message": "boom", "name": "ValueError"})
|
|
parsed = json.loads(frame.split("data: ")[1].split("\n")[0])
|
|
assert parsed["message"] == "boom"
|
|
assert parsed["name"] == "ValueError"
|