Running integration tests against a live model endpoint burns tokens and flakes on rate limits. A fake llm server ci setup gives you a local OpenAI-compatible stub that returns canned responses, so your test suite stays fast and deterministic. This tutorial builds one with FastAPI and wires it into pytest.
Prerequisites
- Python 3.11+ installed locally
pip install fastapi uvicorn httpx pytest pytest-asyncio openai- A project that already calls the OpenAI Chat Completions API
- Environment variable
OPENAI_API_KEYis irrelevant for the stub but keep your client code unchanged
Step 1: Scaffold the fake server
Create fake_llm.py. The server implements the minimal chat completions contract: accept a JSON body, return a completion with id, object, choices, and usage. We echo the last user message so tests can assert round-trips.
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, StreamingResponse
import json
import asyncio
app = FastAPI()
def _last_user(messages):
return next((m["content"] for m in reversed(messages) if m["role"] == "user"), "")
@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
body = await request.json()
model = body.get("model", "gpt-3.5-turbo")
messages = body.get("messages", [])
last_user = _last_user(messages)
stream = body.get("stream", False)
if not stream:
return JSONResponse({
"id": "chatcmpl-fake",
"object": "chat.completion",
"created": 0,
"model": model,
"choices": [{
"index": 0,
"message": {"role": "assistant", "content": f"Echo: {last_user}"},
"finish_reason": "stop"
}],
"usage": {"prompt_tokens": 5, "completion_tokens": 5, "total_tokens": 10}
})
async def event_gen():
words = f"Echo: {last_user}".split()
for i, word in enumerate(words):
chunk = {
"id": "chatcmpl-fake",
"object": "chat.completion.chunk",
"created": 0,
"model": model,
"choices": [{
"index": 0,
"delta": {"content": word + " "},
"finish_reason": "stop" if i == len(words) - 1 else None
}]
}
yield f"data: {json.dumps(chunk)}\n\n"
await asyncio.sleep(0.01)
yield "data: [DONE]\n\n"
return StreamingResponse(event_gen(), media_type="text/event-stream")Start it:
uvicorn fake_llm:app --port 8000Verify with curl:
curl -s http://localhost:8000/v1/chat/completions \
-H "content-type: application/json" \
-d '{"model":"gpt-3.5-turbo","messages":[{"role":"user","content":"ping"}]}'Expected output (formatted for clarity):
{
"id": "chatcmpl-fake",
"object": "chat.completion",
"created": 0,
"model": "gpt-3.5-turbo",
"choices": [
{
"index": 0,
"message": {"role": "assistant", "content": "Echo: ping"},
"finish_reason": "stop"
}
],
"usage": {"prompt_tokens": 5, "completion_tokens": 5, "total_tokens": 10}
}Step 2: Point your client at the stub
The OpenAI Python client reads base_url. In your test configuration, override it via environment variable so production code stays clean.
import os
from openai import OpenAI
client = OpenAI(
base_url=os.getenv("LLM_BASE_URL", "https://api.openai.com/v1"),
api_key=os.getenv("OPENAI_API_KEY", "fake")
)
resp = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": "hello"}]
)
print(resp.choices[0].message.content)With LLM_BASE_URL=http://localhost:8000/v1 this prints Echo: hello. The same binary works against a real provider when the variable is unset.
Step 3: Test streaming without a socket
Use the stream=True flag. The fake llm server ci already yields SSE; confirm the client parses it.
stream = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": "hi"}],
stream=True
)
collected = ""
for piece in stream:
if piece.choices[0].delta.content:
collected += piece.choices[0].delta.content
print(collected.strip())Output: Echo: hi. Latency injection via asyncio.sleep lets you exercise timeout logic.
Step 4: Inject faults and latency
CI must prove your retry and fallback paths. Extend the route to read headers.
Add near the top of chat_completions:
fake_status = request.headers.get("x-fake-status")
if fake_status:
return JSONResponse(status_code=int(fake_status), content={"error": "injected fault"})
fake_delay = request.headers.get("x-fake-delay")
if fake_delay:
await asyncio.sleep(float(fake_delay))Now a test can force a 429:
try:
client.chat.completions.create(
model="gpt-4",
messages=[{"role":"user","content":"x"}],
headers={"x-fake-status": "429"}
)
except Exception as e:
print(type(e).__name__) # RateLimitErrorYour circuit breaker should catch that.
Step 5: In-process pytest fixture
Running uvicorn in CI is fine, but httpx.ASGITransport removes the network entirely. Create conftest.py:
import pytest
from httpx import AsyncClient, ASGITransport
from fake_llm import app
@pytest.fixture
async def fake_client():
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as ac:
yield acThen a test file:
@pytest.mark.asyncio
async def test_echo(fake_client):
r = await fake_client.post("/v1/chat/completions",
json={"model":"x","messages":[{"role":"user","content":"hi"}]})
assert r.status_code == 200
assert "Echo: hi" in r.json()["choices"][0]["message"]["content"]
@pytest.mark.asyncio
async def test_fault(fake_client):
r = await fake_client.post("/v1/chat/completions",
json={"model":"x","messages":[]},
headers={"x-fake-status":"503"})
assert r.status_code == 503Run pytest -q. Both pass in milliseconds.
Step 6: GitHub Actions wiring
Because the fake llm server ci runs in-process, the workflow is just Python setup and pytest.
name: ci
on: [push]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.12"
- run: pip install fastapi uvicorn httpx pytest pytest-asyncio openai
- run: pytest -qNo service containers, no secrets.
Going further
For multi-model routing tests, branch on the model field to return different payloads. When you later point the same client at a real OpenAI-compatible gateway such as n4n.ai, the contract holds and you gain per-token metering without rewriting tests. Keep the stub in your repo; it doubles as a local dev target when the real API is down.