Testing

SSE Testing Guide: How to Test Server-Sent Events

HelpMeTest

HelpMeTest

— 5 min read
SSE Testing Guide: How to Test Server-Sent Events

Server-Sent Events (SSE) are a one-way, server-to-client streaming protocol built on HTTP. They power live dashboards, notification feeds, and AI chat streaming. Testing SSE is trickier than testing regular REST endpoints — the connection stays open, events arrive asynchronously, and you need to test reconnection behavior. Here's how.

How SSE Works

SSE uses a persistent HTTP connection where the server pushes text/event-stream formatted data:

id: 42
event: update
data: {"price": 150.25, "symbol": "AAPL"}

The client connects with EventSource and the server keeps the connection open. If the connection drops, the browser reconnects automatically using the Last-Event-ID header.

Testing the SSE Server Endpoint

For a FastAPI SSE endpoint:

# src/sse_router.py
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio, json

app = FastAPI()

@app.get("/events")
async def event_stream(topic: str = "default"):
    async def generator():
        for i in range(10):
            event = {"id": i, "count": i, "topic": topic}
            yield f"id: {i}\nevent: update\ndata: {json.dumps(event)}\n\n"
            await asyncio.sleep(0.1)
        yield "event: done\ndata: stream complete\n\n"
    
    return StreamingResponse(
        generator(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}
    )

Tests with pytest and TestClient:

# tests/test_sse_endpoint.py
import pytest, json
from fastapi.testclient import TestClient
from src.sse_router import app

client = TestClient(app)

class TestSSEEndpoint:
    def test_returns_correct_content_type(self):
        with client.stream("GET", "/events") as response:
            assert response.status_code == 200
            assert "text/event-stream" in response.headers["content-type"]

    def test_returns_cache_control_header(self):
        with client.stream("GET", "/events") as response:
            assert response.headers.get("cache-control") == "no-cache"

    def test_events_have_correct_structure(self):
        events = []
        with client.stream("GET", "/events") as response:
            for line in response.iter_lines():
                if line.startswith("data:"):
                    events.append(json.loads(line[5:].strip()))
                if len(events) >= 3:
                    break
        
        assert len(events) >= 3
        for event in events:
            assert "id" in event
            assert "count" in event
            assert "topic" in event

    def test_events_increment_sequentially(self):
        counts = []
        with client.stream("GET", "/events") as response:
            for line in response.iter_lines():
                if line.startswith("data:"):
                    counts.append(json.loads(line[5:].strip())["count"])
                if len(counts) >= 5:
                    break
        
        assert counts == list(range(5))

    def test_topic_filter_reflected_in_events(self):
        with client.stream("GET", "/events?topic=prices") as response:
            for line in response.iter_lines():
                if line.startswith("data:"):
                    data = json.loads(line[5:].strip())
                    assert data["topic"] == "prices"
                    break

    def test_event_ids_are_monotonically_increasing(self):
        event_ids = []
        with client.stream("GET", "/events") as response:
            for line in response.iter_lines():
                if line.startswith("id:"):
                    event_ids.append(int(line[3:].strip()))
                if len(event_ids) >= 3:
                    break
        
        assert event_ids == sorted(event_ids)

    def test_stream_ends_with_done_event(self):
        done_received = False
        with client.stream("GET", "/events") as response:
            for line in response.iter_lines():
                if line == "event: done":
                    done_received = True
                    break
        assert done_received

Async Testing

# tests/test_sse_async.py
import pytest, asyncio, json, httpx
from src.sse_router import app

@pytest.mark.asyncio
async def test_sse_async_client():
    async with httpx.AsyncClient(app=app, base_url="http://test") as c:
        async with c.stream("GET", "/events") as response:
            assert response.status_code == 200
            events = []
            async for line in response.aiter_lines():
                if line.startswith("data:"):
                    events.append(json.loads(line[5:].strip()))
                if len(events) >= 3:
                    break
            assert len(events) == 3

@pytest.mark.asyncio
async def test_multiple_concurrent_sse_connections():
    async def collect(client, topic, count):
        events = []
        async with client.stream("GET", f"/events?topic={topic}") as r:
            async for line in r.aiter_lines():
                if line.startswith("data:"):
                    events.append(json.loads(line[5:].strip()))
                if len(events) >= count:
                    break
        return events
    
    async with httpx.AsyncClient(app=app, base_url="http://test") as client:
        results = await asyncio.gather(
            collect(client, "stocks", 3),
            collect(client, "weather", 3),
            collect(client, "sports", 3),
        )
    
    assert results[0][0]["topic"] == "stocks"
    assert results[1][0]["topic"] == "weather"
    assert results[2][0]["topic"] == "sports"

Testing the EventSource Client (JavaScript)

Mock EventSource before importing your module:

// tests/sseClient.test.js
class MockEventSource {
  constructor(url) {
    this.url = url;
    this.readyState = 0;
    this.onopen = null;
    this.onmessage = null;
    this.onerror = null;
    MockEventSource._instances.push(this);
  }
  static _instances = [];
  static CONNECTING = 0;
  static OPEN = 1;
  static CLOSED = 2;
  
