← FastAPI

Testing FastAPI Applications

Deployment & Production

← Testing FastAPI Applications

Select a question from the sidebar

15 questions in this topic

Question 1 / 15

How does TestClient work for testing FastAPI applications?

TestClient (from Starlette, re-exported by FastAPI) wraps httpx to make synchronous HTTP requests to your FastAPI app without starting a real server. It runs the ASGI app in-process.

How it works:

Instantiate TestClient(app) with your FastAPI application
Call client.get(), client.post(), etc. like a real HTTP client
Responses have .status_code, .json(), .text, .headers
No network I/O -- everything runs in-process

Key features:

Automatic JSON serialization with json= parameter
Cookie persistence across requests within the same client
WebSocket testing via client.websocket_connect()
Follows redirects by default (disable with follow_redirects=False)

Requirement:

pip install httpx -- TestClient requires httpx as a dependency.

Pseudo Code

from fastapi.testclient import TestClient

client = TestClient(app)
response = client.get("/items")
assert response.status_code == 200
assert response.json() == [{"id": 1, "name": "item"}]

Code

from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

items_db: list[dict] = []

@app.get("/items")
def list_items():
    return items_db

@app.post("/items", status_code=201)
def create_item(item: Item):
    item_dict = item.model_dump()
    item_dict["id"] = len(items_db) + 1
    items_db.append(item_dict)
    return item_dict

@app.get("/items/{item_id}")
def get_item(item_id: int):
    for item in items_db:
        if item["id"] == item_id:
            return item
    return {"error": "not found"}

# --- Tests ---
client = TestClient(app)

# Test GET empty list
response = client.get("/items")
print(f"GET /items: {response.status_code} {response.json()}")

# Test POST
response = client.post("/items", json={"name": "Widget", "price": 9.99})
print(f"POST /items: {response.status_code} {response.json()}")

# Test GET after creation
response = client.get("/items/1")
print(f"GET /items/1: {response.status_code} {response.json()}")

# Test validation error (missing required field)
response = client.post("/items", json={"name": "Widget"})
print(f"Missing price: {response.status_code}")

GET /items: 200 []
POST /items: 201 {'name': 'Widget', 'price': 9.99, 'id': 1}
GET /items/1: 200 {'name': 'Widget', 'price': 9.99, 'id': 1}
Missing price: 422

Key Points

•TestClient runs the ASGI app in-process with no real network I/O
•Use json= for automatic serialization of request bodies
•response.json() parses the response body; response.status_code for HTTP status
•Requires httpx: pip install httpx

Gotchas

•TestClient is synchronous in its interface but fully supports async def endpoints -- it runs the ASGI app internally via anyio, so all async code works correctly
•items_db state persists between tests in the same module -- clear it in fixtures
•TestClient follows redirects by default, which can mask redirect bugs
•422 status means Pydantic validation failed -- check response.json() for details

TestClient(app) wraps httpx to make in-process HTTP requests to your FastAPI app. Use client.get(), client.post(json=...), then assert on .status_code and .json().

•In-process testing, no server needed
•json= auto-serializes request bodies

from fastapi.testclient import TestClient

client = TestClient(app)
response = client.post("/items", json={"name": "Widget", "price": 9.99})
assert response.status_code == 201
assert response.json()["name"] == "Widget"

TestClient is synchronous -- use httpx.AsyncClient for async tests

Question 2 / 15

How do you set up pytest fixtures for FastAPI tests?

pytest fixtures provide reusable test setup and teardown. For FastAPI, the most common fixtures are:

Essential fixtures:

app: The FastAPI application (possibly with test config)
client: A TestClient bound to the app
db: A database session with rollback after each test
test_user: Pre-created test data

Fixture scopes:

function (default): Fresh fixture per test. Safest for isolation
module: Shared across all tests in a file. Faster but shared state
session: Shared across all tests. Use for expensive setup like DB creation

conftest.py:

Fixtures in conftest.py are automatically discovered by pytest and available to all tests in the same directory and subdirectories.

Pseudo Code

# conftest.py
@pytest.fixture(scope="session")
def engine():
    return create_engine(TEST_DB_URL)

@pytest.fixture
def db(engine):
    connection = engine.connect()
    transaction = connection.begin()
    session = Session(bind=connection)
    yield session
    session.close()
    transaction.rollback()
    connection.close()

@pytest.fixture
def client(app):
    return TestClient(app)

Code

# conftest.py
import pytest
from fastapi import FastAPI, Depends
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, String
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column, Session, sessionmaker
)
from sqlalchemy.pool import StaticPool
from typing import Annotated

# --- App and models ---
class Base(DeclarativeBase): pass

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

def get_db():
    raise NotImplementedError("Override in tests")  # never called in tests

DbDep = Annotated[Session, Depends(get_db)]

def create_app() -> FastAPI:
    app = FastAPI()

    @app.post("/items", status_code=201)
    def create_item(name: str, db: DbDep):
        item = Item(name=name)
        db.add(item)
        db.commit()
        db.refresh(item)
        return {"id": item.id, "name": item.name}

    @app.get("/items")
    def list_items(db: DbDep):
        return db.query(Item).all()

    return app

