FastAPI: Testing FastAPI apps

FastAPI is unusually easy to test: the app is a callable, clients run it in-process, and dependency overrides replace any collaborator. The hard part is not the tools but the design of the suite. Each test gets a fresh app and database, the lifespan runs, authentication is faked at the right seam, async code is tested asynchronously, and the API contract itself is pinned. This part writes such a suite and runs it with pytest.

1. The tools

2. A complete suite, run for real

This block writes a small app (async SQLAlchemy, a bearer-token dependency, a WebSocket) and a test suite, then runs pytest on it:

app.py + conftest.py + test_items.py, then pytest

import subprocess
import sys
from pathlib import Path

Path("app.py").write_text('''
from contextlib import asynccontextmanager
from typing import Annotated
from fastapi import Depends, FastAPI, Header, HTTPException, Request, WebSocket
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import String, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.pool import StaticPool

class Base(DeclarativeBase): pass

class Item(Base):
    __tablename__ = "items"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    owner_id: Mapped[int]

class ItemIn(BaseModel):
    model_config = ConfigDict(extra="forbid")
    name: str = Field(min_length=1, max_length=50)

class ItemOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    name: str
    owner_id: int

def create_app(database_url: str = "sqlite+aiosqlite://") -> FastAPI:
    @asynccontextmanager
    async def lifespan(app):
        engine = create_async_engine(database_url, poolclass=StaticPool)   # one shared in-memory DB
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
        yield {"sessions": async_sessionmaker(engine, expire_on_commit=False)}
        await engine.dispose()

    app = FastAPI(lifespan=lifespan)

    async def get_session(request: Request):
        async with request.state.sessions() as s:
            yield s

    async def current_user(authorization: Annotated[str | None, Header()] = None) -> int:
        raise HTTPException(401, "real token validation lives here")       # replaced in tests

    app.state.current_user = current_user
    Session = Annotated[AsyncSession, Depends(get_session)]
    User = Annotated[int, Depends(current_user)]

    @app.post("/items", status_code=201)
    async def create(body: ItemIn, session: Session, user: User) -> ItemOut:
        item = Item(name=body.name, owner_id=user)
        session.add(item)
        await session.commit()
        return item

    @app.get("/items/{item_id}")
    async def read(item_id: int, session: Session, user: User) -> ItemOut:
        item = await session.get(Item, item_id)
        if item is None or item.owner_id != user:
            raise HTTPException(404, "not found")
        return item

    @app.websocket("/ws/echo")
    async def echo(ws: WebSocket):
        await ws.accept()
        await ws.send_text((await ws.receive_text()).upper())
        await ws.close()
    return app
''', encoding="utf-8")

Path("conftest.py").write_text('''
import httpx
import pytest
from fastapi.testclient import TestClient
from app import create_app

@pytest.fixture
def app():
    return create_app()                        # fresh app + fresh in-memory database per test

@pytest.fixture
def as_user(app):
    def login(user_id: int):
        app.dependency_overrides[app.state.current_user] = lambda: user_id
    yield login
    app.dependency_overrides.clear()           # never leak overrides into the next test

@pytest.fixture
def client(app):
    with TestClient(app) as c:                 # "with" runs the lifespan
        yield c

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

@pytest.fixture
async def aclient(app):
    async with app.router.lifespan_context(app) as state:     # ASGITransport skips lifespan: run it ourselves
        async def with_state(scope, receive, send):           # ...and hand its state to every request
            scope["state"] = dict(state)
            await app(scope, receive, send)
        transport = httpx.ASGITransport(app=with_state)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
            yield c
''', encoding="utf-8")

Path("test_items.py").write_text('''
import json
import pytest

def test_requires_authentication(client):
    assert client.post("/items", json={"name": "tea"}).status_code == 401

def test_create_then_read(client, as_user):
    as_user(7)
    created = client.post("/items", json={"name": "tea"})
    assert created.status_code == 201
    assert client.get(f"/items/{created.json()['id']}").json() == {"id": 1, "name": "tea", "owner_id": 7}

def test_other_users_items_are_invisible(client, as_user):
    as_user(7)
    item_id = client.post("/items", json={"name": "tea"}).json()["id"]
    as_user(9)
    assert client.get(f"/items/{item_id}").status_code == 404

@pytest.mark.parametrize("payload, field", [({"name": ""}, "name"), ({}, "name"),
                                            ({"name": "x", "owner_id": 1}, "owner_id")])
def test_validation(client, as_user, payload, field):
    as_user(7)
    r = client.post("/items", json=payload)
    assert r.status_code == 422 and r.json()["detail"][0]["loc"][-1] == field

def test_each_test_gets_a_fresh_database(client, as_user):
    as_user(7)
    assert client.post("/items", json={"name": "first"}).json()["id"] == 1

def test_websocket(client):
    with client.websocket_connect("/ws/echo") as ws:
        ws.send_text("hello")
        assert ws.receive_text() == "HELLO"

def test_openapi_contract(client):
    paths = client.get("/openapi.json").json()["paths"]
    assert sorted(paths) == ["/items", "/items/{item_id}"]
    assert sorted(paths["/items"]["post"]["responses"]) == ["201", "422"]

@pytest.mark.anyio
async def test_async_client(app, aclient):
    app.dependency_overrides[app.state.current_user] = lambda: 3
    await aclient.post("/items", json={"name": "async tea"})
    r = await aclient.get("/items/1")
    assert r.json()["owner_id"] == 3
    app.dependency_overrides.clear()
''', encoding="utf-8")

result = subprocess.run([sys.executable, "-m", "pytest", "-v", "-p", "no:cacheprovider", "-p", "anyio",
                         "-W", "ignore:Using `httpx`"], capture_output=True, text=True)
for line in result.stdout.splitlines():
    if "::" in line or " passed" in line or " failed" in line:
        print(line.split(" [")[0].replace("test_items.py::", "  ").rstrip() if "::" in line
              else line.split(" in ")[0].strip("= "))

Output

  test_requires_authentication PASSED
  test_create_then_read PASSED
  test_other_users_items_are_invisible PASSED
  test_validation[payload0-name] PASSED
  test_validation[payload1-name] PASSED
  test_validation[payload2-owner_id] PASSED
  test_each_test_gets_a_fresh_database PASSED
  test_websocket PASSED
  test_openapi_contract PASSED
  test_async_client PASSED
10 passed

3. Test databases: strategies

transaction-per-test against a real database (pattern; needs PostgreSQL)

@pytest.fixture
async def session(engine):
    async with engine.connect() as conn:
        outer = await conn.begin()                                   # everything below is rolled back
        async with AsyncSession(bind=conn, join_transaction_mode="create_savepoint") as s:
            yield s                                                  # app commits become SAVEPOINT releases
        await outer.rollback()

@pytest.fixture
def app(session):
    app = create_app()
    app.dependency_overrides[get_session] = lambda: session          # every request uses the test's session
    return app

4. What else senior teams test

Recap