Simulating rate limits and errors in a local LLM mock

You can’t reliably test retry backoff or fallback logic against a live LLM provider—their rate limits are unpredictable and hitting them costs money. To simulate rate limits local llm mock servers let you reproduce 429s, 500s, and timeouts on demand, so your client behaves correctly before production. This guide walks through building one with FastAPI and wiring it into a test suite.

Step 1: Stand up a minimal OpenAI-compatible mock

Start with a server that speaks the same shape as the OpenAI completions endpoint. Most LLM gateways, including n4n.ai, expose an OpenAI-compatible route, so a mock that matches the contract lets you swap the base URL without changing client code.

We’ll use FastAPI because its async model makes latency and concurrency simulation straightforward.

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import uvicorn

app = FastAPI()

@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    body = await request.json()
    return {
        "id": "mock-1",
        "object": "chat.completion",
        "model": body.get("model", "gpt-3.5-turbo"),
        "choices": [{
            "index": 0,
            "message": {"role": "assistant", "content": "echo"},
            "finish_reason": "stop"
        }],
        "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}
    }

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8080)

Save as mock.py and run python mock.py. Verify with a curl:

curl -X POST http://127.0.0.1:8080/v1/chat/completions \
  -H "content-type: application/json" \
  -d '{"model":"gpt-3.5-turbo","messages":[{"role":"user","content":"hi"}]}'

You should get a JSON completion. That’s the baseline; now we break it on purpose.

Step 2: Add deterministic rate limit simulation

To simulate rate limits local llm mock must track request counts per identity and reject excess with a proper 429 and Retry-After. A sliding window per API key is enough for local testing—no need for Redis.

from datetime import datetime, timedelta
from collections import defaultdict

RATE_LIMIT = 5
WINDOW = timedelta(seconds=10)
hits = defaultdict(list)

@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    api_key = request.headers.get("authorization", "anon")
    now = datetime.utcnow()
    hits[api_key] = [t for t in hits[api_key] if now - t < WINDOW]
    if len(hits[api_key]) >= RATE_LIMIT:
        retry_after = int((hits[api_key][0] + WINDOW - now).total_seconds())
        return JSONResponse(
            status_code=429,
            content={"error": {"message": "Rate limit exceeded", "type": "rate_limit_error"}},
            headers={"Retry-After": str(max(retry_after, 1))}
        )
    hits[api_key].append(now)
    body = await request.json()
    return {
        "id": "mock-1",
        "object": "chat.completion",
        "model": body.get("model", "gpt-3.5-turbo"),
        "choices": [{
            "index": 0,
            "message": {"role": "assistant", "content": "echo"},
            "finish_reason": "stop"
        }],
        "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}
    }

Push RATE_LIMIT and WINDOW into environment variables so tests can tighten limits without code edits. A low limit like 3 requests per 5 seconds makes test suites fast.

Verify the limit by looping requests:

import requests
url = "http://127.0.0.1:8080/v1/chat/completions"
headers = {"authorization": "Bearer test", "content-type": "application/json"}
for i in range(3):
    assert requests.post(url, json={"model":"x","messages":[]}, headers=headers).status_code == 200
r = requests.post(url, json={"model":"x","messages":[]}, headers=headers)
assert r.status_code == 429 and "Retry-After" in r.headers

Step 3: Inject errors and latency via headers

When you simulate rate limits local llm mock, you also want to trigger 500s, 503s, and slow responses without changing server state. Header-driven fault injection keeps the mock stateless and reproducible.

Extend the handler:

import asyncio
from fastapi.responses import StreamingResponse

@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    api_key = request.headers.get("authorization", "anon")
    now = datetime.utcnow()
    hits[api_key] = [t for t in hits[api_key] if now - t < WINDOW]
    if len(hits[api_key]) >= RATE_LIMIT:
        retry_after = int((hits[api_key][0] + WINDOW - now).total_seconds())
        return JSONResponse(
            status_code=429,
            content={"error": {"message": "Rate limit exceeded", "type": "rate_limit_error"}},
            headers={"Retry-After": str(max(retry_after, 1))}
        )
    hits[api_key].append(now)

    mock_error = request.headers.get("x-mock-error")
    if mock_error:
        code = int(mock_error)
        return JSONResponse(status_code=code, content={"error": {"message": f"mock {code}"}})

    mock_latency = int(request.headers.get("x-mock-latency", 0))
    if mock_latency:
        await asyncio.sleep(mock_latency / 1000)

    if request.headers.get("x-mock-stream") == "true":
        async def gen():
            yield b'data: {"choices":[{"delta":{"content":"a"}}]}\n\n'
            if request.headers.get("x-mock-stream-fail") == "true":
                return
            yield b'data: {"choices":[{"delta":{"content":"b"}}]}\n\n'
            yield b'data: [DONE]\n\n'
        return StreamingResponse(gen(), media_type="text/event-stream")

    body = await request.json()
    return {
        "id": "mock-1",
        "object": "chat.completion",
        "model": body.get("model", "gpt-3.5-turbo"),
        "choices": [{
            "index": 0,
            "message": {"role": "assistant", "content": "echo"},
            "finish_reason": "stop"
        }],
        "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}
    }