# --- Fixtures ---
@pytest.fixture(scope="session")
def engine():
    """Single in-memory engine for all tests."""
    return create_engine(
        "sqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )

@pytest.fixture(scope="session")
def tables(engine):
    """Create tables once for the test session."""
    Base.metadata.create_all(engine)
    yield
    Base.metadata.drop_all(engine)

@pytest.fixture
def db(engine, tables):
    """Fresh session with rollback after each test."""
    connection = engine.connect()
    transaction = connection.begin()
    session = Session(connection)
    yield session
    session.close()
    transaction.rollback()
    connection.close()

@pytest.fixture
def app():
    return create_app()

@pytest.fixture
def client(app, db):
    """TestClient with overridden DB dependency."""
    def override_get_db():
        yield db

    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()

@pytest.fixture
def sample_item(db):
    """Pre-create test data."""
    item = Item(name="Test Item")
    db.add(item)
    db.commit()
    db.refresh(item)
    return item

# --- Example tests ---
def test_create_item(client):
    response = client.post("/items?name=Widget")
    assert response.status_code == 201
    assert response.json()["name"] == "Widget"

def test_list_items(client, sample_item):
    response = client.get("/items")
    assert response.status_code == 200
    assert len(response.json()) == 1

def test_isolation(client):
    """Previous test's data is rolled back."""
    response = client.get("/items")
    assert response.status_code == 200
    assert len(response.json()) == 0  # clean slate

print("Fixtures demonstrated")

Fixtures demonstrated

Key Points

•session-scoped engine and tables fixture create the DB once per test run
•function-scoped db fixture with transaction rollback gives each test a clean slate
•dependency_overrides injects the test DB into the app
•conftest.py fixtures are auto-discovered by pytest

Gotchas

•StaticPool is required for SQLite in-memory databases to share the same connection
•Forgetting to clear dependency_overrides leaks test config into other tests
•Session-scoped fixtures cannot depend on function-scoped fixtures
•Transaction rollback does not reset auto-increment counters -- don't assert on exact IDs

Put fixtures in conftest.py. Use session-scoped engine and tables, function-scoped db with transaction rollback, and a client fixture that overrides the DB dependency.

•Transaction rollback after each test ensures isolation
•dependency_overrides swaps the real DB for the test DB

@pytest.fixture
def db(engine, tables):
    connection = engine.connect()
    transaction = connection.begin()
    session = Session(connection)
    yield session
    session.close()
    transaction.rollback()
    connection.close()

@pytest.fixture
def client(app, db):
    app.dependency_overrides[get_db] = lambda: (yield db)
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()

Always clear dependency_overrides after tests to prevent leakage

Question 3 / 15

How do you use dependency_overrides to mock dependencies?

app.dependency_overrides is a dictionary that maps original dependency functions to replacement functions. When a test sets an override, FastAPI uses the replacement instead of the original.

How it works:

app.dependency_overrides[original_dependency] = mock_dependency

Use cases:

Replace database session with a test session
Mock authentication to skip auth in tests
Stub external services (email, payment, etc.)
Control return values for specific test scenarios

Important:

The key must be the exact function object used in Depends()
Override functions must have the same return type
Clear overrides after tests with app.dependency_overrides.clear()
Overrides apply to the entire app, including sub-dependencies

Pseudo Code

# Original
def get_current_user():
    # real auth logic
    return user

# In tests
def mock_current_user():
    return User(id=1, name="test")

app.dependency_overrides[get_current_user] = mock_current_user
client = TestClient(app)
response = client.get("/protected")
app.dependency_overrides.clear()

Code

from fastapi import FastAPI, Depends, HTTPException
from fastapi.testclient import TestClient
from pydantic import BaseModel
from typing import Annotated

app = FastAPI()

# --- Production dependencies ---
class User(BaseModel):
    id: int
    username: str
    is_admin: bool = False

def get_current_user() -> User:
    """In production, this validates JWT tokens."""
    raise HTTPException(status_code=401, detail="Not authenticated")

def get_admin_user(
    user: Annotated[User, Depends(get_current_user)],
) -> User:
    if not user.is_admin:
        raise HTTPException(status_code=403, detail="Not an admin")
    return user

UserDep = Annotated[User, Depends(get_current_user)]
AdminDep = Annotated[User, Depends(get_admin_user)]

@app.get("/profile")
def profile(user: UserDep):
    return {"username": user.username}

@app.get("/admin/dashboard")
def admin_dashboard(admin: AdminDep):
    return {"message": f"Welcome admin {admin.username}"}

@app.get("/public")
def public():
    return {"message": "no auth needed"}

# --- Tests with overrides ---

# Override: regular user
def mock_regular_user() -> User:
    return User(id=1, username="testuser", is_admin=False)

# Override: admin user
def mock_admin_user() -> User:
    return User(id=99, username="admin", is_admin=True)

# Test 1: Authenticated user
app.dependency_overrides[get_current_user] = mock_regular_user
client = TestClient(app)
response = client.get("/profile")
print(f"Profile: {response.status_code} {response.json()}")

# Test 2: Non-admin accessing admin route
response = client.get("/admin/dashboard")
print(f"Admin (non-admin user): {response.status_code}")

# Test 3: Admin user
app.dependency_overrides[get_current_user] = mock_admin_user
response = client.get("/admin/dashboard")
print(f"Admin (admin user): {response.status_code} {response.json()}")

# Test 4: Public endpoint (no override needed)
response = client.get("/public")
print(f"Public: {response.status_code}")

# Always clean up
app.dependency_overrides.clear()

Profile: 200 {'username': 'testuser'}
Admin (non-admin user): 403
Admin (admin user): 200 {'message': 'Welcome admin admin'}
Public: 200

Key Points

•dependency_overrides maps original functions to test replacements
•The key must be the exact function object, not a string or copy
•Overrides cascade: overriding get_current_user also affects get_admin_user
•Always clear overrides after tests to prevent state leakage

Gotchas

•If the dependency is defined in another module, you must import the same function object used in Depends()
•Overrides apply globally to the app instance -- concurrent tests may interfere
•Overriding a dependency function cascades to all places where it is used via Depends() -- overriding get_current_user also affects any endpoint or dependency that depends on it
•Yield dependencies need yield in the override too: def mock_db(): yield fake_session

app.dependency_overrides[original_fn] = mock_fn replaces any dependency in tests. The key must be the exact function object. Always call .clear() after tests.

•Maps original dependency to a mock function

def mock_current_user():
    return User(id=1, username="testuser")

app.dependency_overrides[get_current_user] = mock_current_user
client = TestClient(app)
response = client.get("/profile")
assert response.json()["username"] == "testuser"
app.dependency_overrides.clear()

The key must be the exact function object used in Depends(), not a different reference

Question 4 / 15

How do you test with an async client (httpx.AsyncClient)?

When your FastAPI endpoints use async def with truly async dependencies (async DB, async HTTP calls), you need httpx.AsyncClient for proper async testing. The synchronous TestClient works for many cases but cannot exercise the full async path.

Setup:

Install pytest-asyncio and httpx
Mark test functions with @pytest.mark.anyio or configure asyncio_mode = "auto"
Create an AsyncClient with ASGITransport to run the app in-process
Use async with for proper cleanup

When to use async tests:

Endpoints with async def and await calls
Async database sessions (AsyncSession)
Testing async WebSocket behavior
Testing streaming responses

Pseudo Code

import httpx
from httpx import ASGITransport

@pytest.mark.anyio
async def test_endpoint():
    transport = ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.get("/items")
        assert response.status_code == 200

Code

import pytest
import httpx
from httpx import ASGITransport
from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get("/items")
async def list_items():
    # Simulate async operation
    await asyncio.sleep(0.01)
    return [{"id": 1, "name": "Async Item"}]

@app.post("/items", status_code=201)
async def create_item(name: str):
    await asyncio.sleep(0.01)
    return {"id": 1, "name": name}

# --- conftest.py fixtures ---
@pytest.fixture
async def async_client():
    transport = ASGITransport(app=app)
    async with httpx.AsyncClient(
        transport=transport,
        base_url="http://test",
    ) as client:
        yield client

# --- Test examples ---
# Note: requires pytest-asyncio with asyncio_mode = "auto" in pyproject.toml
# [tool.pytest.ini_options]
# asyncio_mode = "auto"

async def test_list_items(async_client):
    response = await async_client.get("/items")
    assert response.status_code == 200
    assert len(response.json()) == 1

async def test_create_item(async_client):
    response = await async_client.post("/items?name=Widget")
    assert response.status_code == 201
    assert response.json()["name"] == "Widget"

# Running inline for demonstration
async def main():
    transport = ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.get("/items")
        print(f"GET /items: {response.status_code} {response.json()}")

        response = await client.post("/items?name=AsyncWidget")
        print(f"POST /items: {response.status_code} {response.json()}")

asyncio.run(main())

GET /items: 200 [{'id': 1, 'name': 'Async Item'}]
POST /items: 201 {'id': 1, 'name': 'AsyncWidget'}

Key Points

•ASGITransport wraps your app for in-process async testing (no real server)
•Use async with httpx.AsyncClient for proper cleanup of connections
•pytest-asyncio with asyncio_mode='auto' avoids marking every test manually
•base_url is required -- it can be any valid URL since requests stay in-process

Gotchas

•httpx.AsyncClient does not follow redirects by default (unlike TestClient which does)
•Forgetting ASGITransport and using base_url directly makes real HTTP requests to that URL
•pytest-asyncio requires explicit configuration -- without it, async tests silently do not run
•Mixing TestClient and AsyncClient in the same test suite is fine but can be confusing

Use httpx.AsyncClient with ASGITransport(app=app) for async FastAPI tests. Install pytest-asyncio and set asyncio_mode = "auto" in config. Use await client.get() in async def test functions.

•ASGITransport runs the ASGI app in-process

import httpx
from httpx import ASGITransport

async def test_list_items():
    transport = ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.get("/items")
        assert response.status_code == 200

Without ASGITransport, httpx makes real HTTP requests to the base_url

Question 5 / 15

How do you isolate test databases?

Test database isolation ensures that each test starts with a clean state. Without it, tests pass or fail depending on execution order.

Strategies (from fastest to most isolated):

Transaction rollback (recommended): Each test runs in a transaction that gets rolled back
- Fastest: no actual data written
- Limitation: Cannot test code that commits
Truncate tables: Delete all rows between tests
- Medium speed: actual commits happen
- Good for: Testing commit behavior
Recreate database: Drop and recreate all tables per test
- Slowest but most isolated
- Good for: Migration testing
In-memory SQLite: Use sqlite:///:memory: with StaticPool
- Very fast, completely isolated per test session
- Limitation: SQLite syntax differs from PostgreSQL

Pseudo Code

# Transaction rollback approach
@pytest.fixture
def db(engine):
    conn = engine.connect()
    txn = conn.begin()
    session = Session(bind=conn)
    yield session
    session.close()
    txn.rollback()  # undo everything
    conn.close()

Code

import pytest
from sqlalchemy import create_engine, String, text, select
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column, Session, sessionmaker
)
from sqlalchemy.pool import StaticPool

