Testing Code That Calls a Model

Almost all the code around a model call is ordinary code, and ordinary code is tested without a network. The trick is to be honest about which of three quite different things you are testing, because each wants a different mechanism and a different cadence.

Three different tests wearing one name

Mixing them produces the worst of each: a CI suite that is slow, costs money, fails randomly, and still does not tell you the model got worse. Keep them in separate directories with separate commands.

The seam that makes it testable

Untestable LLM code is almost always code that constructs its own HTTP client inside the function that uses it. One parameter fixes it: pass the client (or a callable) in.

# summarise.py
from typing import Protocol


class ChatFn(Protocol):
    def __call__(self, messages: list[dict], **kwargs) -> str: ...


def summarise(text: str, chat: ChatFn) -> str:
    if not text.strip():
        raise ValueError("nothing to summarise")
    reply = chat([
        {"role": "system", "content": "Summarise in one sentence."},
        {"role": "user", "content": text[:8000]},
    ])
    reply = reply.strip()
    if not reply:
        raise ValueError("model returned an empty summary")
    return reply

Every rule in that function — the empty-input guard, the 8,000-character truncation, the empty-reply guard, the strip — is now testable in microseconds with a two-line fake. That is the entire argument for the seam.

# tests/test_summarise.py
import pytest

from summarise import summarise


def test_truncates_long_input():
    seen = {}

    def fake_chat(messages, **kwargs):
        seen["user"] = messages[-1]["content"]
        return "A summary."

    summarise("x" * 20_000, fake_chat)
    assert len(seen["user"]) == 8000


def test_rejects_empty_reply():
    with pytest.raises(ValueError, match="empty summary"):
        summarise("some text", lambda messages, **kw: "   ")


@pytest.mark.parametrize("reply,expected", [
    ("  A summary.  ", "A summary."),
    ("A summary.", "A summary."),
])
def test_strips(reply, expected):
    assert summarise("text", lambda m, **kw: reply) == expected

A fake transport, not a mocked SDK

For the layer that does speak HTTP, test it against a fake transport rather than by patching methods. httpx has one built in — httpx.MockTransport takes a function from request to response — so the real client, the real headers, the real JSON encoding and the real status handling all execute. Patching client.post tests none of that.

# tests/conftest.py
import json

import httpx
import pytest


def chat_response(text: str, *, prompt_tokens: int = 10,
                  completion_tokens: int = 5) -> dict:
    return {
        "id": "chatcmpl-test",
        "model": "test/model",
        "choices": [{
            "index": 0,
            "message": {"role": "assistant", "content": text},
            "finish_reason": "stop",
        }],
        "usage": {
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "total_tokens": prompt_tokens + completion_tokens,
        },
    }


@pytest.fixture
def transport_factory():
    """Build an httpx client whose responses come from a list of handlers."""
    def make(*responses: httpx.Response) -> httpx.Client:
        queue = list(responses)

        def handler(request: httpx.Request) -> httpx.Response:
            assert request.headers["authorization"].startswith("Bearer ")
            assert json.loads(request.content)["model"]
            return queue.pop(0) if queue else httpx.Response(200,
                                                             json=chat_response("ok"))

        return httpx.Client(
            transport=httpx.MockTransport(handler),
            base_url="https://example.invalid/v1",
            headers={"Authorization": "Bearer test-key"},
        )
    return make

# tests/test_retry.py
import httpx
import pytest

from retrying import call_model


def test_retries_a_503_then_succeeds(transport_factory):
    client = transport_factory(
        httpx.Response(503, text="upstream unavailable"),
        httpx.Response(200, json=chat_response("second time lucky")),
    )
    body = call_model(client, {"model": "test/model", "messages": []})
    assert body["choices"][0]["message"]["content"] == "second time lucky"


def test_does_not_retry_a_400(transport_factory):
    client = transport_factory(
        httpx.Response(400, json={"error": {"message": "bad parameter"}}),
        httpx.Response(200, json=chat_response("should never be reached")),
    )
    with pytest.raises(httpx.HTTPStatusError) as exc:
        call_model(client, {"model": "test/model", "messages": []})
    assert exc.value.response.status_code == 400

The second test is the valuable one and the one nobody writes. A retry policy that retries too much passes every happy-path test, and its cost only appears as a bill and a latency graph.

Recorded responses

Handwritten fixtures cover the shapes you thought of. Recorded ones cover the shapes that actually occurred — the reply with the code fence, the one that hit max_tokens, the 429 with the odd body. Record once, commit, replay for ever.

1. Capture. Point the real client at the real endpoint once, with a wrapper that writes each response body to tests/fixtures/<name>.json. The logging decorator from logging every model call already has the data; this is a small script over the JSONL.
2. Redact. Before committing, strip keys, personal data and anything customer-identifiable from the recorded bodies. A fixtures directory is committed to a repository many people can read.
3. Replay. A fixture loader plus MockTransport is all that is needed; the libraries that do this automatically (vcrpy, respx) are convenient but add a dependency whose own API you then have to track.
4. Keep the ugly ones. The fixture worth its disk space is the malformed one. A directory of six perfect responses tests nothing that was ever going to break.

Record and replay for LLM calls goes further into the trade-offs, and testing without the model covers how much of a system can be verified with no model at all.

Testing the streaming path

Streaming code is the part most often left untested, because it looks like it needs a real server. It does not: MockTransport can return a response whose body is a byte stream, so the SSE frames your parser has to survive are ordinary test data.

One change to the code under test is required, and it is the same seam as before: stream_chat takes the client as its first argument rather than building one. A streaming function that constructs its own httpx.Client cannot be tested at all without patching, which is reason enough to move the client out.