For streaming endpoints, x-mock-stream-fail sends a few tokens then closes the connection mid-stream. That exercises client-side stream recovery. A client can force a 503 with headers={"x-mock-error":"503"} or a 2-second delay with x-mock-latency: 2000.

Step 4: Drive the mock from a real LLM client

Point the OpenAI Python SDK at the mock by setting base_url. This validates that your production client code—including any middleware—handles failures correctly.

from openai import OpenAI

client = OpenAI(base_url="http://127.0.0.1:8080/v1", api_key="test")

def complete_with_retry(max_retries=3):
    for attempt in range(max_retries):
        try:
            resp = client.chat.completions.create(
                model="gpt-3.5-turbo",
                messages=[{"role":"user","content":"hi"}],
                extra_headers={"x-mock-error": "429"} if attempt == 0 else {}
            )
            return resp.choices[0].message.content
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            import time
            time.sleep(2 ** attempt)

If you use the SDK’s built-in max_retries, set it to 0 in tests and implement your own backoff, or let the SDK handle it and assert on final success. The key is that the mock supplies the failure; the client supplies the resilience.

Step 5: Write automated tests that prove resilience

The point of simulate rate limits local llm mock is to fail in CI if retry logic regresses. Use pytest with an httpx fixture so you control headers precisely.

import pytest, asyncio, httpx

@pytest.fixture
def mock_url():
    return "http://127.0.0.1:8080/v1/chat/completions"

def test_rate_limit_then_success(mock_url):
    headers = {"authorization":"Bearer t", "content-type":"application/json"}
    # assume RATE_LIMIT=2, WINDOW=1s set via env
    for _ in range(2):
        assert httpx.post(mock_url, json={"model":"x","messages":[]}, headers=headers).status_code == 200
    r = httpx.post(mock_url, json={"model":"x","messages":[]}, headers=headers)
    assert r.status_code == 429
    asyncio.sleep(1)
    assert httpx.post(mock_url, json={"model":"x","messages":[]}, headers=headers).status_code == 200

def test_500_triggers_retry(mock_url):
    headers = {"authorization":"Bearer t", "content-type":"application/json", "x-mock-error":"500"}
    r = httpx.post(mock_url, json={"model":"x","messages":[]}, headers=headers)
    assert r.status_code == 500

Run pytest. Green means your client will survive real provider hiccups. Add a streaming test that asserts x-mock-stream-fail raises a connection error in the client and that your code reconnects or falls back.

Step 6: Validate fallback and cache-control behavior

Once the mock proves your client retries, extend it to emulate multiple providers behind one route. Return different errors based on x-mock-provider header, and include cache-control hints to confirm your client forwards them.

If you later point the same client at n4n.ai, its automatic fallback when a provider is rate-limited or degraded will exercise the same retry paths you validated locally. The gateway honors client routing directives and forwards provider cache-control hints, so the mock’s cache-control header should be passed through unchanged.

Add a check in the mock:

@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    # ... rate limit and error logic above ...
    body = await request.json()
    response = JSONResponse(content={
        "id": "mock-1",
        "object": "chat.completion",
        "model": body.get("model", "gpt-3.5-turbo"),
        "choices": [{"index": 0, "message": {"role": "assistant", "content": "echo"}, "finish_reason": "stop"}],
        "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}
    })
    if request.headers.get("x-mock-cache"):
        response.headers["cache-control"] = "max-age=60"
    return response

Then assert in tests that your client copies that header to its upstream calls. This catches bugs where a caching proxy or gateway hint is silently dropped.

Operational notes: run the mock in a Docker container in CI with ENV RATE_LIMIT=2 WINDOW=1. Kill it after tests. Because the mock is stateless except for the in-memory counter, restarting between test modules avoids cross-test contamination.

Building a local fault-injecting mock takes an hour and removes guesswork from resilience testing. You now have a repeatable way to simulate rate limits local llm mock errors and confirm your client does the right thing every time.