class Base(DeclarativeBase): pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))

# --- Strategy 1: Transaction rollback (recommended) ---
@pytest.fixture(scope="session")
def engine():
    engine = create_engine(
        "sqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(engine)
    return engine

@pytest.fixture
def db(engine):
    """Each test gets a session that rolls back."""
    connection = engine.connect()
    transaction = connection.begin()
    session = Session(connection)

    yield session

    session.close()
    transaction.rollback()
    connection.close()

def test_create_user(db):
    db.add(User(name="Alice"))
    db.commit()  # commits to the nested transaction, not the DB
    assert db.execute(select(User)).scalars().all() != []

def test_isolation(db):
    """Alice does not exist here -- previous test was rolled back."""
    users = db.execute(select(User)).scalars().all()
    assert users == []

# --- Strategy 2: Truncate tables ---
@pytest.fixture
def db_truncate(engine):
    session = Session(engine)
    yield session
    session.close()
    # Truncate all tables
    with engine.connect() as conn:
        for table in reversed(Base.metadata.sorted_tables):
            conn.execute(table.delete())
        conn.commit()

# --- Strategy 3: Separate test database URL ---
# In conftest.py or via environment variable
# TEST_DATABASE_URL = "postgresql://localhost/myapp_test"
# Production code reads from DATABASE_URL
# Tests override with TEST_DATABASE_URL

print("Test isolation strategies demonstrated")

Test isolation strategies demonstrated

Key Points

•Transaction rollback is the fastest strategy -- no data hits the disk
•StaticPool + sqlite:///:memory: gives a fast isolated test database
•session-scoped engine avoids recreating the DB for every test
•Function-scoped db fixture ensures each test gets a clean session

Gotchas

•Transaction rollback does not work if your code uses session.begin_nested() (savepoints)
•In-memory SQLite is not identical to PostgreSQL -- some SQL syntax differs
•StaticPool is required for SQLite in-memory to share the connection across threads
•commit() inside a rolled-back transaction commits to the savepoint, not the actual DB

Use transaction rollback per test: open a connection, begin a transaction, create a session, yield it, then rollback. This gives each test a clean slate without the overhead of recreating tables.

•Rollback after each test gives perfect isolation with zero overhead

@pytest.fixture
def db(engine):
    connection = engine.connect()
    transaction = connection.begin()
    session = Session(connection)
    yield session
    session.close()
    transaction.rollback()
    connection.close()

SQLite in-memory requires StaticPool to share the single connection

Question 6 / 15

How do you mock external API calls in tests?

External API calls in tests are problematic: they are slow, flaky, rate-limited, and make tests non-deterministic. There are several mocking strategies:

Strategy 1: Dependency override

Wrap the external call in a dependency and override it in tests. This is the most FastAPI-idiomatic approach.

Strategy 2: respx (mock httpx)

respx intercepts httpx requests and returns pre-configured responses. Works for both sync and async httpx clients.

Strategy 3: unittest.mock.patch

Patch the function that makes the external call. Works but is more fragile (tied to import paths).

Best practice:

Design your code so external services are behind a dependency or service class. This makes testing trivial with dependency_overrides.

Pseudo Code

# Dependency approach
def get_weather_service():
    return RealWeatherService()

# In tests
def mock_weather_service():
    return FakeWeatherService(temp=72)

app.dependency_overrides[get_weather_service] = mock_weather_service

Code

import pytest
from fastapi import FastAPI, Depends
from fastapi.testclient import TestClient
from typing import Annotated
from unittest.mock import AsyncMock, patch

app = FastAPI()

# --- Service layer (wraps external API) ---
class WeatherService:
    async def get_temperature(self, city: str) -> float:
        """In production, calls an external weather API."""
        import httpx
        async with httpx.AsyncClient() as client:
            resp = await client.get(f"https://api.weather.com/{city}")
            return resp.json()["temp"]

def get_weather_service() -> WeatherService:
    return WeatherService()

WeatherDep = Annotated[WeatherService, Depends(get_weather_service)]

@app.get("/weather/{city}")
async def get_weather(city: str, weather: WeatherDep):
    temp = await weather.get_temperature(city)
    return {"city": city, "temperature": temp}

# --- Strategy 1: Dependency override with fake ---
class FakeWeatherService:
    async def get_temperature(self, city: str) -> float:
        return 72.0  # always return 72

def test_weather_with_override():
    app.dependency_overrides[get_weather_service] = lambda: FakeWeatherService()
    client = TestClient(app)
    response = client.get("/weather/London")
    assert response.status_code == 200
    assert response.json() == {"city": "London", "temperature": 72.0}
    app.dependency_overrides.clear()

# --- Strategy 2: respx (mock httpx) ---
# pip install respx
# import respx
#
# @respx.mock
# async def test_weather_with_respx(async_client):
#     respx.get("https://api.weather.com/London").mock(
#         return_value=httpx.Response(200, json={"temp": 65.0})
#     )
#     response = await async_client.get("/weather/London")
#     assert response.json()["temperature"] == 65.0

# --- Strategy 3: unittest.mock.patch ---
def test_weather_with_patch():
    with patch.object(
        WeatherService, "get_temperature",
        new_callable=AsyncMock, return_value=55.0
    ):
        client = TestClient(app)
        response = client.get("/weather/Paris")
        assert response.json()["temperature"] == 55.0

test_weather_with_override()
test_weather_with_patch()
print("All mock strategies work")

All mock strategies work

Key Points

•Wrap external APIs in a service class or dependency for easy mocking
•dependency_overrides is the cleanest approach -- no import path fragility
•respx intercepts httpx at the transport level -- useful when you cannot refactor the dependency
•AsyncMock is needed for mocking async methods with unittest.mock

Gotchas

•unittest.mock.patch is fragile -- it depends on the exact import path of the function
•respx only works with httpx, not requests -- use responses library for the requests package
•Forgetting to mock external calls makes tests slow and flaky in CI
•The fake service must match the interface exactly or you get runtime errors in tests only

Wrap external APIs in a dependency or service class. In tests, use dependency_overrides to swap in a fake, or use respx to mock httpx calls, or unittest.mock.patch with AsyncMock.

•dependency_overrides is cleanest -- no import path fragility

class FakeWeatherService:
    async def get_temperature(self, city: str) -> float:
        return 72.0

def test_weather():
    app.dependency_overrides[get_weather_service] = lambda: FakeWeatherService()
    client = TestClient(app)
    response = client.get("/weather/London")
    assert response.json()["temperature"] == 72.0
    app.dependency_overrides.clear()

The fake must match the real service's interface exactly

Question 7 / 15

How do you test file upload endpoints?

Testing file uploads in FastAPI uses the files parameter of TestClient (or httpx.AsyncClient). Files are sent as multipart form data.

TestClient file format:

client.post("/upload", files={"file": ("name.txt", content, "mime/type")})

Tuple format:

(filename, file_content) -- minimal
(filename, file_content, content_type) -- with MIME type
(filename, file_object, content_type) -- with a file-like object

Testing multiple files:

Pass a list of tuples: files=[("files", (name, content)), ...]

Testing file + form data:

Use both files= and data= parameters.

Pseudo Code

# Single file
client.post("/upload", files={"file": ("test.txt", b"content", "text/plain")})

# Multiple files
client.post("/upload", files=[
    ("files", ("a.txt", b"aaa")),
    ("files", ("b.txt", b"bbb")),
])

# File + form data
client.post("/upload", files={"file": ...}, data={"description": "test"})

Code

from fastapi import FastAPI, UploadFile, File, Form
from fastapi.testclient import TestClient
import io

app = FastAPI()

@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    content = await file.read()
    return {
        "filename": file.filename,
        "size": len(content),
        "content_type": file.content_type,
    }

@app.post("/upload-multiple")
async def upload_multiple(files: list[UploadFile] = File(...)):
    results = []
    for f in files:
        content = await f.read()
        results.append({"filename": f.filename, "size": len(content)})
    return results

@app.post("/upload-with-metadata")
async def upload_with_metadata(
    file: UploadFile = File(...),
    description: str = Form(...),
):
    content = await file.read()
    return {
        "filename": file.filename,
        "size": len(content),
        "description": description,
    }

client = TestClient(app)

# Test 1: Single file upload
response = client.post(
    "/upload",
    files={"file": ("test.txt", b"Hello, World!", "text/plain")},
)
print(f"Single: {response.json()}")

# Test 2: Multiple files
response = client.post(
    "/upload-multiple",
    files=[
        ("files", ("a.txt", b"AAA", "text/plain")),
        ("files", ("b.csv", b"1,2,3\n4,5,6", "text/csv")),
    ],
)
print(f"Multiple: {response.json()}")

# Test 3: File with form data
response = client.post(
    "/upload-with-metadata",
    files={"file": ("report.pdf", b"%PDF-fake-content", "application/pdf")},
    data={"description": "Monthly report"},
)
print(f"With metadata: {response.json()}")

# Test 4: Using io.BytesIO (simulating a real file)
buffer = io.BytesIO(b"Binary content here")
response = client.post(
    "/upload",
    files={"file": ("data.bin", buffer, "application/octet-stream")},
)
print(f"BytesIO: {response.json()}")

Single: {'filename': 'test.txt', 'size': 13, 'content_type': 'text/plain'}
Multiple: [{'filename': 'a.txt', 'size': 3}, {'filename': 'b.csv', 'size': 11}]
With metadata: {'filename': 'report.pdf', 'size': 17, 'description': 'Monthly report'}
BytesIO: {'filename': 'data.bin', 'size': 19, 'content_type': 'application/octet-stream'}

Key Points

•files= parameter sends multipart form data
•Tuple format: (filename, content, content_type) or (filename, content)
•Multiple files use a list of tuples with the same field name
•Mix files= and data= for file uploads with form fields

Gotchas

•Cannot use json= and files= together -- use data= for additional form fields
•UploadFile.read() is async in the endpoint but TestClient handles this transparently
•Large file tests should use io.BytesIO, not real files on disk
•The content_type in the tuple overrides automatic detection

Use files={"file": ("name.txt", b"content", "text/plain")} in TestClient. For multiple files, use a list of tuples. For file + form data, combine files= and data=.

•files= sends multipart; tuple is (filename, content, mime_type)

client = TestClient(app)

# Single file
response = client.post("/upload",
    files={"file": ("test.txt", b"Hello", "text/plain")})
assert response.json()["filename"] == "test.txt"

# File + form data
response = client.post("/upload-with-metadata",
    files={"file": ("doc.pdf", b"content")},
    data={"description": "My doc"})

Cannot combine json= and files= -- use data= for form fields alongside files

Question 8 / 15

How do you test WebSocket endpoints?

TestClient provides websocket_connect() for testing WebSocket endpoints. It returns a context manager that simulates a WebSocket connection.

API:

client.websocket_connect(url): Opens a WebSocket connection
ws.send_text(data): Send a text message
ws.send_json(data): Send JSON (auto-serialized)
ws.receive_text(): Receive a text message
ws.receive_json(): Receive JSON (auto-parsed)
ws.close(): Close the connection

Testing patterns:

Test the connection lifecycle (connect, message, disconnect)
Test JSON message exchange
Test error handling (invalid data, disconnection)
Test authentication rejection

Pseudo Code

with client.websocket_connect("/ws") as ws:
    ws.send_json({"message": "hello"})
    data = ws.receive_json()
    assert data["response"] == "hello back"

Code

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.testclient import TestClient

app = FastAPI()

@app.websocket("/ws")
async def websocket_echo(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        pass

@app.websocket("/ws/json")
async def websocket_json(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_json()
            response = {
                "received": data,
                "type": type(data).__name__,
            }
            await websocket.send_json(response)
    except WebSocketDisconnect:
        pass

@app.websocket("/ws/auth")
async def websocket_auth(websocket: WebSocket):
    token = websocket.query_params.get("token")
    if token != "valid-token":
        await websocket.close(code=4001, reason="Invalid token")
        return
    await websocket.accept()
    await websocket.send_text("Authenticated!")

client = TestClient(app)

# Test 1: Echo text messages
with client.websocket_connect("/ws") as ws:
    ws.send_text("Hello")
    data = ws.receive_text()
    print(f"Echo: {data}")

    ws.send_text("World")
    data = ws.receive_text()
    print(f"Echo: {data}")

# Test 2: JSON messages
with client.websocket_connect("/ws/json") as ws:
    ws.send_json({"action": "ping", "value": 42})
    response = ws.receive_json()
    print(f"JSON: {response}")

# Test 3: Authentication rejection
try:
    with client.websocket_connect("/ws/auth?token=bad-token") as ws:
        ws.receive_text()  # should raise
except Exception as e:
    print(f"Auth rejected (connection closed)")

# Test 4: Successful authentication
with client.websocket_connect("/ws/auth?token=valid-token") as ws:
    data = ws.receive_text()
    print(f"Auth: {data}")

Echo: Echo: Hello
Echo: Echo: World
JSON: {'received': {'action': 'ping', 'value': 42}, 'type': 'dict'}
Auth rejected (connection closed)
Auth: Authenticated!

Key Points

•websocket_connect() returns a context manager simulating a WS connection
•send_json/receive_json handle serialization automatically
•Query parameters in the URL work for auth tokens: /ws?token=abc
•Connection close during receive raises an exception that you can assert on

Gotchas

•websocket_connect is synchronous even though the WebSocket endpoint is async
•Receiving from a closed connection raises -- use try/except in tests
•There is no built-in timeout on receive -- tests can hang if the server never sends
•WebSocket test connections go through the full ASGI middleware stack just like HTTP requests -- middleware that checks scope['type'] must handle both 'http' and 'websocket' types

Use client.websocket_connect("/ws") as a context manager. Call ws.send_text() / ws.send_json() to send and ws.receive_text() / ws.receive_json() to receive.

•websocket_connect provides a sync context manager for WS testing

client = TestClient(app)

with client.websocket_connect("/ws") as ws:
    ws.send_text("Hello")
    data = ws.receive_text()
    assert data == "Echo: Hello"

receive calls can hang if the server endpoint does not send a response

Question 9 / 15

How do you test authentication-protected routes?

Testing authenticated routes requires either mocking the auth dependency (faster, simpler) or generating real tokens (more realistic). The recommended approach depends on what you are testing.

Approach 1: dependency_overrides (recommended for most tests)

Override get_current_user to return a fake user. Tests skip token validation entirely.

Approach 2: Real tokens (for auth-specific tests)

Generate a valid JWT in the test fixture and send it in the Authorization header.

Approach 3: Fixture-based token helper

Create a client fixture that pre-configures the Authorization header.

What to test:

Unauthenticated access returns 401
Valid token returns the protected resource
Expired/invalid tokens return 401
Insufficient permissions return 403

Pseudo Code

# Approach 1: Override
app.dependency_overrides[get_current_user] = lambda: fake_user

# Approach 2: Real token
token = create_access_token({"sub": "testuser"})
client.get("/protected", headers={"Authorization": f"Bearer {token}"})

# Approach 3: Fixture
@pytest.fixture
def auth_client():
    token = create_access_token({"sub": "testuser"})
    client.headers = {"Authorization": f"Bearer {token}"}
    return client

Code

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from fastapi.testclient import TestClient
from pydantic import BaseModel
from typing import Annotated
from datetime import datetime, timedelta, timezone
import jwt

SECRET_KEY = "test-secret-key"
ALGORITHM = "HS256"

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")

class User(BaseModel):
    username: str
    is_admin: bool = False

def create_access_token(data: dict, expires_minutes: int = 30) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=expires_minutes)
    return jwt.encode({**data, "exp": expire}, SECRET_KEY, algorithm=ALGORITHM)

def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
) -> User:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return User(username=payload["sub"], is_admin=payload.get("admin", False))
    except jwt.PyJWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

UserDep = Annotated[User, Depends(get_current_user)]

@app.get("/profile")
def profile(user: UserDep):
    return {"username": user.username}

@app.get("/admin")
def admin(user: UserDep):
    if not user.is_admin:
        raise HTTPException(status_code=403, detail="Admins only")
    return {"admin": user.username}

# --- Approach 1: dependency_overrides (simplest) ---
def test_profile_with_override():
    app.dependency_overrides[get_current_user] = lambda: User(username="alice")
    client = TestClient(app)
    response = client.get("/profile")
    assert response.status_code == 200
    assert response.json()["username"] == "alice"
    app.dependency_overrides.clear()

# --- Approach 2: Real token ---
def test_profile_with_real_token():
    token = create_access_token({"sub": "bob"})
    client = TestClient(app)
    response = client.get("/profile", headers={"Authorization": f"Bearer {token}"})
    assert response.status_code == 200
    assert response.json()["username"] == "bob"

def test_unauthenticated():
    app.dependency_overrides.clear()  # ensure no overrides
    client = TestClient(app)
    response = client.get("/profile")
    assert response.status_code == 401

def test_forbidden():
    app.dependency_overrides[get_current_user] = lambda: User(
        username="regular", is_admin=False
    )
    client = TestClient(app)
    response = client.get("/admin")
    assert response.status_code == 403
    app.dependency_overrides.clear()

# Run tests
test_profile_with_override()
test_profile_with_real_token()
test_unauthenticated()
test_forbidden()
print("All auth tests passed")

All auth tests passed

Key Points

•dependency_overrides on get_current_user is the simplest way to test protected routes
•Real token tests validate the full auth pipeline including JWT decoding
•Test all cases: unauthenticated (401), forbidden (403), and successful access
•Create a fixture that generates tokens to avoid repeating JWT creation code

Gotchas

•dependency_overrides skips the real auth logic -- use real tokens to test the auth itself
•TestClient does not set Authorization headers automatically -- pass them explicitly
•Expired tokens in tests can cause flaky failures if test timing is unpredictable
•OAuth2PasswordBearer returns 401 with a WWW-Authenticate header, not just a plain 403

Override get_current_user for most tests. Use real JWT tokens when testing the auth pipeline itself. Always test 401 (unauthenticated) and 403 (forbidden) cases.

•dependency_overrides for most tests; real tokens for auth-specific tests

# Override for fast tests
app.dependency_overrides[get_current_user] = lambda: User(username="alice")
response = client.get("/profile")
assert response.json()["username"] == "alice"

# Real token for auth tests
token = create_access_token({"sub": "bob"})
response = client.get("/profile", headers={"Authorization": f"Bearer {token}"})
assert response.status_code == 200

Overriding auth skips the real validation -- test the auth pipeline separately

Question 10 / 15

How do you test error handling and validation errors?

FastAPI automatically returns 422 Unprocessable Entity for Pydantic validation failures and allows custom exception handlers for other error types. Tests should verify both the status code and the error response body.

What to test:

422 validation errors: Missing fields, wrong types, constraint violations
404 not found: Requesting non-existent resources
Custom exceptions: Application-specific error types
Error response format: Verify the error body structure, not just the status code

FastAPI's 422 response format:

{
  "detail": [
    {
      "type": "missing",
      "loc": ["body", "name"],
      "msg": "Field required",
      "input": {...}
    }
  ]
}

Pseudo Code

# Test validation error
response = client.post("/items", json={})
assert response.status_code == 422
assert response.json()["detail"][0]["loc"] == ["body", "name"]

# Test custom error
response = client.get("/items/999")
assert response.status_code == 404
assert "not found" in response.json()["detail"]

Code

from fastapi import FastAPI, HTTPException, Request
from fastapi.testclient import TestClient
from fastapi.responses import JSONResponse
from pydantic import BaseModel, field_validator

app = FastAPI()

# Custom exception
class ItemNotFoundError(Exception):
    def __init__(self, item_id: int):
        self.item_id = item_id

@app.exception_handler(ItemNotFoundError)
async def item_not_found_handler(request: Request, exc: ItemNotFoundError):
    return JSONResponse(
        status_code=404,
        content={"error": "item_not_found", "item_id": exc.item_id},
    )

class ItemCreate(BaseModel):
    name: str
    price: float
    quantity: int = 0

    @field_validator("price")
    @classmethod
    def price_must_be_positive(cls, v: float) -> float:
        if v <= 0:
            raise ValueError("Price must be positive")
        return v

items_db = {1: {"id": 1, "name": "Widget", "price": 9.99}}

@app.post("/items", status_code=201)
def create_item(item: ItemCreate):
    return item.model_dump()

@app.get("/items/{item_id}")
def get_item(item_id: int):
    if item_id not in items_db:
        raise ItemNotFoundError(item_id)
    return items_db[item_id]

client = TestClient(app)

# Test 1: Missing required field
response = client.post("/items", json={"price": 9.99})
print(f"Missing name: {response.status_code}")
detail = response.json()["detail"]
print(f"  Error loc: {detail[0]['loc']}")
print(f"  Error type: {detail[0]['type']}")

# Test 2: Wrong type
response = client.post("/items", json={"name": "Widget", "price": "not-a-number"})
print(f"Wrong type: {response.status_code}")

# Test 3: Validation constraint (negative price)
response = client.post("/items", json={"name": "Widget", "price": -5.0})
print(f"Negative price: {response.status_code}")
print(f"  Message: {response.json()['detail'][0]['msg']}")

# Test 4: Custom exception handler
response = client.get("/items/999")
print(f"Not found: {response.status_code} {response.json()}")

# Test 5: Path parameter validation (string instead of int)
response = client.get("/items/abc")
print(f"Invalid path param: {response.status_code}")

# Test 6: Valid request
response = client.post("/items", json={"name": "Widget", "price": 9.99})
print(f"Valid: {response.status_code}")

Missing name: 422
Error loc: ['body', 'name']
Error type: missing
Wrong type: 422
Negative price: 422
Message: Value error, Price must be positive
Not found: 404 {'error': 'item_not_found', 'item_id': 999}
Invalid path param: 422
Valid: 201

Key Points

•422 is returned for all Pydantic validation failures with a structured detail array
•Each validation error includes loc (field path), msg (description), and type (error kind)
•Custom exception handlers return consistent error formats for domain errors
•Path parameter type mismatches (e.g., string for int) also return 422

Gotchas

•FastAPI's 422 format changed between Pydantic v1 and v2 -- test the actual structure
•Custom exception handlers must be async even if the logic is synchronous
•HTTPException detail can be a string or dict -- custom handlers give more control
•Testing only status codes is insufficient -- always assert on the error body structure

Test validation errors (422) by sending invalid data and asserting on both status_code and response.json()["detail"]. Test custom exceptions (404, etc.) by triggering the error condition and verifying the response body.

•Always assert on the error body, not just status codes

# Missing required field
response = client.post("/items", json={"price": 9.99})
assert response.status_code == 422
assert response.json()["detail"][0]["loc"] == ["body", "name"]

# Custom error
response = client.get("/items/999")
assert response.status_code == 404
assert response.json()["error"] == "item_not_found"

422 error format differs between Pydantic v1 and v2

Question 11 / 15

How do you organize FastAPI test files and fixtures?

Well-organized tests are essential for maintainability. The standard pattern mirrors the application structure.

Recommended structure:

project/
  app/
    __init__.py
    main.py
    models.py
    schemas.py
    routes/
      users.py
      items.py
  tests/
    __init__.py
    conftest.py          # shared fixtures
    test_users.py        # tests for user routes
    test_items.py        # tests for item routes
    fixtures/
      __init__.py
      database.py        # DB fixtures
      auth.py            # auth fixtures
      factories.py       # test data factories

conftest.py hierarchy:

tests/conftest.py: Shared fixtures (app, client, db)
tests/fixtures/*.py: Importable fixture modules
Each test file can have local fixtures for specific needs

Naming conventions:

Test files: test_<module>.py
Test functions: test_<what_it_does>()
Fixtures: Descriptive nouns (db, client, sample_user)

Test categories:

Unit tests: Test functions in isolation
Integration tests: Test endpoints with real DB
Contract tests: Verify API response shapes

Pseudo Code

# tests/conftest.py -- shared fixtures
@pytest.fixture(scope="session")
def engine(): ...

@pytest.fixture
def db(engine): ...

@pytest.fixture
def client(app, db): ...

@pytest.fixture
def auth_headers(): ...

# tests/test_users.py
def test_create_user(client): ...
def test_get_user(client, sample_user): ...
def test_create_duplicate_email(client, sample_user): ...

Code

# tests/conftest.py
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool

# Import from your app
# from app.main import create_app
# from app.database import Base, get_db
# from app.models import User, Item

# --- Database fixtures ---
@pytest.fixture(scope="session")
def engine():
    return create_engine(
        "sqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )

@pytest.fixture(scope="session")
def tables(engine):
    # Base.metadata.create_all(engine)
    yield
    # Base.metadata.drop_all(engine)

@pytest.fixture
def db(engine, tables):
    connection = engine.connect()
    transaction = connection.begin()
    session = Session(bind=connection)
    yield session
    session.close()
    transaction.rollback()
    connection.close()

# --- App fixtures ---
@pytest.fixture
def app():
    # return create_app()
    return FastAPI()

@pytest.fixture
def client(app, db):
    # app.dependency_overrides[get_db] = lambda: (yield db)
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()

# --- Auth fixtures ---
@pytest.fixture
def auth_headers():
    """Headers for an authenticated regular user."""
    # token = create_access_token({"sub": "testuser"})
    token = "fake-token"
    return {"Authorization": f"Bearer {token}"}

@pytest.fixture
def admin_headers():
    """Headers for an admin user."""
    token = "fake-admin-token"
    return {"Authorization": f"Bearer {token}"}

# --- Factory fixtures ---
@pytest.fixture
def user_factory(db):
    """Create users on demand with customizable fields."""
    def _create(name="Test User", email=None):
        # email = email or f"{name.lower().replace(' ', '.')}@test.com"
        # user = User(name=name, email=email)
        # db.add(user)
        # db.commit()
        # return user
        return {"name": name, "email": email}
    return _create

@pytest.fixture
def sample_user(user_factory):
    return user_factory(name="Alice", email="alice@test.com")

# --- Example test file: tests/test_users.py ---
def test_create_user(client):
    response = client.post("/users", json={"name": "Bob", "email": "bob@test.com"})
    # assert response.status_code == 201

def test_get_user(client, sample_user):
    # response = client.get(f"/users/{sample_user.id}")
    # assert response.status_code == 200
    pass

def test_list_users(client, sample_user):
    # response = client.get("/users")
    # assert len(response.json()) >= 1
    pass

def test_duplicate_email(client, sample_user):
    # response = client.post("/users", json={"name": "Dup", "email": "alice@test.com"})
    # assert response.status_code == 409
    pass

print("Test organization demonstrated")

Test organization demonstrated

Key Points

•conftest.py fixtures are auto-discovered by pytest for all tests in the directory
•Use factory fixtures for creating test data with customizable fields
•Separate fixtures by concern: database, auth, data factories
•Mirror the app structure in your test directory for easy navigation

Gotchas

•conftest.py must be in the tests directory (or a parent) to be discovered
•Factory fixtures return a callable, not the object itself
•Importing from app modules in conftest.py can cause circular imports if not careful
•Session-scoped fixtures cannot use function-scoped fixtures as dependencies

Put shared fixtures (app, client, db) in tests/conftest.py. Mirror your app structure with test_<module>.py files. Use factory fixtures for reusable test data creation.

•conftest.py for shared fixtures, factory fixtures for test data

# tests/conftest.py
@pytest.fixture
def client(app, db):
    app.dependency_overrides[get_db] = lambda: (yield db)
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()

@pytest.fixture
def user_factory(db):
    def _create(name="Test User"):
        user = User(name=name)
        db.add(user)
        db.commit()
        return user
    return _create

Session-scoped fixtures cannot depend on function-scoped fixtures

Question 12 / 15

How do you test background tasks?

BackgroundTasks in FastAPI run after the response is sent. Testing them is tricky because:

They execute asynchronously after the endpoint returns
Their side effects (sending emails, writing files) are not visible in the response
With TestClient, background tasks actually run synchronously before the test continues

Testing strategies:

TestClient runs tasks synchronously: With the sync TestClient, background tasks complete before response is returned. You can directly check side effects.
Mock the task function: Verify the task was called with correct arguments.
Extract business logic: Move the logic into a testable function and test it independently.

Pseudo Code

# Strategy 1: Check side effects (TestClient runs tasks synchronously)
response = client.post("/send-email")
assert response.status_code == 200
assert email_was_sent()  # task already ran

# Strategy 2: Mock the task
with patch("app.tasks.send_email") as mock_send:
    response = client.post("/send-email")
    mock_send.assert_called_once_with("user@test.com")

# Strategy 3: Test task function directly
def test_send_email_task():
    send_email("user@test.com")  # test in isolation
    assert email_log[-1] == "user@test.com"

Code

from fastapi import FastAPI, BackgroundTasks
from fastapi.testclient import TestClient
from unittest.mock import patch, MagicMock

app = FastAPI()

# Shared log for demonstrating side effects
email_log: list[str] = []

def send_notification_email(email: str, message: str):
    """Simulates sending an email (would be async HTTP call in production)."""
    email_log.append(f"{email}: {message}")

@app.post("/register")
def register_user(
    email: str,
    background_tasks: BackgroundTasks,
):
    # Register user (synchronous)
    user = {"email": email, "id": 1}

    # Schedule email (runs after response)
    background_tasks.add_task(
        send_notification_email,
        email=email,
        message="Welcome!",
    )

    return {"user_id": user["id"], "message": "Registration started"}

client = TestClient(app)

# Strategy 1: Check side effects
# TestClient runs background tasks synchronously before returning
email_log.clear()
response = client.post("/register?email=alice@test.com")
print(f"Status: {response.status_code}")
print(f"Email log: {email_log}")
assert len(email_log) == 1
assert "alice@test.com" in email_log[0]

# Strategy 2: Mock the task function
with patch("__main__.send_notification_email") as mock_send:
    response = client.post("/register?email=bob@test.com")
    print(f"Mock called: {mock_send.called}")
    mock_send.assert_called_once_with(
        email="bob@test.com",
        message="Welcome!",
    )

# Strategy 3: Test the task function directly
email_log.clear()
send_notification_email("charlie@test.com", "Direct test")
print(f"Direct call log: {email_log}")
assert email_log == ["charlie@test.com: Direct test"]

print("All background task tests passed")

Status: 200
Email log: ['alice@test.com: Welcome!']
Mock called: True
Direct call log: ['charlie@test.com: Direct test']
All background task tests passed

Key Points

•TestClient runs background tasks synchronously -- side effects are visible immediately
•Mock the task function to verify it was called with correct arguments
•Extract task logic into standalone functions for independent unit testing
•Background tasks in production run after the response -- test timing differs

Gotchas

•TestClient behavior differs from production: tasks run synchronously in tests, asynchronously in production
•If the background task has a bug, TestClient raises the exception -- in production it would be silently lost
•Background tasks do not have access to yield-dependency cleanup -- the session is already closed
•asyncio.create_task() in background tasks requires careful handling in tests

TestClient runs background tasks synchronously, so side effects are visible right after the response. Use unittest.mock.patch to verify the task was called, or test the task function directly.

•TestClient runs tasks synchronously -- check side effects directly

# Side effects visible immediately with TestClient
response = client.post("/register?email=alice@test.com")
assert len(email_log) == 1  # task already ran

# Or mock it
with patch("app.tasks.send_email") as mock:
    client.post("/register?email=bob@test.com")
    mock.assert_called_once()

Production runs tasks asynchronously -- timing behavior differs from tests

Question 13 / 15

How do you test middleware behavior?

Middleware in FastAPI processes requests before they reach endpoints and responses before they are sent. Testing middleware requires verifying both the request transformation and the response transformation.

What to test:

Response headers added by middleware (CORS, timing, request ID)
Request blocking (rate limiting, IP filtering)
Request modification (adding headers, logging)
Error handling in middleware

Testing approach:

Use TestClient normally -- middleware runs automatically for every request. Check response headers, status codes, and body to verify middleware behavior.

Pseudo Code

# Middleware adds a header
@app.middleware("http")
async def add_timing(request, call_next):
    start = time.time()
    response = await call_next(request)
    response.headers["X-Process-Time"] = str(time.time() - start)
    return response

# Test
response = client.get("/anything")
assert "X-Process-Time" in response.headers

Code

import time
import uuid
from fastapi import FastAPI, Request, Response
from fastapi.testclient import TestClient
from starlette.middleware.base import BaseHTTPMiddleware

app = FastAPI()

# Middleware 1: Add request timing header
@app.middleware("http")
async def timing_middleware(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    process_time = time.time() - start
    response.headers["X-Process-Time"] = f"{process_time:.4f}"
    return response

# Middleware 2: Add request ID
@app.middleware("http")
async def request_id_middleware(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response

# Middleware 3: Block certain paths
@app.middleware("http")
async def block_middleware(request: Request, call_next):
    if request.url.path.startswith("/internal"):
        return Response(content="Forbidden", status_code=403)
    return await call_next(request)

@app.get("/health")
def health():
    return {"status": "ok"}

@app.get("/internal/debug")
def debug():
    return {"debug": True}

client = TestClient(app)

# Test 1: Timing header is present
response = client.get("/health")
print(f"Timing header: {response.headers.get('X-Process-Time')}")
assert "X-Process-Time" in response.headers
assert float(response.headers["X-Process-Time"]) >= 0

# Test 2: Request ID propagation
response = client.get("/health", headers={"X-Request-ID": "test-123"})
print(f"Request ID: {response.headers['X-Request-ID']}")
assert response.headers["X-Request-ID"] == "test-123"

# Test 3: Auto-generated request ID
response = client.get("/health")
request_id = response.headers["X-Request-ID"]
print(f"Auto request ID: {request_id}")
assert len(request_id) > 0  # UUID was generated

# Test 4: Path blocking
response = client.get("/internal/debug")
print(f"Blocked: {response.status_code} {response.text}")
assert response.status_code == 403

# Test 5: Non-blocked path works normally
response = client.get("/health")
print(f"Health: {response.status_code}")
assert response.status_code == 200

print("All middleware tests passed")

Timing header: 0.0001
Request ID: test-123
Auto request ID: <uuid-string>
Blocked: 403 Forbidden
Health: 200
All middleware tests passed

Key Points

•Middleware runs automatically on every TestClient request
•Check response.headers for middleware-added headers
•Test both normal flow and blocked/rejected requests
•Send custom headers in TestClient to test header propagation

Gotchas

•Middleware added with @app.middleware('http') runs in registration order on the request path (first registered = outermost = runs first) and reverse order on the response path
•Timing values in tests are very small -- assert >= 0 rather than exact values
•If middleware raises an exception, the response may not have expected headers
•StreamingResponse in middleware may cause body reading issues

Middleware runs automatically with TestClient. Check response.headers for added headers, verify blocked paths return 403, and send custom request headers to test propagation.

•Middleware runs automatically; check response headers and status codes

client = TestClient(app)

# Test timing header exists
response = client.get("/health")
assert "X-Process-Time" in response.headers

# Test path blocking
response = client.get("/internal/debug")
assert response.status_code == 403

# Test header propagation
response = client.get("/health", headers={"X-Request-ID": "test-123"})
assert response.headers["X-Request-ID"] == "test-123"

Middleware execution order is the reverse of registration order

Question 14 / 15

How do you achieve test coverage for FastAPI applications?

Test coverage measures what percentage of your code is executed during tests. For FastAPI applications, you use pytest-cov to track coverage and identify untested code paths.

Setup:

Install: pip install pytest-cov
Run: pytest --cov=app --cov-report=term-missing
The term-missing option shows which lines are not covered

Coverage targets:

90%+ overall: Good target for production applications
100% on critical paths: Auth, payment, data mutations
Skip low-value coverage: Config files, repr methods

What to cover:

All endpoint paths (happy path + error paths)
Validation edge cases
Authentication/authorization branches
Error handlers and middleware
Background tasks

Coverage configuration in pyproject.toml:

Configure source, exclude patterns, and failure thresholds.

Pseudo Code

# Run coverage
pytest --cov=app --cov-report=term-missing -v

# pyproject.toml
[tool.coverage.run]
source = ["app"]

[tool.coverage.report]
fail_under = 90
exclude_lines = ["pragma: no cover", "if TYPE_CHECKING:"]

Code

# pyproject.toml configuration
"""
[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"

[tool.coverage.run]
source = ["app"]
omit = [
    "*/tests/*",
    "*/migrations/*",
    "app/config.py",
]

[tool.coverage.report]
fail_under = 90
show_missing = true
exclude_lines = [
    "pragma: no cover",
    "if TYPE_CHECKING:",
    "if __name__ == .__main__.:",
    "raise NotImplementedError",
    "\\.\\.\\.",
]

[tool.coverage.html]
directory = "htmlcov"
"""

# Example: testing all branches of an endpoint
from fastapi import FastAPI, HTTPException
from fastapi.testclient import TestClient

app = FastAPI()

items_db = {}

@app.post("/items", status_code=201)
def create_item(name: str, price: float):
    if price <= 0:               # branch 1: validation
        raise HTTPException(400, "Price must be positive")
    if name in items_db:         # branch 2: duplicate
        raise HTTPException(409, "Item already exists")
    items_db[name] = price       # branch 3: success
    return {"name": name, "price": price}

client = TestClient(app)

# Test all branches for 100% coverage of this endpoint
def test_create_success():
    items_db.clear()
    r = client.post("/items?name=Widget&price=9.99")
    assert r.status_code == 201

def test_create_negative_price():
    r = client.post("/items?name=Bad&price=-1")
    assert r.status_code == 400

def test_create_duplicate():
    items_db.clear()
    items_db["Widget"] = 9.99
    r = client.post("/items?name=Widget&price=5.00")
    assert r.status_code == 409

# Run all
test_create_success()
test_create_negative_price()
test_create_duplicate()
print("All coverage branches tested")
print("Run: pytest --cov=app --cov-report=term-missing -v")

All coverage branches tested
Run: pytest --cov=app --cov-report=term-missing -v

Key Points

•pytest --cov=app --cov-report=term-missing shows uncovered lines
•fail_under in pyproject.toml fails CI if coverage drops below threshold
•Test every branch: happy path, validation errors, auth failures, edge cases
•Use exclude_lines to skip untestable code like TYPE_CHECKING blocks

Gotchas

•High coverage does not mean good tests -- you can cover code without meaningful assertions
•Branch coverage (--cov-branch) is stricter than line coverage -- use it for critical code
•Background tasks and middleware are often missed in coverage -- test them explicitly
•Coverage omit patterns need careful tuning to avoid excluding important code

Use pytest --cov=app --cov-report=term-missing to measure coverage. Set fail_under = 90 in pyproject.toml to enforce a minimum. Test every branch: success, validation error, auth failure, not found.

•term-missing shows exact uncovered lines

# Run coverage
# pytest --cov=app --cov-report=term-missing -v

# pyproject.toml
# [tool.coverage.run]
# source = ["app"]
# [tool.coverage.report]
# fail_under = 90
# show_missing = true

High coverage does not mean quality tests -- assertions matter more than lines covered

Question 15 / 15

How do you test with different configurations (testing vs production)?

FastAPI applications should use different configurations for testing, development, and production. Pydantic Settings (pydantic-settings package) is the standard way to manage configuration with environment variable support.

Pattern:

Define settings as a Pydantic BaseSettings class
Create a get_settings dependency
Override the dependency in tests with test-specific values

Key differences between environments:

Setting	Production	Testing
Database	PostgreSQL	SQLite in-memory
Debug	False	True
Email	Real SMTP	Mock/log
External APIs	Real endpoints	Mock/fake
Secret key	From env var	Hardcoded test value

Pseudo Code

class Settings(BaseSettings):
    database_url: str = "sqlite:///./app.db"
    debug: bool = False
    secret_key: str

def get_settings():
    return Settings()

# In tests:
def test_settings():
    return Settings(database_url="sqlite:///:memory:", debug=True)

app.dependency_overrides[get_settings] = test_settings

Code

from fastapi import FastAPI, Depends
from fastapi.testclient import TestClient
from pydantic_settings import BaseSettings
from pydantic import ConfigDict
from typing import Annotated
from functools import lru_cache

# --- Settings class ---
class Settings(BaseSettings):
    app_name: str = "MyApp"
    database_url: str = "postgresql://localhost/myapp"
    secret_key: str = "change-me-in-production"
    debug: bool = False
    testing: bool = False
    email_backend: str = "smtp"

    model_config = ConfigDict(
        env_file=".env",        # load from .env file
        env_prefix="APP_",      # APP_DATABASE_URL, APP_SECRET_KEY, etc.
        extra="ignore",         # ignore unknown env vars
    )

# Cache settings to avoid re-reading env vars on every request
@lru_cache
def get_settings() -> Settings:
    return Settings()

SettingsDep = Annotated[Settings, Depends(get_settings)]

# --- App ---
app = FastAPI()

@app.get("/info")
def info(settings: SettingsDep):
    return {
        "app_name": settings.app_name,
        "debug": settings.debug,
        "testing": settings.testing,
    }

@app.get("/db-url")
def db_url(settings: SettingsDep):
    if settings.testing:
        return {"db": "test database"}
    return {"db": "production database"}

# --- Test configuration ---
def get_test_settings() -> Settings:
    return Settings(
        app_name="MyApp-Test",
        database_url="sqlite:///:memory:",
        secret_key="test-secret",
        debug=True,
        testing=True,
        email_backend="console",
    )

# Override in tests
app.dependency_overrides[get_settings] = get_test_settings
client = TestClient(app)

response = client.get("/info")
print(f"Info: {response.json()}")

response = client.get("/db-url")
print(f"DB: {response.json()}")

# Verify test config is active
assert response.json()["db"] == "test database"

app.dependency_overrides.clear()

# --- Alternative: conftest.py pattern ---
# @pytest.fixture(autouse=True)
# def override_settings():
#     app.dependency_overrides[get_settings] = get_test_settings
#     yield
#     app.dependency_overrides.clear()

print("Configuration override demonstrated")

Info: {'app_name': 'MyApp-Test', 'debug': True, 'testing': True}
DB: {'db': 'test database'}
Configuration override demonstrated

Key Points

•Pydantic BaseSettings reads from environment variables and .env files
•env_prefix avoids naming collisions (APP_DATABASE_URL instead of DATABASE_URL)
•Override get_settings in tests to inject test-specific configuration
•lru_cache on get_settings avoids re-parsing env vars on every request

Gotchas

•lru_cache means settings are frozen after first call -- clear the cache if you change env vars at runtime
•The .env file is not loaded in tests unless you explicitly point to it
•pydantic-settings is a separate package from pydantic: pip install pydantic-settings
•autouse=True on the settings fixture ensures all tests get the test configuration

Use Pydantic BaseSettings for configuration. Create get_settings() as a dependency, then override it in tests with app.dependency_overrides[get_settings] = get_test_settings.

•Override get_settings dependency in tests for different configuration

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str = "postgresql://localhost/myapp"
    testing: bool = False

def get_test_settings():
    return Settings(database_url="sqlite:///:memory:", testing=True)

app.dependency_overrides[get_settings] = get_test_settings

pydantic-settings is a separate package: pip install pydantic-settings

Built by onllm.dev

Home GitHub Support awesomePrep