# tests/test_streaming.py
import httpx
import pytest

from stream import stream_chat


def sse(*frames: str) -> bytes:
    return "".join(f"data: {f}\n\n" for f in frames).encode("utf-8")


def streaming_client(payload: bytes, status: int = 200) -> httpx.Client:
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(status, content=payload,
                              headers={"content-type": "text/event-stream"})
    return httpx.Client(transport=httpx.MockTransport(handler),
                        base_url="https://example.invalid/v1")


def test_assembles_fragments_and_stops_at_done():
    body = sse(
        '{"choices":[{"delta":{"role":"assistant"}}]}',     # no content: must not crash
        '{"choices":[{"delta":{"content":"Hello"}}]}',
        '{"choices":[{"delta":{"content":" world"}}]}',
        '{"choices":[{"delta":{},"finish_reason":"stop"}]}',
        "[DONE]",
        '{"choices":[{"delta":{"content":"NEVER"}}]}',      # after DONE: ignored
    )
    client = streaming_client(body)
    assert "".join(stream_chat(client, [], "test/model")) == "Hello world"


def test_survives_a_keepalive_comment_and_a_bad_frame():
    body = (b": keep-alive\n\n"
            + sse('{"choices":[{"delta":{"content":"a"}}]}',
                  "{not json",
                  '{"choices":[{"delta":{"content":"b"}}]}',
                  "[DONE]"))
    assert "".join(stream_chat(streaming_client(body), [], "test/model")) == "ab"


def test_truncated_stream_is_not_silently_a_success():
    body = sse('{"choices":[{"delta":{"content":"partial"}}]}')   # no DONE
    with pytest.raises(RuntimeError, match="ended without"):
        list(stream_chat(streaming_client(body), [], "test/model"))

The third test is the one that matters and the one that will fail against the generator as written earlier in this cluster, because it returns quietly when the stream ends without a terminator. That is the point: the test states the behaviour you want — a cut stream must not look like a short answer — and the fix is a flag set on [DONE] or a finish_reason and a raise if the loop exits without either.

The first test encodes two facts about real streams in six lines: the opening frame has no content, and anything after [DONE] is not yours to read. Both are cheap to assert and both are bugs that otherwise surface in production against one provider and not another.

The two tests worth running live

Two, and they run on a schedule rather than on every push — nightly, or on deploy. They cost a few cents a day and they catch the class of failure no offline test can.

# tests/live/test_contract.py
import json
import os

import httpx
import pytest

from retrying import call_model

pytestmark = pytest.mark.skipif(
    not os.environ.get("RUN_LIVE_TESTS"),
    reason="set RUN_LIVE_TESTS=1 to run tests that call the provider",
)


@pytest.fixture(scope="module")
def real_client():
    with httpx.Client(
        base_url=os.environ["LLM_BASE_URL"].rstrip("/"),
        headers={"Authorization": f"Bearer {os.environ['LLM_API_KEY']}"},
        timeout=60.0,
    ) as client:
        yield client


def test_response_shape_is_unchanged(real_client):
    """1. The contract: the fields our parser depends on still exist."""
    body = call_model(real_client, {
        "model": os.environ["LLM_MODEL"],
        "messages": [{"role": "user", "content": "Reply with the word OK."}],
        "max_tokens": 5,
        "temperature": 0,
    })
    assert body["choices"][0]["message"]["content"]
    assert body["choices"][0]["finish_reason"] in {"stop", "length"}
    assert body["usage"]["prompt_tokens"] > 0
    assert body["usage"]["completion_tokens"] > 0


def test_structured_output_is_still_accepted(real_client):
    """2. The features we rely on are still supported for this model."""
    body = call_model(real_client, {
        "model": os.environ["LLM_MODEL"],
        "messages": [{"role": "user", "content": "Any two-word city name."}],
        "response_format": {
            "type": "json_schema",
            "json_schema": {"name": "city", "strict": True, "schema": {
                "type": "object",
                "properties": {"city": {"type": "string"}},
                "required": ["city"],
                "additionalProperties": False,
            }},
        },
        "max_tokens": 50,
    })
    parsed = json.loads(body["choices"][0]["message"]["content"])
    assert isinstance(parsed["city"], str)

These two answer “has anything changed underneath us”, which is the failure mode that has no local reproduction. A provider dropping a parameter, a model id being retired, or a gateway rejecting a field it used to ignore all show up here and nowhere else — silent model updates is the general problem.

Mark them and exclude them by default. In pyproject.toml:

[tool.pytest.ini_options]
markers = [
  "live: hits a real provider; costs money; requires RUN_LIVE_TESTS=1",
]
addopts = "-m 'not live'"

Making the suite deterministic

• Never assert on model text in an offline test. There is no model in an offline test; the text came from your fixture. Asserting on it tests the fixture.
• Assert on structure and behaviour, not wording. “is valid JSON”, “has these keys”, “urgency is between 1 and 5”, “the retry happened twice”. Never assert reply == "Paris", even live.
• Freeze time and randomness. Backoff uses random, cache keys embed timestamps, and prompts sometimes include today’s date. Seed the first, inject a clock for the second, and pass the date in for the third.
• Make the network unreachable in CI. Point LLM_BASE_URL at https://example.invalid and give CI no API key. A test that quietly reaches a real provider is a test that works on your machine and bills you from the build server.
• Test the async paths with the async tools. pytest-asyncio is the usual choice; its default mode changed across versions, so set asyncio_mode explicitly in pyproject.toml instead of relying on the default that happened to be installed.