Mock streaming LLM responses is the only way to get fast, deterministic CI for any app that consumes token-by-token completions. Real inference endpoints add latency, rate limits, and non-determinism that break unit tests; you need to fake the SSE protocol locally. This guide walks through building reusable mocks in Python and TypeScript that match the OpenAI streaming contract.
Step 1: Understand the streaming contract you must emulate
OpenAI-compatible endpoints send Server-Sent Events over HTTP. Each event is a data: line containing a JSON-serialized chat.completion.chunk object, terminated by a blank line. The stream ends with data: [DONE].
A minimal raw frame looks like this:
data: {"id":"chatcmpl-abc","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"Hi"},"finish_reason":null}]}
data: [DONE]If you proxy through a gateway like n4n.ai that adds automatic fallback across providers, your mocked tests should still simulate this exact raw shape so the client parsing code stays identical. The client does not care which backend produced the bytes.
Key fields your mock must include:
choices[0].delta.content(ortool_calls)choices[0].finish_reasonon the final chunk (usually"stop")idandobjectfor schema compatibility
Missing finish_reason will cause some SDKs to hang waiting for the [DONE] sentinel.
Step 2: Build a chunk generator in Python
Start by splitting a known string into deltas. Keep it pure and synchronous; you can adapt to async later.
import json
def make_chunks(text: str, size: int = 4):
"""Yield OpenAI-compatible chunk dicts for a given string."""
chunk_id = "chatcmpl-test"
for i in range(0, len(text), size):
piece = text[i:i+size]
yield {
"id": chunk_id,
"object": "chat.completion.chunk",
"choices": [
{
"index": 0,
"delta": {"content": piece},
"finish_reason": None,
}
],
}
# final chunk carries finish_reason, empty delta
yield {
"id": chunk_id,
"object": "chat.completion.chunk",
"choices": [
{"index": 0, "delta": {}, "finish_reason": "stop"}
],
}
def serialize_sse(chunks):
"""Turn chunk dicts into SSE byte frames."""
for ch in chunks:
yield f"data: {json.dumps(ch)}\n\n".encode("utf-8")
yield b"data: [DONE]\n\n"This gives you a deterministic stream. Use size=1 to simulate slow token emission, or size=len(text) to simulate a single shot.
Step 3: Serve the mock stream over HTTP for integration tests
If you need to test the actual HTTP client (e.g., the OpenAI Python SDK, or your own httpx wrapper), stand up an ASGI app in-process. Starlette is the lightest dependency that gets this right.
from starlette.applications import Starlette
from starlette.responses import StreamingResponse
from starlette.routing import Route
from starlette.testclient import TestClient
def make_app(text: str):
async def stream(request):
return StreamingResponse(
serialize_sse(make_chunks(text)),
media_type="text/event-stream",
)
return Starlette(routes=[Route("/v1/chat/completions", stream, methods=["POST"])])
def test_openai_client_stream():
from openai import OpenAI
app = make_app("Hello world, this is a mock.")
with TestClient(app) as client:
# Point the SDK at the in-process server
openai = OpenAI(base_url="http://testserver/v1", api_key="fake")
resp = openai.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "hi"}],
stream=True,
)
collected = "".join(c.choices[0].delta.content or "" for c in resp)
assert collected == "Hello world, this is a mock."The TestClient runs the ASGI app in the same event loop, so no sockets are opened. This is the correct way to mock streaming LLM responses for integration-level tests without touching the network.
Step 4: Mock at the SDK level for fast unit tests
Spinning up even an in-process server is overkill when you only want to test your prompt builder or response aggregator. Patch the client method and return an iterator of fake objects.
from types import SimpleNamespace
class FakeDelta(SimpleNamespace):
pass
class FakeChoice(SimpleNamespace):
pass
class FakeChunk(SimpleNamespace):
pass
def fake_stream_obj(text, size=4):
for i in range(0, len(text), size):
yield FakeChunk(
choices=[FakeChoice(delta=FakeDelta(content=text[i:i+size]))]
)
yield FakeChunk(choices=[FakeChoice(delta=FakeDelta(content=None), finish_reason="stop")])
def test_aggregator():
from myapp.logic import aggregate
result = aggregate(fake_stream_obj("deterministic output"))
assert result == "deterministic output"This runs in microseconds and isolates your business logic from transport concerns. Use this for the bulk of your suite; reserve Step 3 for one or two smoke tests.
Step 5: Do the same in TypeScript
Node’s http module is enough to emit SSE. Below is a minimal server you can launch in a beforeAll hook in Vitest.
import http from 'node:http';
import { AddressInfo } from 'node:net';
function makeChunks(text: string, size = 4) {
const chunks = [];
for (let i = 0; i < text.length; i += size) {
chunks.push({
id: 'chatcmpl-ts',
object: 'chat.completion.chunk',
choices: [{ index: 0, delta: { content: text.slice(i, i + size) }, finish_reason: null }],
});
}
chunks.push({
id: 'chatcmpl-ts',
object: 'chat.completion.chunk',
choices: [{ index: 0, delta: {}, finish_reason: 'stop' }],
});
return chunks;
}
export function startMockServer(text: string) {
const server = http.createServer((req, res) => {
if (req.url?.includes('/v1/chat/completions')) {
res.writeHead(200, { 'Content-Type': 'text/event-stream' });
for (const c of makeChunks(text)) {
res.write(`data: ${JSON.stringify(c)}\n\n`);
}
res.write('data: [DONE]\n\n');
res.end();
}
});
return new Promise<{ url: string; close: () => void }>((resolve) => {
server.listen(0, () => {
const port = (server.address() as AddressInfo).port;
resolve({ url: `http://localhost:${port}`, close: () => server.close() });
});
});
}Consume it with the OpenAI Node SDK:
import OpenAI from 'openai';
import { startMockServer } from './mock';
test('streams tokens', async () => {
const mock = await startMockServer('TypeScript mock stream');
const client = new OpenAI({ baseURL: mock.url + '/v1', apiKey: 'fake' });
const stream = await client.chat.completions.create({
model: 'gpt-4o-mini',
messages: [{ role: 'user', content: 'hi' }],
stream: true,
});
let text = '';
for await (const part of stream) {
text += part.choices[0]?.delta?.content ?? '';
}
expect(text).toBe('TypeScript mock stream');
mock.close();
});For unit-level tests, skip the server and inject a fake async iterable directly into your parser.
Step 6: Verify success and wire into CI
A mock is only useful if it fails loudly when the contract changes. Assert three things:
- Reassembled text equals the fixture string.
- Chunk count matches expectation (e.g.,
Math.ceil(len/size) + 1for the finalfinish_reasonchunk). - No partial UTF-8 sequences when splitting on byte boundaries—Python’s
sizeonstris fine for ASCII, but usetext.encode()slicing if you support multibyte.
Run the suite headless:
pytest tests/test_stream_mocks.py -q
vitest run src/__tests__/stream.mock.test.tsIn GitHub Actions, set PYTHONUNBUFFERED=1 and use npm ci to keep installs deterministic. Because the mocks never hit the network, these tests should complete in under a second and never flake due to provider degradation.
If you also run a weekly integration test against a live gateway, keep it in a separate job marked continue-on-error. That way a provider outage doesn’t block your merge queue, but you still learn about breaking API changes.
Caveats worth noting
- Some SDKs buffer SSE and parse
usagemetadata sent in the final chunk. If your app readsusage, add"usage": {"total_tokens": 10}to the last chunk. - Tool-call streaming uses
delta.tool_callswith incrementalfunction.argumentsstrings. Mirror that shape in your generator if you test agents. - Don’t assert on
iduniqueness across tests; it’s irrelevant to correctness.
Mock streaming LLM responses correctly and your CI stays green when the provider is down, the bill stays low, and your parser gets exercised on every commit.