100%

Page 13 of 22 · 289 min left

Chapter 3 · Production FastAPI

Testing Professional APIs

Page 13 of 22
26 min read

In Testing Your API with pytest you tested endpoints that had no outside world. Real AI services are mostly outside world: a database, an upstream API, an LLM provider that costs money per call and never answers the same way twice, rate limits, retries, streams and webhooks. A professional test suite checks all of that in seconds, without the network, without spending a single token, and gives the same result on every run.

The trick is to design the app with seams (dependencies you can swap) and to build a small toolkit of fixtures that swap them. This page builds that toolkit around one realistic service and then tests every risky part of it.

What you will learn

The testing pyramid, unit vs integration tests, and markers to run them separately
Fixtures with scopes, factory fixtures, parametrize, and app.dependency_overrides
Mocking external HTTP (httpx.MockTransport and respx) and LLM providers; database tests that roll back
Testing auth, rate limits, retries, timeouts, streaming responses and webhook signatures
Contract tests with an OpenAPI snapshot, coverage, flaky tests, load testing and security regression tests

The testing pyramid

Output

                 ▲  slow, brittle, expensive
                ╱ ╲
               ╱E2E╲          a few: the deployed system, real browser or client
              ╱─────╲
             ╱ integ-╲        some: app + real DB / real HTTP / real provider sandbox
            ╱ ration  ╲
           ╱───────────╲
          ╱  API tests  ╲     many: the whole app through TestClient, outside world faked
         ╱───────────────╲
        ╱   unit tests    ╲   most: pure functions and classes, no I/O at all
       ╱───────────────────╲
                 ▼  fast, stable, cheap

A unit test checks one function or class with no I/O: a signature check, a rate limiter, a prompt builder. Thousands run in a second. An API test sends real HTTP requests through FastAPI's TestClient into the whole app (routing, validation, dependencies, error handlers), with the outside world replaced by fakes. An integration test lets one real dependency in (a real database, the real upstream), so it is slower and can fail for reasons outside your code; you run fewer of them. End-to-end tests drive the deployed system. Most bugs are caught cheaply at the bottom; the higher layers prove the pieces fit.

The service under test

A course-review API: users post reviews (checked against the course service), delete their own, ask an LLM for a summary (plain or streamed), and the payment provider calls a signed webhook. Set up the project:

Bash

mkdir -p ~/projects/course-reviews && cd ~/projects/course-reviews
uv init -q --bare --name course-reviews
uv add -q "fastapi[standard]" sqlalchemy
uv add -q --dev pytest httpx2 respx
source .venv/bin/activate
mkdir -p app tests && touch app/__init__.py tests/__init__.py
# pip instead of uv:
#   python -m venv .venv && source .venv/bin/activate
#   pip install -q "fastapi[standard]" sqlalchemy pytest httpx2 respx

httpx2 is there for TestClient (Starlette now prefers it, as explained in the beginner tutorial); the app itself calls other services with httpx, the API most documentation and the respx mocking library use. Read the app with one question in mind: where can a test cut in? Every outside thing arrives through a dependency (get_session, get_http, get_llm, get_limiter), and anything time-based takes a clock or a now.

Python

# file: app/main.py
"""Course Reviews API: the service this page tests."""
import hashlib
import hmac
import json
import os
import time
from collections.abc import Iterator
from contextlib import asynccontextmanager
from typing import Annotated, Protocol

import httpx
from fastapi import Depends, FastAPI, Header, HTTPException, Request
from fastapi.sse import EventSourceResponse, ServerSentEvent
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import String, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column

PRACTICE_API_URL = os.environ.get("PRACTICE_API_URL", "http://localhost:8800")
DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///reviews.db")
WEBHOOK_SECRET = os.environ.get("WEBHOOK_SECRET", "dev-only-secret").encode()
TOKENS = {"token-nadia": "nadia", "token-tanvir": "tanvir", "token-admin": "admin"}  # demo only
ADMINS = {"admin"}


class Base(DeclarativeBase):
    pass


class Review(Base):
    __tablename__ = "reviews"
    id: Mapped[int] = mapped_column(primary_key=True)
    author: Mapped[str] = mapped_column(String(50))
    course_id: Mapped[int]
    rating: Mapped[int]
    comment: Mapped[str] = mapped_column(String(500))


class ReviewIn(BaseModel):
    course_id: int = Field(ge=1)
    rating: int = Field(ge=1, le=5)
    comment: str = Field(min_length=3, max_length=500)


class ReviewOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    author: str
    course_id: int
    rating: int
    comment: str


engine = create_engine(DATABASE_URL)


@asynccontextmanager
async def lifespan(app: FastAPI):
    Base.metadata.create_all(engine)
    app.state.http = httpx.Client(base_url=PRACTICE_API_URL, timeout=httpx.Timeout(3.0, connect=1.0))
    yield
    app.state.http.close()


app = FastAPI(title="Course Reviews", lifespan=lifespan)


# ---------- dependencies: each one is a seam that tests can replace
def get_session() -> Iterator[Session]:
    with Session(engine) as session:
        yield session


def get_http(request: Request) -> httpx.Client:
    return request.app.state.http


class LLM(Protocol):
    def complete(self, prompt: str) -> str: ...
    def stream(self, prompt: str) -> Iterator[str]: ...


class PracticeLLM:
    """Calls the Practice API's fake LLM; a real provider adapter would sit here."""

    def __init__(self, http: httpx.Client):
        self.http = http

    def _request(self, prompt: str, stream: bool) -> dict:
        return {"headers": {"Authorization": "Bearer token-nadia-001"}, "timeout": 30,
                "json": {"messages": [{"role": "user", "content": prompt}], "stream": stream}}

    def complete(self, prompt: str) -> str:
        resp = self.http.post("/v1/chat", **self._request(prompt, stream=False))
        resp.raise_for_status()
        return resp.json()["message"]["content"]

    def stream(self, prompt: str) -> Iterator[str]:
        with self.http.stream("POST", "/v1/chat", **self._request(prompt, stream=True)) as resp:
            resp.raise_for_status()
            for line in resp.iter_lines():
                if line.startswith("data: ") and "delta" in (event := json.loads(line[6:])):
                    yield event["delta"]


def get_llm(http: Annotated[httpx.Client, Depends(get_http)]) -> LLM:
    return PracticeLLM(http)


class RateLimiter:
    """Sliding-window limiter kept in memory (use Redis in production, as on the previous page)."""

    def __init__(self, limit: int, window: float, clock=time.monotonic):
        self.limit, self.window, self.clock = limit, window, clock
        self.hits: dict[str, list[float]] = {}

    def allow(self, key: str) -> bool:
        now = self.clock()
        recent = [t for t in self.hits.get(key, []) if now - t < self.window]
        allowed = len(recent) < self.limit
        if allowed:
            recent.append(now)
        self.hits[key] = recent
        return allowed


review_limiter = RateLimiter(limit=3, window=60)


def get_limiter() -> RateLimiter:
    return review_limiter


def current_user(authorization: Annotated[str | None, Header()] = None) -> str:
    scheme, _, token = (authorization or "").partition(" ")
    user = TOKENS.get(token) if scheme.lower() == "bearer" else None
    if user is None:
        raise HTTPException(401, "Not authenticated", headers={"WWW-Authenticate": "Bearer"})
    return user


SessionDep = Annotated[Session, Depends(get_session)]
HttpDep = Annotated[httpx.Client, Depends(get_http)]
LLMDep = Annotated[LLM, Depends(get_llm)]
LimiterDep = Annotated[RateLimiter, Depends(get_limiter)]
UserDep = Annotated[str, Depends(current_user)]


# ---------- business logic
def fetch_course(http: httpx.Client, course_id: int, attempts: int = 3) -> dict | None:
    """Look a course up in the course service, retrying 503s."""
    for attempt in range(1, attempts + 1):
        try:
            resp = http.get(f"/v1/courses/{course_id}")
        except httpx.TimeoutException:
            raise HTTPException(504, "Course service timed out")
        if resp.status_code == 404:
            return None
        if resp.status_code == 503 and attempt < attempts:
            continue                       # a real client also backs off here
        if resp.status_code != 200:
            raise HTTPException(502, "Course service error")
        return resp.json()


def verify_signature(secret: bytes, timestamp: str, body: bytes, signature: str,
                     now: float, tolerance: int = 300) -> bool:
    if not timestamp.isdigit() or abs(now - int(timestamp)) > tolerance:
        return False                       # missing, malformed or too old: replay protection
    expected = hmac.new(secret, timestamp.encode() + b"." + body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)


# ---------- routes
@app.get("/health")
def health() -> dict:
    return {"status": "ok"}


@app.post("/reviews", status_code=201)
def create_review(review: ReviewIn, user: UserDep, session: SessionDep,
                  http: HttpDep, limiter: LimiterDep) -> ReviewOut:
    if not limiter.allow(user):
        raise HTTPException(429, "Too many reviews, slow down", headers={"Retry-After": "60"})
    if fetch_course(http, review.course_id) is None:
        raise HTTPException(422, f"Course {review.course_id} does not exist")
    row = Review(author=user, **review.model_dump())
    session.add(row)
    session.commit()
    return ReviewOut.model_validate(row)