  simulateMessage(data) {
    const event = new MessageEvent('message', { data: JSON.stringify(data) });
    if (this.onmessage) this.onmessage(event);
  }
  simulateOpen() {
    this.readyState = 1;
    if (this.onopen) this.onopen(new Event('open'));
  }
  simulateError() {
    this.readyState = 2;
    if (this.onerror) this.onerror(new Event('error'));
  }
  close() { this.readyState = 2; }
  addEventListener() {}
  dispatchEvent() { return true; }
}
global.EventSource = MockEventSource;

const { createSSEConnection } = require('../src/sseClient');

describe('SSEClient', () => {
  beforeEach(() => { MockEventSource._instances = []; });

  test('creates EventSource with correct URL', () => {
    createSSEConnection('/api/events?topic=updates');
    expect(MockEventSource._instances[0].url).toBe('/api/events?topic=updates');
  });

  test('calls onMessage when event received', () => {
    const onMessage = jest.fn();
    createSSEConnection('/api/events', { onMessage });
    MockEventSource._instances[0].simulateOpen();
    MockEventSource._instances[0].simulateMessage({ text: 'Hello', count: 1 });
    expect(onMessage).toHaveBeenCalledWith(expect.objectContaining({ text: 'Hello' }));
  });

  test('calls onError when connection fails', () => {
    const onError = jest.fn();
    createSSEConnection('/api/events', { onError });
    MockEventSource._instances[0].simulateError();
    expect(onError).toHaveBeenCalled();
  });

  test('closes connection when close() called', () => {
    const connection = createSSEConnection('/api/events');
    connection.close();
    expect(MockEventSource._instances[0].readyState).toBe(MockEventSource.CLOSED);
  });

  test('accumulates events in order', () => {
    const received = [];
    createSSEConnection('/api/events', { onMessage: (d) => received.push(d) });
    const source = MockEventSource._instances[0];
    source.simulateOpen();
    source.simulateMessage({ seq: 1 });
    source.simulateMessage({ seq: 2 });
    source.simulateMessage({ seq: 3 });
    expect(received.map(e => e.seq)).toEqual([1, 2, 3]);
  });
});

Testing Reconnection Logic

describe('SSE Reconnection', () => {
  test('reconnects after connection error', () => {
    jest.useFakeTimers();
    const onReconnect = jest.fn();
    createSSEConnectionWithRetry('/api/events', { onReconnect, reconnectDelay: 3000 });
    
    MockEventSource._instances[0].simulateOpen();
    MockEventSource._instances[0].simulateError();
    jest.advanceTimersByTime(3100);
    
    expect(MockEventSource._instances).toHaveLength(2);
    expect(onReconnect).toHaveBeenCalledTimes(1);
    jest.useRealTimers();
  });

  test('stops reconnecting after max retries', () => {
    jest.useFakeTimers();
    const onGiveUp = jest.fn();
    createSSEConnectionWithRetry('/api/events', { maxRetries: 3, reconnectDelay: 100, onGiveUp });
    
    for (let i = 0; i < 4; i++) {
      MockEventSource._instances[i].simulateError();
      jest.advanceTimersByTime(200);
    }
    
    expect(MockEventSource._instances).toHaveLength(4);
    expect(onGiveUp).toHaveBeenCalledTimes(1);
    jest.useRealTimers();
  });
});

Testing SSE Event Parsing

# tests/test_sse_parser.py
import pytest
from src.sse_parser import parse_sse_event

class TestSSEParser:
    def test_parses_simple_data_event(self):
        event = parse_sse_event("data: hello world\n\n")
        assert event.data == "hello world"
        assert event.event_type == "message"

    def test_parses_typed_event(self):
        event = parse_sse_event("event: price-update\ndata: 150.25\n\n")
        assert event.event_type == "price-update"
        assert event.data == "150.25"

    def test_parses_event_with_id(self):
        event = parse_sse_event("id: 42\ndata: some data\n\n")
        assert event.id == "42"

    def test_parses_multiline_data(self):
        event = parse_sse_event("data: line 1\ndata: line 2\ndata: line 3\n\n")
        assert event.data == "line 1\nline 2\nline 3"

    def test_ignores_comment_lines(self):
        event = parse_sse_event(": this is a comment\ndata: actual data\n\n")
        assert event.data == "actual data"

    def test_parses_retry_field(self):
        event = parse_sse_event("retry: 5000\ndata: reconnect hint\n\n")
        assert event.retry_ms == 5000

End-to-End SSE Testing with HelpMeTest

Browser-level SSE testing validates the complete user experience — that real-time updates actually appear in the UI.

HelpMeTest can test SSE-powered UIs:

*** Test Cases ***
Live Dashboard Updates In Real-Time
    Go To    https://your-app.com/dashboard
    Wait Until Element Is Visible    .live-indicator    timeout=5s
    ${initial_value}=    Get Text    .live-counter
    Sleep    3s
    ${updated_value}=    Get Text    .live-counter
    Should Not Be Equal    ${initial_value}    ${updated_value}

SSE Connection Status Shows Connected
    Go To    https://your-app.com/dashboard
    Wait Until Element Contains    .connection-status    Connected    timeout=10s

Summary

  • Test content-type — text/event-stream is required for browsers to interpret SSE
  • Test event format — id:, event:, data: parsing is a common bug source
  • Test sequential ordering — events must arrive in order with monotonically increasing IDs
  • Mock EventSource in frontend unit tests — real SSE requires a live server
  • Test reconnection — verify retry caps and Last-Event-ID is preserved
  • Use end-to-end tests to verify UI updates in real-time, not just that events are sent

Read more