@app.get("/reviews/{review_id}")
def get_review(review_id: int, session: SessionDep) -> ReviewOut:
    row = session.get(Review, review_id)
    if row is None:
        raise HTTPException(404, "Review not found")
    return ReviewOut.model_validate(row)


@app.delete("/reviews/{review_id}", status_code=204)
def delete_review(review_id: int, user: UserDep, session: SessionDep) -> None:
    row = session.get(Review, review_id)
    if row is None or (row.author != user and user not in ADMINS):
        raise HTTPException(404, "Review not found")     # don't reveal other people's reviews
    session.delete(row)
    session.commit()


def course_prompt(course_id: int, session: SessionDep) -> tuple[str, int]:
    """A dependency, so a 404 is raised before any streaming starts."""
    comments = session.scalars(select(Review.comment).where(Review.course_id == course_id)).all()
    if not comments:
        raise HTTPException(404, "No reviews for this course yet")
    return "Summarise these course reviews in one sentence:\n" + "\n".join(comments), len(comments)


PromptDep = Annotated[tuple[str, int], Depends(course_prompt)]


@app.post("/courses/{course_id}/summary")
def summarise(course_id: int, prompt: PromptDep, llm: LLMDep) -> dict:
    text, count = prompt
    return {"course_id": course_id, "reviews": count, "summary": llm.complete(text)}


@app.get("/courses/{course_id}/summary/stream", response_class=EventSourceResponse)
def summarise_stream(prompt: PromptDep, llm: LLMDep) -> Iterator[ServerSentEvent]:
    text, count = prompt
    for delta in llm.stream(text):
        yield ServerSentEvent(data={"delta": delta})
    yield ServerSentEvent(event="done", data={"reviews": count})


@app.post("/webhooks/payments")
async def payment_webhook(request: Request, x_timestamp: Annotated[str, Header()],
                          x_signature: Annotated[str, Header()]) -> dict:
    body = await request.body()
    if not verify_signature(WEBHOOK_SECRET, x_timestamp, body, x_signature, time.time()):
        raise HTTPException(401, "Invalid signature")
    return {"received": json.loads(body)["event"]}

Tell pytest where the tests are and declare the one custom marker the suite uses (with --strict-markers, a typo in a marker name is an error instead of a silently ignored test):

Bash

cat >> pyproject.toml <<'EOF'

[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "--strict-markers"
markers = ["integration: talks to real services (deselect with -m 'not integration')"]
EOF
export PRACTICE_API_URL=http://localhost:8800

Fixtures: the test toolkit

All shared setup lives in tests/conftest.py, which pytest loads automatically. Each fixture replaces one seam:

Python

# file: tests/conftest.py
"""Shared fixtures: a rolled-back database, fake HTTP, a fake LLM and a fresh rate limiter."""
import httpx
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, event
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool

from app import main
from app.main import Base, RateLimiter, Review, app

COURSES = {1: {"id": 1, "title": "How AI Works"}, 8: {"id": 8, "title": "RAG Systems"}}


class FakeLLM:
    """Stands in for the LLM provider: instant, free, predictable, and it records prompts."""

    def __init__(self):
        self.prompts = []

    def complete(self, prompt):
        self.prompts.append(prompt)
        return "Learners find it clear."

    def stream(self, prompt):
        self.prompts.append(prompt)
        yield from ["Learners ", "find ", "it ", "clear."]


def fake_course_service(request: httpx.Request) -> httpx.Response:
    course_id = int(request.url.path.rsplit("/", 1)[-1])
    if course_id in COURSES:
        return httpx.Response(200, json=COURSES[course_id])
    return httpx.Response(404, json={"detail": "Course not found"})


@pytest.fixture(scope="session")
def engine():
    """One in-memory database for the whole test run."""
    engine = create_engine("sqlite://", poolclass=StaticPool,
                           connect_args={"check_same_thread": False})

    @event.listens_for(engine, "connect")
    def _no_driver_transactions(dbapi_connection, record):
        dbapi_connection.isolation_level = None    # let SQLAlchemy emit BEGIN/SAVEPOINT itself

    @event.listens_for(engine, "begin")
    def _begin(connection):
        connection.exec_driver_sql("BEGIN")

    Base.metadata.create_all(engine)
    yield engine
    engine.dispose()


@pytest.fixture
def db_session(engine):
    """Each test runs inside a transaction that is rolled back afterwards."""
    with engine.connect() as connection:
        transaction = connection.begin()
        session = Session(bind=connection, join_transaction_mode="create_savepoint")
        yield session
        session.close()
        transaction.rollback()


@pytest.fixture
def llm():
    return FakeLLM()


@pytest.fixture
def course_service():
    """What the fake course service does; a test may swap the handler."""
    return {"handler": fake_course_service, "calls": 0}


@pytest.fixture
def client(db_session, llm, course_service):
    def transport_handler(request):
        course_service["calls"] += 1
        return course_service["handler"](request)

    http = httpx.Client(base_url="http://courses.test", transport=httpx.MockTransport(transport_handler))
    limiter = RateLimiter(limit=3, window=60)       # a fresh limiter per test
    app.dependency_overrides[main.get_session] = lambda: db_session
    app.dependency_overrides[main.get_http] = lambda: http
    app.dependency_overrides[main.get_llm] = lambda: llm
    app.dependency_overrides[main.get_limiter] = lambda: limiter
    yield TestClient(app)
    app.dependency_overrides.clear()
    http.close()


@pytest.fixture
def make_review(db_session):
    """A factory fixture: call it to create reviews with sensible defaults."""
    def make(author="nadia", course_id=1, rating=5, comment="Very clear examples"):
        review = Review(author=author, course_id=course_id, rating=rating, comment=comment)
        db_session.add(review)
        db_session.commit()
        return review
    return make


NADIA = {"Authorization": "Bearer token-nadia"}
TANVIR = {"Authorization": "Bearer token-tanvir"}
ADMIN = {"Authorization": "Bearer token-admin"}

The ideas in it, one by one:

Scopes. engine has scope="session": creating the database and tables once per run is fast. Everything else is function-scoped (the default): a fresh fake LLM, fake course service and limiter for every test, so no test sees another's leftovers. Other scopes are module, class and package.
Rollback per test. db_session opens a transaction, and the session inside it turns every commit() into a savepoint (join_transaction_mode="create_savepoint"). At the end the outer transaction is rolled back, so the next test starts with empty tables, and no time is spent recreating them. SQLite's Python driver manages transactions in its own way, so the two event hooks hand that job to SQLAlchemy; PostgreSQL needs no hooks.
app.dependency_overrides maps a dependency function to a replacement. FastAPI calls the replacement everywhere the original was used, including inside other dependencies. Clear the overrides after each test.
httpx.MockTransport is a fake network: the client works exactly as usual, but every request is answered by your function instead of a server. No sockets, no flakiness.
A fake LLM with the same methods as the real adapter (the LLM protocol). It answers instantly, costs nothing, and records the prompts, so tests can check what would have been sent.
A factory fixture (make_review) returns a function, so a test can create as many rows as it needs with only the fields that matter to it.

API tests and parametrize

Python

# file: tests/test_reviews.py
import pytest

from tests.conftest import NADIA

GOOD = {"course_id": 1, "rating": 5, "comment": "Very clear examples"}


def test_create_review(client):
    resp = client.post("/reviews", json=GOOD, headers=NADIA)
    assert resp.status_code == 201
    assert resp.json() == {"id": 1, "author": "nadia", **GOOD}


def test_database_is_clean_for_every_test(client):
    assert client.get("/reviews/1").status_code == 404


@pytest.mark.parametrize("field, value", [
    ("rating", 0), ("rating", 6), ("comment", "ok"), ("course_id", 0), ("comment", "x" * 501),
])
def test_invalid_reviews_are_rejected(client, field, value):
    resp = client.post("/reviews", json={**GOOD, field: value}, headers=NADIA)
    assert resp.status_code == 422
    assert resp.json()["detail"][0]["loc"] == ["body", field]


def test_unknown_course_is_rejected(client):
    resp = client.post("/reviews", json={**GOOD, "course_id": 99}, headers=NADIA)
    assert resp.status_code == 422
    assert resp.json() == {"detail": "Course 99 does not exist"}

Bash

pytest -q

Output

........                                                                                     [100%]
8 passed in 0.04s

@pytest.mark.parametrize turned one test function into five tests, one per bad input, and each is reported separately when it fails. The second test proves the rollback works: the review created by the first test is gone. --setup-show shows the fixtures being built and torn down around a test, which is the quickest way to understand scopes:

Bash

pytest -q --setup-show tests/test_reviews.py::test_create_review

Output


SETUP    S engine
        SETUP    F db_session (fixtures used: engine)
        SETUP    F llm
        SETUP    F course_service
        SETUP    F client (fixtures used: course_service, db_session, llm)
        tests/test_reviews.py::test_create_review (fixtures used: client, course_service, db_session, engine, llm) .
        TEARDOWN F client
        TEARDOWN F course_service
        TEARDOWN F llm
        TEARDOWN F db_session
TEARDOWN S engine
1 passed in 0.02s

The session-scoped engine (S) wraps everything; the function-scoped fixtures (F) are created for this one test and torn down in reverse order.

Testing auth and permissions

Test every route for three callers: nobody, the wrong user and the right user. The wrong-user case is the one that catches BOLA (broken object level authorisation, number one in the OWASP API Top 10): an endpoint that checks you are logged in but not that the object is yours. A test written for a fixed security bug is a security regression test; it makes sure the hole never comes back.

Python

# file: tests/test_auth.py
import pytest

from tests.conftest import ADMIN, NADIA, TANVIR


@pytest.mark.parametrize("headers", [{}, {"Authorization": "Bearer wrong"}, {"Authorization": "token-nadia"}])
def test_writing_needs_a_valid_token(client, headers):
    resp = client.post("/reviews", json={"course_id": 1, "rating": 5, "comment": "Nice"}, headers=headers)
    assert resp.status_code == 401
    assert resp.headers["WWW-Authenticate"] == "Bearer"


def test_owner_can_delete(client, make_review):
    review = make_review(author="nadia")
    assert client.delete(f"/reviews/{review.id}", headers=NADIA).status_code == 204
    assert client.get(f"/reviews/{review.id}").status_code == 404


def test_other_users_cannot_delete(client, make_review):
    """Security regression test for BOLA: Tanvir must not delete Nadia's review."""
    review = make_review(author="nadia")
    assert client.delete(f"/reviews/{review.id}", headers=TANVIR).status_code == 404
    assert client.get(f"/reviews/{review.id}").status_code == 200


def test_admin_can_delete_any_review(client, make_review):
    review = make_review(author="tanvir")
    assert client.delete(f"/reviews/{review.id}", headers=ADMIN).status_code == 204

For JWT-based auth (from Authentication and Authorisation) add cases for an expired token, a token signed with another key and one with alg set to none; make tokens in a fixture with PyJWT and a fixed exp, or override the current_user dependency in tests that are not about auth.

Rate limits, retries and timeouts

These depend on time and on failures, two things that are slow and random in real life. Tests control both: the limiter takes a fake clock, and the fake course service returns whatever sequence of answers, or exceptions, the test wants.

Python

# file: tests/test_limits_and_upstream.py
import httpx

from app.main import RateLimiter
from tests.conftest import NADIA, TANVIR

GOOD = {"course_id": 1, "rating": 4, "comment": "Good pace"}


def test_fourth_review_in_a_minute_is_limited(client):
    codes = [client.post("/reviews", json=GOOD, headers=NADIA).status_code for _ in range(4)]
    assert codes == [201, 201, 201, 429]
    assert client.post("/reviews", json=GOOD, headers=TANVIR).status_code == 201


def test_limiter_window_slides():
    now = [0.0]
    limiter = RateLimiter(limit=2, window=60, clock=lambda: now[0])
    assert [limiter.allow("nadia") for _ in range(3)] == [True, True, False]
    now[0] = 60.0                                   # time passes without sleeping
    assert limiter.allow("nadia") is True


def test_course_service_503s_are_retried(client, course_service):
    replies = iter([httpx.Response(503), httpx.Response(503), httpx.Response(200, json={"id": 1})])
    course_service["handler"] = lambda request: next(replies)
    assert client.post("/reviews", json=GOOD, headers=NADIA).status_code == 201
    assert course_service["calls"] == 3


def test_course_service_keeps_failing(client, course_service):
    course_service["handler"] = lambda request: httpx.Response(503)
    resp = client.post("/reviews", json=GOOD, headers=NADIA)
    assert (resp.status_code, course_service["calls"]) == (502, 3)


def test_course_service_timeout_becomes_504(client, course_service):
    def hang(request):
        raise httpx.ReadTimeout("timed out", request=request)

    course_service["handler"] = hang
    resp = client.post("/reviews", json=GOOD, headers=NADIA)
    assert resp.status_code == 504
    assert course_service["calls"] == 1             # timeouts are not retried here

Notice what is asserted: not only the final status code, but how many calls reached the upstream. That is what proves the retry logic retries 503s exactly as often as intended and does not retry a timeout. No test sleeps; moving the fake clock forward 60 seconds takes no time. If production code sleeps between retries, make the sleep function a parameter too, or patch it with monkeypatch.setattr.

Mocking the LLM provider, and streaming

Python

# file: tests/test_llm.py
from tests.conftest import NADIA


def test_summary_uses_the_reviews(client, llm, make_review):
    make_review(comment="Very clear examples")
    make_review(comment="Too fast at the end", rating=3)
    resp = client.post("/courses/1/summary")
    assert resp.json() == {"course_id": 1, "reviews": 2, "summary": "Learners find it clear."}
    assert "Too fast at the end" in llm.prompts[0]


def test_no_reviews_means_404_and_no_llm_call(client, llm):
    assert client.post("/courses/8/summary").status_code == 404
    assert llm.prompts == []                        # no tokens spent


def test_summary_stream(client, make_review):
    make_review()
    with client.stream("GET", "/courses/1/summary/stream") as resp:
        assert resp.headers["content-type"].startswith("text/event-stream")
        lines = [line for line in resp.iter_lines() if line]
    assert lines == [
        'data: {"delta": "Learners "}', 'data: {"delta": "find "}',
        'data: {"delta": "it "}', 'data: {"delta": "clear."}',
        "event: done", 'data: {"reviews": 1}',
    ]

Testing against a fake LLM checks your code: that the prompt contains the reviews, that no tokens are spent when there is nothing to summarise, and that the stream has the right shape. client.stream(...) reads a streaming response line by line, which works for Server-Sent Events. What these tests cannot check is whether the real model writes good summaries; that is evaluation, a separate activity with its own datasets and scores (see Building an LLM Backend with Streaming). Keep the two apart: unit and API tests must be deterministic and free; evals are statistical and cost money.

Webhook signatures

A webhook endpoint must reject anything not signed by the sender with the shared secret, and old messages replayed by an attacker. Test the accepted case and each way of being wrong, plus the pure signature function on its own with a fixed now:

Python

# file: tests/test_webhooks.py
import hashlib
import hmac
import json
import time

import pytest

from app.main import WEBHOOK_SECRET, verify_signature


def signed(body: dict, secret: bytes = WEBHOOK_SECRET, timestamp: int | None = None):
    raw = json.dumps(body).encode()
    ts = str(timestamp if timestamp is not None else int(time.time()))
    sig = hmac.new(secret, ts.encode() + b"." + raw, hashlib.sha256).hexdigest()
    return raw, {"X-Timestamp": ts, "X-Signature": sig, "Content-Type": "application/json"}


def test_valid_webhook_is_accepted(client):
    raw, headers = signed({"event": "payment.succeeded"})
    resp = client.post("/webhooks/payments", content=raw, headers=headers)
    assert resp.json() == {"received": "payment.succeeded"}


@pytest.mark.parametrize("case", ["wrong secret", "tampered body", "replayed"])
def test_bad_webhooks_are_rejected(client, case):
    if case == "wrong secret":
        raw, headers = signed({"event": "payment.succeeded"}, secret=b"guess")
    elif case == "tampered body":
        _, headers = signed({"event": "payment.succeeded", "amount": 500})
        raw = json.dumps({"event": "payment.succeeded", "amount": 5}).encode()
    else:
        raw, headers = signed({"event": "payment.succeeded"}, timestamp=int(time.time()) - 3600)
    assert client.post("/webhooks/payments", content=raw, headers=headers).status_code == 401


def test_signature_check_is_a_pure_function():
    """A unit test: no app, no HTTP, a fixed clock."""
    raw = b'{"event": "x"}'
    sig = hmac.new(b"k", b"1000." + raw, hashlib.sha256).hexdigest()
    assert verify_signature(b"k", "1000", raw, sig, now=1100)
    assert not verify_signature(b"k", "1000", raw, sig, now=1400)     # 400 s later: too old
    assert not verify_signature(b"k", "abc", raw, sig, now=1000)

Bash

pytest -q

Output

...........................                                                                  [100%]
27 passed in 0.11s

Contract tests: an OpenAPI snapshot

Your OpenAPI schema is a promise to every client: a frontend, a mobile app, another team, an AI agent calling your API as a tool. A contract test fails when that promise changes by accident. The simplest form saves the schema once and compares later runs with it:

Python

# file: tests/test_contract.py
"""Contract test: the OpenAPI schema is the promise you made to clients."""
import json
import os
from pathlib import Path

from app.main import app

SNAPSHOT = Path(__file__).parent / "openapi.snapshot.json"


def test_openapi_matches_snapshot():
    current = app.openapi()
    if os.environ.get("UPDATE_SNAPSHOT") == "1" or not SNAPSHOT.exists():
        SNAPSHOT.write_text(json.dumps(current, indent=2, sort_keys=True) + "\n")
    saved = json.loads(SNAPSHOT.read_text())
    assert sorted(current["paths"]) == sorted(saved["paths"]), "endpoints were added or removed"
    for name, schema in saved["components"]["schemas"].items():
        now = current["components"]["schemas"].get(name, {})
        assert now.get("properties") == schema.get("properties"), f"schema {name} changed"
        assert now.get("required") == schema.get("required"), f"required fields of {name} changed"

Bash

pytest -q tests/test_contract.py
ls tests/openapi.snapshot.json

Output

.                                                                                            [100%]
1 passed in 0.01s
tests/openapi.snapshot.json

Now a teammate renames comment to text in the request model, a breaking change for every client that sends comment:

Bash

sed -i 's/    comment: str = Field(min_length=3, max_length=500)/    text: str = Field(min_length=3, max_length=500)/' app/main.py
pytest -q --tb=short tests/test_contract.py

Output

F                                                                                            [100%]
============================================= FAILURES =============================================
__________________________________ test_openapi_matches_snapshot ___________________________________
tests/test_contract.py:20: in test_openapi_matches_snapshot
    assert now.get("properties") == schema.get("properties"), f"schema {name} changed"
E   AssertionError: schema ReviewIn changed
E   assert {'course_id':...tle': 'Text'}} == {'comment': {...': 'integer'}}
E     
E     Omitting 2 identical items, use -vv to show
E     Left contains 1 more item:
E     {'text': {'maxLength': 500, 'minLength': 3, 'title': 'Text', 'type': 'string'}}
E     Right contains 1 more item:
E     {'comment': {'maxLength': 500,
E                  'minLength': 3,...
E     
E     ...Full output truncated (3 lines hidden), use '-vv' to show
===================================== short test summary info ======================================
FAILED tests/test_contract.py::test_openapi_matches_snapshot - AssertionError: schema ReviewIn ch...
1 failed in 0.04s

The test does not decide whether the change is allowed; it forces a human to decide. If it is intended, version the API or announce it, then refresh the snapshot with UPDATE_SNAPSHOT=1 pytest tests/test_contract.py and commit the new file so reviewers see the change in the pull request. Here it was an accident, so undo it:

Bash

sed -i 's/    text: str = Field(min_length=3, max_length=500)/    comment: str = Field(min_length=3, max_length=500)/' app/main.py
pytest -q tests/test_contract.py

Output

.                                                                                            [100%]
1 passed in 0.01s

Dedicated tools go further: oasdiff classifies differences between two OpenAPI files as breaking or not, and Schemathesis generates thousands of requests from your schema to find inputs that crash the server.

Integration tests, respx and markers

Some tests should touch the real thing: the real course service, the real adapter's streaming code, the full app with its lifespan. Mark them, so the fast suite can leave them out:

Python

# file: tests/test_integration.py
"""Tests that talk to real services. Slower; run them on purpose with -m integration."""
import httpx
import pytest
import respx
from fastapi.testclient import TestClient

from app import main

pytestmark = pytest.mark.integration


def test_real_course_service_lookup():
    with httpx.Client(base_url=main.PRACTICE_API_URL, timeout=5) as http:
        assert main.fetch_course(http, 2)["title"] == "Python for AI"
        assert main.fetch_course(http, 999) is None


def test_practice_llm_adapter_streams():
    with httpx.Client(base_url=main.PRACTICE_API_URL, timeout=5) as http:
        text = "".join(main.PracticeLLM(http).stream("What is MLOps?"))
    assert text.startswith("This is a practice answer about: What is MLOps.")


@respx.mock
def test_app_with_respx_and_lifespan(tmp_path, monkeypatch):
    """The whole app, real lifespan and real httpx client; respx answers the HTTP calls."""
    monkeypatch.setattr(main, "engine", main.create_engine(f"sqlite:///{tmp_path}/test.db"))
    route = respx.get(f"{main.PRACTICE_API_URL}/v1/courses/1").mock(
        return_value=httpx.Response(200, json={"id": 1}))
    with TestClient(main.app) as client:            # "with" runs the lifespan
        resp = client.post("/reviews", json={"course_id": 1, "rating": 5, "comment": "Great"},
                           headers={"Authorization": "Bearer token-nadia"})
    assert resp.status_code == 201
    assert route.call_count == 1

The third test uses respx, which patches httpx itself: every request any httpx client makes is matched against the routes you declare, and an unmatched request is an error. That suits code that creates its own clients, where you cannot inject a transport. monkeypatch swaps the module's engine for a temporary file database and undoes the swap after the test. (respx patches the httpx package only; for a client built on httpx2, inject a MockTransport as the fixtures above do.)

Bash

pytest -q -m "not integration"
pytest -q -m integration

Output

............................                                                                 [100%]
28 passed, 3 deselected in 0.17s
...                                                                                          [100%]
3 passed, 28 deselected in 0.87s

In CI you typically run the fast suite on every push and the integration tests on merges to main or nightly, against a test database and provider sandboxes, never production.

Async tests

To test async code directly (or an app from inside an async test), use httpx.AsyncClient with ASGITransport, which calls the app in memory. The AnyIO pytest plugin, installed with FastAPI, runs async def tests marked anyio; pytest-asyncio is the popular alternative.

Python

# file: tests/test_async.py
import httpx
import pytest

from app.main import app


@pytest.fixture
def anyio_backend():
    return "asyncio"


@pytest.mark.anyio
async def test_health_with_an_async_client():
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as ac:
        resp = await ac.get("/health")
    assert resp.json() == {"status": "ok"}

Bash

pytest -q tests/test_async.py

Output

.                                                                                            [100%]
1 passed in 0.01s

Coverage

Coverage measures which lines (and branches) your tests executed. It is good at finding code no test touches, such as an error path, and bad as a target: 100% coverage with weak assertions proves little. The usual tool is the pytest-cov plugin:

Bash

uv add --dev pytest-cov                 # or: pip install pytest-cov
pytest --cov=app --cov-branch --cov-report=term-missing

The report lists each file with its percentage and the line numbers that never ran. Read the missing lines, not the percentage: an untested except branch around a payment or an LLM call is worth a test; a __repr__ is not. Many teams fail CI below a threshold (--cov-fail-under=85) to stop coverage from sliding.

Flaky tests

A flaky test passes and fails without any code change. It is worse than no test: people learn to press "re-run" and stop trusting red builds. The usual causes, and their fixes:

Cause	Fix
Real time: `sleep`, timestamps, "within 1 second"	Inject a clock or a `now` parameter; never assert on wall-clock duration
Real network: an upstream, an LLM, DNS	Mock transport or respx in the fast suite; real calls only in marked integration tests
Shared state: a module-level limiter, cache or dict, rows left by another test	Function-scoped fixtures, rollback per test, reset globals with `monkeypatch`
Order dependence: a test passes only after another one ran	Run in random order (the `pytest-randomly` plugin) to expose it
Randomness: shuffles, sampling, model training	Fixed seeds (`random_state=42`)
Concurrency: background tasks, threads	Wait for completion explicitly; test the logic synchronously

Useful commands while hunting one: pytest -x stops at the first failure, pytest --lf reruns only the last failures, pytest -p no:randomly turns random order off, and pytest --count=50 -k name (pytest-repeat) runs one test many times. Quarantine a test you cannot fix today with an xfail and a ticket; never leave it red.

Load testing basics

Unit and API tests prove correctness for one request. A load test answers capacity questions: how many requests per second before p95 latency passes 500 ms? Do the rate limiter and the connection pool behave under pressure? Does memory grow? Locust describes user behaviour in Python:

Python

# file: locustfile.py
from locust import HttpUser, between, task


class Learner(HttpUser):
    wait_time = between(1, 3)                       # think time between actions

    @task(5)
    def read_review(self):
        self.client.get("/reviews/1", name="/reviews/[id]")

    @task(1)
    def summary(self):
        self.client.post("/courses/1/summary")

Bash

uv run --with locust locust -f locustfile.py --host http://localhost:8000 \
    --users 50 --spawn-rate 5 --run-time 2m --headless

Locust reports requests per second, failures and latency percentiles per endpoint. k6 (scripts in JavaScript) is the other popular choice. Rules that keep load tests honest: run them against a staging copy, not production; point the LLM adapter at a fake (or you load-test your credit card); ramp up gradually; and watch the server's CPU, memory, database connections and error logs, not just the client's numbers.

Real-world use

A typical AI backend team runs thousands of unit and API tests in under a minute on every push, with the LLM, vector store and payment provider faked, and a smaller integration suite against a test database and provider sandboxes before deploy.
Bugs found in production become tests first: reproduce with a failing test, fix, keep the test. Security fixes (BOLA, a missing signature check) always get one.
OpenAPI snapshots or oasdiff run in CI so breaking changes to a public or partner API never ship unnoticed; for AI agents that call your API as a tool, the schema is the interface.
Recorded real responses ("cassettes", with VCR.py or pytest-recording) are a middle ground for complex upstreams: record once, replay in tests, re-record when the upstream changes.

Common mistakes

Tests that call the real LLM or upstream in the fast suite. They are slow, cost money, fail when the network blips and give different answers each run. Inject the provider and use a fake; keep real calls in marked integration tests.
Forgetting app.dependency_overrides.clear(). An override leaks into the next test and makes failures depend on test order. Set and clear overrides inside a fixture with yield.
Shared state between tests, like the module-level review_limiter: test four would hit the limit set by tests one to three. Give each test fresh instances through fixtures.
Asserting only the status code. A 201 with the wrong body, or a retry loop that hit the upstream ten times, still passes. Assert the body, the side effects (rows, calls, prompts) and the counts.
Testing only the happy path. The bugs live in 401s, 404s, timeouts, empty lists and the second user. Write the wrong-user test for every endpoint that touches someone's data.

Try it yourself

Easy: Add a parametrized test that GET /reviews/{id} returns 422 for the ids "abc" and "1.5", and 404 for 12345.
Medium: Write a test that the stream endpoint returns 404, and calls the LLM zero times, when the course has no reviews.
Hard: Add a test that a 500 from the course service is not retried and becomes 502. Run it, and if it fails, decide whether the test or the app is wrong.

Answers

Python

# file: tests/test_exercises.py
import httpx
import pytest

from tests.conftest import NADIA


# 1. Easy
@pytest.mark.parametrize("review_id, status", [("abc", 422), ("1.5", 422), ("12345", 404)])
def test_get_review_ids(client, review_id, status):
    assert client.get(f"/reviews/{review_id}").status_code == status


# 2. Medium
def test_stream_without_reviews(client, llm):
    assert client.get("/courses/8/summary/stream").status_code == 404
    assert llm.prompts == []


# 3. Hard
def test_500_is_not_retried(client, course_service):
    course_service["handler"] = lambda request: httpx.Response(500)
    resp = client.post("/reviews", json={"course_id": 1, "rating": 5, "comment": "Nice"}, headers=NADIA)
    assert (resp.status_code, course_service["calls"]) == (502, 1)

Bash

pytest -q tests/test_exercises.py

Output

.....                                                                                        [100%]
5 passed in 0.03s

The third exercise is the interesting one: fetch_course retries only 503, so a 500 is tried once and becomes 502, and the test passes. If the app retried every 5xx, the test would see three calls and fail, and you would have to decide whether retrying a 500 is safe for that upstream. Tests are how you write such decisions down.

Summary

Most tests should be fast unit and API tests with the outside world faked; a few marked integration tests touch real services.
Design for testability: every outside dependency comes through Depends, and every clock is injectable. Tests swap them with app.dependency_overrides, MockTransport, respx and fakes.
Fixtures with the right scope, factory fixtures and parametrize keep tests short; a rolled-back transaction per test keeps the database clean.
Test the risky paths: wrong user, limits, retries counted by calls, timeouts, streams, signatures and replay.
Guard your API contract with an OpenAPI snapshot; use coverage to find untested code, not as a goal; fix flaky tests at the root; load-test staging with a fake LLM.

Next: Serving an ML Model puts a trained scikit-learn model behind a FastAPI endpoint, with validated features, versioning, warm-up and tests that use a tiny model.

Continue learning

Watch it taught, then check what stuck.

Test yourselfGit, APIs and Backend Advanced — Quiz 2: Production FastAPI
Free15 minTake quiz

Git, APIs and Backend for AI: Advanced

100%

Page 13 of 22 · 289 min left

Chapter 3 · Production FastAPI

Testing Professional APIs

Page 13 of 22
26 min read

What you will learn

The testing pyramid, unit vs integration tests, and markers to run them separately
Fixtures with scopes, factory fixtures, parametrize, and app.dependency_overrides
Mocking external HTTP (httpx.MockTransport and respx) and LLM providers; database tests that roll back
Testing auth, rate limits, retries, timeouts, streaming responses and webhook signatures
Contract tests with an OpenAPI snapshot, coverage, flaky tests, load testing and security regression tests

The testing pyramid

Output

                 ▲  slow, brittle, expensive
                ╱ ╲
               ╱E2E╲          a few: the deployed system, real browser or client
              ╱─────╲
             ╱ integ-╲        some: app + real DB / real HTTP / real provider sandbox
            ╱ ration  ╲
           ╱───────────╲
          ╱  API tests  ╲     many: the whole app through TestClient, outside world faked
         ╱───────────────╲
        ╱   unit tests    ╲   most: pure functions and classes, no I/O at all
       ╱───────────────────╲
                 ▼  fast, stable, cheap

The service under test

Bash

mkdir -p ~/projects/course-reviews && cd ~/projects/course-reviews
uv init -q --bare --name course-reviews
uv add -q "fastapi[standard]" sqlalchemy
uv add -q --dev pytest httpx2 respx
source .venv/bin/activate
mkdir -p app tests && touch app/__init__.py tests/__init__.py
# pip instead of uv:
#   python -m venv .venv && source .venv/bin/activate
#   pip install -q "fastapi[standard]" sqlalchemy pytest httpx2 respx

Python

# file: app/main.py
"""Course Reviews API: the service this page tests."""
import hashlib
import hmac
import json
import os
import time
from collections.abc import Iterator
from contextlib import asynccontextmanager
from typing import Annotated, Protocol

import httpx
from fastapi import Depends, FastAPI, Header, HTTPException, Request
from fastapi.sse import EventSourceResponse, ServerSentEvent
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import String, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column

PRACTICE_API_URL = os.environ.get("PRACTICE_API_URL", "http://localhost:8800")
DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///reviews.db")
WEBHOOK_SECRET = os.environ.get("WEBHOOK_SECRET", "dev-only-secret").encode()
TOKENS = {"token-nadia": "nadia", "token-tanvir": "tanvir", "token-admin": "admin"}  # demo only
ADMINS = {"admin"}


class Base(DeclarativeBase):
    pass


class Review(Base):
    __tablename__ = "reviews"
    id: Mapped[int] = mapped_column(primary_key=True)
    author: Mapped[str] = mapped_column(String(50))
    course_id: Mapped[int]
    rating: Mapped[int]
    comment: Mapped[str] = mapped_column(String(500))


class ReviewIn(BaseModel):
    course_id: int = Field(ge=1)
    rating: int = Field(ge=1, le=5)
    comment: str = Field(min_length=3, max_length=500)


class ReviewOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    author: str
    course_id: int
    rating: int
    comment: str


engine = create_engine(DATABASE_URL)


@asynccontextmanager
async def lifespan(app: FastAPI):
    Base.metadata.create_all(engine)
    app.state.http = httpx.Client(base_url=PRACTICE_API_URL, timeout=httpx.Timeout(3.0, connect=1.0))
    yield
    app.state.http.close()


app = FastAPI(title="Course Reviews", lifespan=lifespan)


# ---------- dependencies: each one is a seam that tests can replace
def get_session() -> Iterator[Session]:
    with Session(engine) as session:
        yield session


def get_http(request: Request) -> httpx.Client:
    return request.app.state.http


class LLM(Protocol):
    def complete(self, prompt: str) -> str: ...
    def stream(self, prompt: str) -> Iterator[str]: ...


class PracticeLLM:
    """Calls the Practice API's fake LLM; a real provider adapter would sit here."""

    def __init__(self, http: httpx.Client):
        self.http = http

    def _request(self, prompt: str, stream: bool) -> dict:
        return {"headers": {"Authorization": "Bearer token-nadia-001"}, "timeout": 30,
                "json": {"messages": [{"role": "user", "content": prompt}], "stream": stream}}

    def complete(self, prompt: str) -> str:
        resp = self.http.post("/v1/chat", **self._request(prompt, stream=False))
        resp.raise_for_status()
        return resp.json()["message"]["content"]

    def stream(self, prompt: str) -> Iterator[str]:
        with self.http.stream("POST", "/v1/chat", **self._request(prompt, stream=True)) as resp:
            resp.raise_for_status()
            for line in resp.iter_lines():
                if line.startswith("data: ") and "delta" in (event := json.loads(line[6:])):
                    yield event["delta"]


def get_llm(http: Annotated[httpx.Client, Depends(get_http)]) -> LLM:
    return PracticeLLM(http)


class RateLimiter:
    """Sliding-window limiter kept in memory (use Redis in production, as on the previous page)."""

    def __init__(self, limit: int, window: float, clock=time.monotonic):
        self.limit, self.window, self.clock = limit, window, clock
        self.hits: dict[str, list[float]] = {}

    def allow(self, key: str) -> bool:
        now = self.clock()
        recent = [t for t in self.hits.get(key, []) if now - t < self.window]
        allowed = len(recent) < self.limit
        if allowed:
            recent.append(now)
        self.hits[key] = recent
        return allowed


review_limiter = RateLimiter(limit=3, window=60)


def get_limiter() -> RateLimiter:
    return review_limiter


def current_user(authorization: Annotated[str | None, Header()] = None) -> str:
    scheme, _, token = (authorization or "").partition(" ")
    user = TOKENS.get(token) if scheme.lower() == "bearer" else None
    if user is None:
        raise HTTPException(401, "Not authenticated", headers={"WWW-Authenticate": "Bearer"})
    return user


SessionDep = Annotated[Session, Depends(get_session)]
HttpDep = Annotated[httpx.Client, Depends(get_http)]
LLMDep = Annotated[LLM, Depends(get_llm)]
LimiterDep = Annotated[RateLimiter, Depends(get_limiter)]
UserDep = Annotated[str, Depends(current_user)]


# ---------- business logic
def fetch_course(http: httpx.Client, course_id: int, attempts: int = 3) -> dict | None:
    """Look a course up in the course service, retrying 503s."""
    for attempt in range(1, attempts + 1):
        try:
            resp = http.get(f"/v1/courses/{course_id}")
        except httpx.TimeoutException:
            raise HTTPException(504, "Course service timed out")
        if resp.status_code == 404:
            return None
        if resp.status_code == 503 and attempt < attempts:
            continue                       # a real client also backs off here
        if resp.status_code != 200:
            raise HTTPException(502, "Course service error")
        return resp.json()


def verify_signature(secret: bytes, timestamp: str, body: bytes, signature: str,
                     now: float, tolerance: int = 300) -> bool:
    if not timestamp.isdigit() or abs(now - int(timestamp)) > tolerance:
        return False                       # missing, malformed or too old: replay protection
    expected = hmac.new(secret, timestamp.encode() + b"." + body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)


# ---------- routes
@app.get("/health")
def health() -> dict:
    return {"status": "ok"}


@app.post("/reviews", status_code=201)
def create_review(review: ReviewIn, user: UserDep, session: SessionDep,
                  http: HttpDep, limiter: LimiterDep) -> ReviewOut:
    if not limiter.allow(user):
        raise HTTPException(429, "Too many reviews, slow down", headers={"Retry-After": "60"})
    if fetch_course(http, review.course_id) is None:
        raise HTTPException(422, f"Course {review.course_id} does not exist")
    row = Review(author=user, **review.model_dump())
    session.add(row)
    session.commit()
    return ReviewOut.model_validate(row)


@app.get("/reviews/{review_id}")
def get_review(review_id: int, session: SessionDep) -> ReviewOut:
    row = session.get(Review, review_id)
    if row is None:
        raise HTTPException(404, "Review not found")
    return ReviewOut.model_validate(row)


@app.delete("/reviews/{review_id}", status_code=204)
def delete_review(review_id: int, user: UserDep, session: SessionDep) -> None:
    row = session.get(Review, review_id)
    if row is None or (row.author != user and user not in ADMINS):
        raise HTTPException(404, "Review not found")     # don't reveal other people's reviews
    session.delete(row)
    session.commit()


def course_prompt(course_id: int, session: SessionDep) -> tuple[str, int]:
    """A dependency, so a 404 is raised before any streaming starts."""
    comments = session.scalars(select(Review.comment).where(Review.course_id == course_id)).all()
    if not comments:
        raise HTTPException(404, "No reviews for this course yet")
    return "Summarise these course reviews in one sentence:\n" + "\n".join(comments), len(comments)


PromptDep = Annotated[tuple[str, int], Depends(course_prompt)]


@app.post("/courses/{course_id}/summary")
def summarise(course_id: int, prompt: PromptDep, llm: LLMDep) -> dict:
    text, count = prompt
    return {"course_id": course_id, "reviews": count, "summary": llm.complete(text)}


@app.get("/courses/{course_id}/summary/stream", response_class=EventSourceResponse)
def summarise_stream(prompt: PromptDep, llm: LLMDep) -> Iterator[ServerSentEvent]:
    text, count = prompt
    for delta in llm.stream(text):
        yield ServerSentEvent(data={"delta": delta})
    yield ServerSentEvent(event="done", data={"reviews": count})


@app.post("/webhooks/payments")
async def payment_webhook(request: Request, x_timestamp: Annotated[str, Header()],
                          x_signature: Annotated[str, Header()]) -> dict:
    body = await request.body()
    if not verify_signature(WEBHOOK_SECRET, x_timestamp, body, x_signature, time.time()):
        raise HTTPException(401, "Invalid signature")
    return {"received": json.loads(body)["event"]}

Tell pytest where the tests are and declare the one custom marker the suite uses (with --strict-markers, a typo in a marker name is an error instead of a silently ignored test):

Bash

cat >> pyproject.toml <<'EOF'

[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "--strict-markers"
markers = ["integration: talks to real services (deselect with -m 'not integration')"]
EOF
export PRACTICE_API_URL=http://localhost:8800

Fixtures: the test toolkit

All shared setup lives in tests/conftest.py, which pytest loads automatically. Each fixture replaces one seam:

Python

# file: tests/conftest.py
"""Shared fixtures: a rolled-back database, fake HTTP, a fake LLM and a fresh rate limiter."""
import httpx
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, event
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool

from app import main
from app.main import Base, RateLimiter, Review, app

COURSES = {1: {"id": 1, "title": "How AI Works"}, 8: {"id": 8, "title": "RAG Systems"}}


class FakeLLM:
    """Stands in for the LLM provider: instant, free, predictable, and it records prompts."""

    def __init__(self):
        self.prompts = []

    def complete(self, prompt):
        self.prompts.append(prompt)
        return "Learners find it clear."

    def stream(self, prompt):
        self.prompts.append(prompt)
        yield from ["Learners ", "find ", "it ", "clear."]


def fake_course_service(request: httpx.Request) -> httpx.Response:
    course_id = int(request.url.path.rsplit("/", 1)[-1])
    if course_id in COURSES:
        return httpx.Response(200, json=COURSES[course_id])
    return httpx.Response(404, json={"detail": "Course not found"})


@pytest.fixture(scope="session")
def engine():
    """One in-memory database for the whole test run."""
    engine = create_engine("sqlite://", poolclass=StaticPool,
                           connect_args={"check_same_thread": False})

    @event.listens_for(engine, "connect")
    def _no_driver_transactions(dbapi_connection, record):
        dbapi_connection.isolation_level = None    # let SQLAlchemy emit BEGIN/SAVEPOINT itself

    @event.listens_for(engine, "begin")
    def _begin(connection):
        connection.exec_driver_sql("BEGIN")

    Base.metadata.create_all(engine)
    yield engine
    engine.dispose()


@pytest.fixture
def db_session(engine):
    """Each test runs inside a transaction that is rolled back afterwards."""
    with engine.connect() as connection:
        transaction = connection.begin()
        session = Session(bind=connection, join_transaction_mode="create_savepoint")
        yield session
        session.close()
        transaction.rollback()


@pytest.fixture
def llm():
    return FakeLLM()


@pytest.fixture
def course_service():
    """What the fake course service does; a test may swap the handler."""
    return {"handler": fake_course_service, "calls": 0}


@pytest.fixture
def client(db_session, llm, course_service):
    def transport_handler(request):
        course_service["calls"] += 1
        return course_service["handler"](request)

    http = httpx.Client(base_url="http://courses.test", transport=httpx.MockTransport(transport_handler))
    limiter = RateLimiter(limit=3, window=60)       # a fresh limiter per test
    app.dependency_overrides[main.get_session] = lambda: db_session
    app.dependency_overrides[main.get_http] = lambda: http
    app.dependency_overrides[main.get_llm] = lambda: llm
    app.dependency_overrides[main.get_limiter] = lambda: limiter
    yield TestClient(app)
    app.dependency_overrides.clear()
    http.close()


@pytest.fixture
def make_review(db_session):
    """A factory fixture: call it to create reviews with sensible defaults."""
    def make(author="nadia", course_id=1, rating=5, comment="Very clear examples"):
        review = Review(author=author, course_id=course_id, rating=rating, comment=comment)
        db_session.add(review)
        db_session.commit()
        return review
    return make


NADIA = {"Authorization": "Bearer token-nadia"}
TANVIR = {"Authorization": "Bearer token-tanvir"}
ADMIN = {"Authorization": "Bearer token-admin"}

The ideas in it, one by one:

Scopes. engine has scope="session": creating the database and tables once per run is fast. Everything else is function-scoped (the default): a fresh fake LLM, fake course service and limiter for every test, so no test sees another's leftovers. Other scopes are module, class and package.
Rollback per test. db_session opens a transaction, and the session inside it turns every commit() into a savepoint (join_transaction_mode="create_savepoint"). At the end the outer transaction is rolled back, so the next test starts with empty tables, and no time is spent recreating them. SQLite's Python driver manages transactions in its own way, so the two event hooks hand that job to SQLAlchemy; PostgreSQL needs no hooks.
app.dependency_overrides maps a dependency function to a replacement. FastAPI calls the replacement everywhere the original was used, including inside other dependencies. Clear the overrides after each test.
httpx.MockTransport is a fake network: the client works exactly as usual, but every request is answered by your function instead of a server. No sockets, no flakiness.
A fake LLM with the same methods as the real adapter (the LLM protocol). It answers instantly, costs nothing, and records the prompts, so tests can check what would have been sent.
A factory fixture (make_review) returns a function, so a test can create as many rows as it needs with only the fields that matter to it.

API tests and parametrize

Python

# file: tests/test_reviews.py
import pytest

from tests.conftest import NADIA

GOOD = {"course_id": 1, "rating": 5, "comment": "Very clear examples"}


def test_create_review(client):
    resp = client.post("/reviews", json=GOOD, headers=NADIA)
    assert resp.status_code == 201
    assert resp.json() == {"id": 1, "author": "nadia", **GOOD}


def test_database_is_clean_for_every_test(client):
    assert client.get("/reviews/1").status_code == 404


@pytest.mark.parametrize("field, value", [
    ("rating", 0), ("rating", 6), ("comment", "ok"), ("course_id", 0), ("comment", "x" * 501),
])
def test_invalid_reviews_are_rejected(client, field, value):
    resp = client.post("/reviews", json={**GOOD, field: value}, headers=NADIA)
    assert resp.status_code == 422
    assert resp.json()["detail"][0]["loc"] == ["body", field]


def test_unknown_course_is_rejected(client):
    resp = client.post("/reviews", json={**GOOD, "course_id": 99}, headers=NADIA)
    assert resp.status_code == 422
    assert resp.json() == {"detail": "Course 99 does not exist"}

Bash

pytest -q

Output

........                                                                                     [100%]
8 passed in 0.04s

Bash

pytest -q --setup-show tests/test_reviews.py::test_create_review

Output


SETUP    S engine
        SETUP    F db_session (fixtures used: engine)
        SETUP    F llm
        SETUP    F course_service
        SETUP    F client (fixtures used: course_service, db_session, llm)
        tests/test_reviews.py::test_create_review (fixtures used: client, course_service, db_session, engine, llm) .
        TEARDOWN F client
        TEARDOWN F course_service
        TEARDOWN F llm
        TEARDOWN F db_session
TEARDOWN S engine
1 passed in 0.02s

The session-scoped engine (S) wraps everything; the function-scoped fixtures (F) are created for this one test and torn down in reverse order.

Testing auth and permissions

Python

# file: tests/test_auth.py
import pytest

from tests.conftest import ADMIN, NADIA, TANVIR


@pytest.mark.parametrize("headers", [{}, {"Authorization": "Bearer wrong"}, {"Authorization": "token-nadia"}])
def test_writing_needs_a_valid_token(client, headers):
    resp = client.post("/reviews", json={"course_id": 1, "rating": 5, "comment": "Nice"}, headers=headers)
    assert resp.status_code == 401
    assert resp.headers["WWW-Authenticate"] == "Bearer"


def test_owner_can_delete(client, make_review):
    review = make_review(author="nadia")
    assert client.delete(f"/reviews/{review.id}", headers=NADIA).status_code == 204
    assert client.get(f"/reviews/{review.id}").status_code == 404


def test_other_users_cannot_delete(client, make_review):
    """Security regression test for BOLA: Tanvir must not delete Nadia's review."""
    review = make_review(author="nadia")
    assert client.delete(f"/reviews/{review.id}", headers=TANVIR).status_code == 404
    assert client.get(f"/reviews/{review.id}").status_code == 200


def test_admin_can_delete_any_review(client, make_review):
    review = make_review(author="tanvir")
    assert client.delete(f"/reviews/{review.id}", headers=ADMIN).status_code == 204

Rate limits, retries and timeouts

Python

# file: tests/test_limits_and_upstream.py
import httpx

from app.main import RateLimiter
from tests.conftest import NADIA, TANVIR

GOOD = {"course_id": 1, "rating": 4, "comment": "Good pace"}


def test_fourth_review_in_a_minute_is_limited(client):
    codes = [client.post("/reviews", json=GOOD, headers=NADIA).status_code for _ in range(4)]
    assert codes == [201, 201, 201, 429]
    assert client.post("/reviews", json=GOOD, headers=TANVIR).status_code == 201


def test_limiter_window_slides():
    now = [0.0]
    limiter = RateLimiter(limit=2, window=60, clock=lambda: now[0])
    assert [limiter.allow("nadia") for _ in range(3)] == [True, True, False]
    now[0] = 60.0                                   # time passes without sleeping
    assert limiter.allow("nadia") is True


def test_course_service_503s_are_retried(client, course_service):
    replies = iter([httpx.Response(503), httpx.Response(503), httpx.Response(200, json={"id": 1})])
    course_service["handler"] = lambda request: next(replies)
    assert client.post("/reviews", json=GOOD, headers=NADIA).status_code == 201
    assert course_service["calls"] == 3


def test_course_service_keeps_failing(client, course_service):
    course_service["handler"] = lambda request: httpx.Response(503)
    resp = client.post("/reviews", json=GOOD, headers=NADIA)
    assert (resp.status_code, course_service["calls"]) == (502, 3)


def test_course_service_timeout_becomes_504(client, course_service):
    def hang(request):
        raise httpx.ReadTimeout("timed out", request=request)

    course_service["handler"] = hang
    resp = client.post("/reviews", json=GOOD, headers=NADIA)
    assert resp.status_code == 504
    assert course_service["calls"] == 1             # timeouts are not retried here

Mocking the LLM provider, and streaming

Python

# file: tests/test_llm.py
from tests.conftest import NADIA


def test_summary_uses_the_reviews(client, llm, make_review):
    make_review(comment="Very clear examples")
    make_review(comment="Too fast at the end", rating=3)
    resp = client.post("/courses/1/summary")
    assert resp.json() == {"course_id": 1, "reviews": 2, "summary": "Learners find it clear."}
    assert "Too fast at the end" in llm.prompts[0]


def test_no_reviews_means_404_and_no_llm_call(client, llm):
    assert client.post("/courses/8/summary").status_code == 404
    assert llm.prompts == []                        # no tokens spent


def test_summary_stream(client, make_review):
    make_review()
    with client.stream("GET", "/courses/1/summary/stream") as resp:
        assert resp.headers["content-type"].startswith("text/event-stream")
        lines = [line for line in resp.iter_lines() if line]
    assert lines == [
        'data: {"delta": "Learners "}', 'data: {"delta": "find "}',
        'data: {"delta": "it "}', 'data: {"delta": "clear."}',
        "event: done", 'data: {"reviews": 1}',
    ]

Webhook signatures

Python

# file: tests/test_webhooks.py
import hashlib
import hmac
import json
import time

import pytest

from app.main import WEBHOOK_SECRET, verify_signature


def signed(body: dict, secret: bytes = WEBHOOK_SECRET, timestamp: int | None = None):
    raw = json.dumps(body).encode()
    ts = str(timestamp if timestamp is not None else int(time.time()))
    sig = hmac.new(secret, ts.encode() + b"." + raw, hashlib.sha256).hexdigest()
    return raw, {"X-Timestamp": ts, "X-Signature": sig, "Content-Type": "application/json"}


def test_valid_webhook_is_accepted(client):
    raw, headers = signed({"event": "payment.succeeded"})
    resp = client.post("/webhooks/payments", content=raw, headers=headers)
    assert resp.json() == {"received": "payment.succeeded"}


@pytest.mark.parametrize("case", ["wrong secret", "tampered body", "replayed"])
def test_bad_webhooks_are_rejected(client, case):
    if case == "wrong secret":
        raw, headers = signed({"event": "payment.succeeded"}, secret=b"guess")
    elif case == "tampered body":
        _, headers = signed({"event": "payment.succeeded", "amount": 500})
        raw = json.dumps({"event": "payment.succeeded", "amount": 5}).encode()
    else:
        raw, headers = signed({"event": "payment.succeeded"}, timestamp=int(time.time()) - 3600)
    assert client.post("/webhooks/payments", content=raw, headers=headers).status_code == 401


def test_signature_check_is_a_pure_function():
    """A unit test: no app, no HTTP, a fixed clock."""
    raw = b'{"event": "x"}'
    sig = hmac.new(b"k", b"1000." + raw, hashlib.sha256).hexdigest()
    assert verify_signature(b"k", "1000", raw, sig, now=1100)
    assert not verify_signature(b"k", "1000", raw, sig, now=1400)     # 400 s later: too old
    assert not verify_signature(b"k", "abc", raw, sig, now=1000)

Bash

pytest -q

Output

...........................                                                                  [100%]
27 passed in 0.11s

Contract tests: an OpenAPI snapshot

Python

# file: tests/test_contract.py
"""Contract test: the OpenAPI schema is the promise you made to clients."""
import json
import os
from pathlib import Path

from app.main import app

SNAPSHOT = Path(__file__).parent / "openapi.snapshot.json"


def test_openapi_matches_snapshot():
    current = app.openapi()
    if os.environ.get("UPDATE_SNAPSHOT") == "1" or not SNAPSHOT.exists():
        SNAPSHOT.write_text(json.dumps(current, indent=2, sort_keys=True) + "\n")
    saved = json.loads(SNAPSHOT.read_text())
    assert sorted(current["paths"]) == sorted(saved["paths"]), "endpoints were added or removed"
    for name, schema in saved["components"]["schemas"].items():
        now = current["components"]["schemas"].get(name, {})
        assert now.get("properties") == schema.get("properties"), f"schema {name} changed"
        assert now.get("required") == schema.get("required"), f"required fields of {name} changed"

Bash

pytest -q tests/test_contract.py
ls tests/openapi.snapshot.json

Output

.                                                                                            [100%]
1 passed in 0.01s
tests/openapi.snapshot.json

Now a teammate renames comment to text in the request model, a breaking change for every client that sends comment:

Bash

sed -i 's/    comment: str = Field(min_length=3, max_length=500)/    text: str = Field(min_length=3, max_length=500)/' app/main.py
pytest -q --tb=short tests/test_contract.py

Output

F                                                                                            [100%]
============================================= FAILURES =============================================
__________________________________ test_openapi_matches_snapshot ___________________________________
tests/test_contract.py:20: in test_openapi_matches_snapshot
    assert now.get("properties") == schema.get("properties"), f"schema {name} changed"
E   AssertionError: schema ReviewIn changed
E   assert {'course_id':...tle': 'Text'}} == {'comment': {...': 'integer'}}
E     
E     Omitting 2 identical items, use -vv to show
E     Left contains 1 more item:
E     {'text': {'maxLength': 500, 'minLength': 3, 'title': 'Text', 'type': 'string'}}
E     Right contains 1 more item:
E     {'comment': {'maxLength': 500,
E                  'minLength': 3,...
E     
E     ...Full output truncated (3 lines hidden), use '-vv' to show
===================================== short test summary info ======================================
FAILED tests/test_contract.py::test_openapi_matches_snapshot - AssertionError: schema ReviewIn ch...
1 failed in 0.04s

Bash

sed -i 's/    text: str = Field(min_length=3, max_length=500)/    comment: str = Field(min_length=3, max_length=500)/' app/main.py
pytest -q tests/test_contract.py

Output

.                                                                                            [100%]
1 passed in 0.01s

Integration tests, respx and markers

Some tests should touch the real thing: the real course service, the real adapter's streaming code, the full app with its lifespan. Mark them, so the fast suite can leave them out:

Python

# file: tests/test_integration.py
"""Tests that talk to real services. Slower; run them on purpose with -m integration."""
import httpx
import pytest
import respx
from fastapi.testclient import TestClient

from app import main

pytestmark = pytest.mark.integration


def test_real_course_service_lookup():
    with httpx.Client(base_url=main.PRACTICE_API_URL, timeout=5) as http:
        assert main.fetch_course(http, 2)["title"] == "Python for AI"
        assert main.fetch_course(http, 999) is None


def test_practice_llm_adapter_streams():
    with httpx.Client(base_url=main.PRACTICE_API_URL, timeout=5) as http:
        text = "".join(main.PracticeLLM(http).stream("What is MLOps?"))
    assert text.startswith("This is a practice answer about: What is MLOps.")


@respx.mock
def test_app_with_respx_and_lifespan(tmp_path, monkeypatch):
    """The whole app, real lifespan and real httpx client; respx answers the HTTP calls."""
    monkeypatch.setattr(main, "engine", main.create_engine(f"sqlite:///{tmp_path}/test.db"))
    route = respx.get(f"{main.PRACTICE_API_URL}/v1/courses/1").mock(
        return_value=httpx.Response(200, json={"id": 1}))
    with TestClient(main.app) as client:            # "with" runs the lifespan
        resp = client.post("/reviews", json={"course_id": 1, "rating": 5, "comment": "Great"},
                           headers={"Authorization": "Bearer token-nadia"})
    assert resp.status_code == 201
    assert route.call_count == 1

Bash

pytest -q -m "not integration"
pytest -q -m integration

Output

............................                                                                 [100%]
28 passed, 3 deselected in 0.17s
...                                                                                          [100%]
3 passed, 28 deselected in 0.87s

In CI you typically run the fast suite on every push and the integration tests on merges to main or nightly, against a test database and provider sandboxes, never production.

Async tests

Python

# file: tests/test_async.py
import httpx
import pytest

from app.main import app


@pytest.fixture
def anyio_backend():
    return "asyncio"


@pytest.mark.anyio
async def test_health_with_an_async_client():
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as ac:
        resp = await ac.get("/health")
    assert resp.json() == {"status": "ok"}

Bash

pytest -q tests/test_async.py

Output

.                                                                                            [100%]
1 passed in 0.01s

Coverage

Bash

uv add --dev pytest-cov                 # or: pip install pytest-cov
pytest --cov=app --cov-branch --cov-report=term-missing

Flaky tests

A flaky test passes and fails without any code change. It is worse than no test: people learn to press "re-run" and stop trusting red builds. The usual causes, and their fixes:

Cause	Fix
Real time: `sleep`, timestamps, "within 1 second"	Inject a clock or a `now` parameter; never assert on wall-clock duration
Real network: an upstream, an LLM, DNS	Mock transport or respx in the fast suite; real calls only in marked integration tests
Shared state: a module-level limiter, cache or dict, rows left by another test	Function-scoped fixtures, rollback per test, reset globals with `monkeypatch`
Order dependence: a test passes only after another one ran	Run in random order (the `pytest-randomly` plugin) to expose it
Randomness: shuffles, sampling, model training	Fixed seeds (`random_state=42`)
Concurrency: background tasks, threads	Wait for completion explicitly; test the logic synchronously

Load testing basics

Python

# file: locustfile.py
from locust import HttpUser, between, task


class Learner(HttpUser):
    wait_time = between(1, 3)                       # think time between actions

    @task(5)
    def read_review(self):
        self.client.get("/reviews/1", name="/reviews/[id]")

    @task(1)
    def summary(self):
        self.client.post("/courses/1/summary")

Bash

uv run --with locust locust -f locustfile.py --host http://localhost:8000 \
    --users 50 --spawn-rate 5 --run-time 2m --headless

Real-world use

A typical AI backend team runs thousands of unit and API tests in under a minute on every push, with the LLM, vector store and payment provider faked, and a smaller integration suite against a test database and provider sandboxes before deploy.
Bugs found in production become tests first: reproduce with a failing test, fix, keep the test. Security fixes (BOLA, a missing signature check) always get one.
OpenAPI snapshots or oasdiff run in CI so breaking changes to a public or partner API never ship unnoticed; for AI agents that call your API as a tool, the schema is the interface.
Recorded real responses ("cassettes", with VCR.py or pytest-recording) are a middle ground for complex upstreams: record once, replay in tests, re-record when the upstream changes.

Common mistakes

Tests that call the real LLM or upstream in the fast suite. They are slow, cost money, fail when the network blips and give different answers each run. Inject the provider and use a fake; keep real calls in marked integration tests.
Forgetting app.dependency_overrides.clear(). An override leaks into the next test and makes failures depend on test order. Set and clear overrides inside a fixture with yield.
Shared state between tests, like the module-level review_limiter: test four would hit the limit set by tests one to three. Give each test fresh instances through fixtures.
Asserting only the status code. A 201 with the wrong body, or a retry loop that hit the upstream ten times, still passes. Assert the body, the side effects (rows, calls, prompts) and the counts.
Testing only the happy path. The bugs live in 401s, 404s, timeouts, empty lists and the second user. Write the wrong-user test for every endpoint that touches someone's data.

Try it yourself

Easy: Add a parametrized test that GET /reviews/{id} returns 422 for the ids "abc" and "1.5", and 404 for 12345.
Medium: Write a test that the stream endpoint returns 404, and calls the LLM zero times, when the course has no reviews.
Hard: Add a test that a 500 from the course service is not retried and becomes 502. Run it, and if it fails, decide whether the test or the app is wrong.

Answers

Python

# file: tests/test_exercises.py
import httpx
import pytest

from tests.conftest import NADIA


# 1. Easy
@pytest.mark.parametrize("review_id, status", [("abc", 422), ("1.5", 422), ("12345", 404)])
def test_get_review_ids(client, review_id, status):
    assert client.get(f"/reviews/{review_id}").status_code == status


# 2. Medium
def test_stream_without_reviews(client, llm):
    assert client.get("/courses/8/summary/stream").status_code == 404
    assert llm.prompts == []


# 3. Hard
def test_500_is_not_retried(client, course_service):
    course_service["handler"] = lambda request: httpx.Response(500)
    resp = client.post("/reviews", json={"course_id": 1, "rating": 5, "comment": "Nice"}, headers=NADIA)
    assert (resp.status_code, course_service["calls"]) == (502, 1)

Bash

pytest -q tests/test_exercises.py

Output

.....                                                                                        [100%]
5 passed in 0.03s

Summary

Most tests should be fast unit and API tests with the outside world faked; a few marked integration tests touch real services.
Design for testability: every outside dependency comes through Depends, and every clock is injectable. Tests swap them with app.dependency_overrides, MockTransport, respx and fakes.
Fixtures with the right scope, factory fixtures and parametrize keep tests short; a rolled-back transaction per test keeps the database clean.
Test the risky paths: wrong user, limits, retries counted by calls, timeouts, streams, signatures and replay.
Guard your API contract with an OpenAPI snapshot; use coverage to find untested code, not as a goal; fix flaky tests at the root; load-test staging with a fake LLM.

Next: Serving an ML Model puts a trained scikit-learn model behind a FastAPI endpoint, with validated features, versioning, warm-up and tests that use a tiny model.

Continue learning

Watch it taught, then check what stuck.

Test yourselfGit, APIs and Backend Advanced — Quiz 2: Production FastAPI
Free15 minTake quiz