KAHIBARO
Discord Login Register ☰
☰

19.10. Testing FastAPI

Table of Contents

Overview

When you build FastAPI applications, you must verify that your endpoints behave correctly, handle errors, respect authentication, and integrate properly with the database and other services. This chapter focuses on how to test FastAPI apps specifically, assuming you already understand basic testing concepts and pytest from earlier chapters.

You will learn how to:

Basic FastAPI Test Setup

FastAPI ships with an excellent testing helper built on requests: TestClient. It lets you call your app as if it were running, but entirely in memory and very fast.

A minimal FastAPI app:

python 
# app/main.py
from fastapi import FastAPI
app = FastAPI()
@app.get("/hello")
def read_hello():
    return {"message": "Hello, world!"}

A minimal test using pytest and TestClient:

python 
# tests/test_hello.py
from fastapi.testclient import TestClient
from app.main import app
client = TestClient(app)
def test_read_hello():
    response = client.get("/hello")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello, world!"}

Run:

bash 
pytest

You did not start a real HTTP server. TestClient runs the ASGI app directly in the same process.

Using a `client` Fixture

Instead of creating TestClient in every test file, you can use a pytest fixture:

python 
# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
@pytest.fixture
def client():
    with TestClient(app) as c:
        yield c

Then in tests:

python 
# tests/test_hello.py
def test_read_hello(client):
    resp = client.get("/hello")
    assert resp.status_code == 200

This pattern makes it easy to swap configurations in one place if needed.

Testing Path Operations

Each FastAPI endpoint is a path operation. In tests, you call them using the matching HTTP method, then make assertions on:

Query Parameters

Suppose your app has:

python 
# app/main.py
from typing import Optional
from fastapi import FastAPI
app = FastAPI()
@app.get("/items")
def list_items(limit: int = 10, q: Optional[str] = None):
    items = [{"id": i, "name": f"Item {i}"} for i in range(1, limit + 1)]
    if q:
        items = [item for item in items if q.lower() in item["name"].lower()]
    return items

Test:

python 
# tests/test_items.py
def test_list_items_default_limit(client):
    resp = client.get("/items")
    assert resp.status_code == 200
    data = resp.json()
    assert len(data) == 10
    assert data[0]["id"] == 1
def test_list_items_custom_limit(client):
    resp = client.get("/items", params={"limit": 3})
    assert resp.status_code == 200
    data = resp.json()
    assert len(data) == 3
def test_list_items_with_search_query(client):
    resp = client.get("/items", params={"limit": 5, "q": "Item 3"})
    assert resp.status_code == 200
    data = resp.json()
    assert len(data) == 1
    assert data[0]["name"] == "Item 3"

Note the params argument in client.get, which sends query parameters.

Path Parameters

App code:

python 
# app/main.py
from fastapi import HTTPException
items = {
    1: {"id": 1, "name": "Apple"},
    2: {"id": 2, "name": "Banana"},
}
@app.get("/items/{item_id}")
def get_item(item_id: int):
    item = items.get(item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

Test:

python 
# tests/test_item_detail.py
def test_get_existing_item(client):
    resp = client.get("/items/1")
    assert resp.status_code == 200
    assert resp.json() == {"id": 1, "name": "Apple"}
def test_get_missing_item_returns_404(client):
    resp = client.get("/items/999")
    assert resp.status_code == 404
    assert resp.json() == {"detail": "Item not found"}

You simply include the path parameter in the URL.

Testing JSON Request Bodies

FastAPI automatically parses JSON request bodies into Pydantic models or plain Python types. With TestClient, you pass JSON using the json argument.

App:

python 
# app/main.py
from pydantic import BaseModel, Field
from fastapi import FastAPI, HTTPException
app = FastAPI()
class ItemCreate(BaseModel):
    name: str = Field(..., min_length=3)
    price: float = Field(..., gt=0)
fake_db = []
_next_id = 1
@app.post("/items", status_code=201)
def create_item(payload: ItemCreate):
    global _next_id
    item = {"id": _next_id, "name": payload.name, "price": payload.price}
    fake_db.append(item)
    _next_id += 1
    return item

Test:

python 
# tests/test_items_create.py
def test_create_item_success(client):
    body = {"name": "Laptop", "price": 999.99}
    resp = client.post("/items", json=body)
    assert resp.status_code == 201
    data = resp.json()
    assert data["id"] == 1
    assert data["name"] == "Laptop"
    assert data["price"] == 999.99
def test_create_item_validation_error(client):
    body = {"name": "AB", "price": -10}
    resp = client.post("/items", json=body)
    assert resp.status_code == 422  # Unprocessable Entity
    data = resp.json()
    # You can assert structure or specific messages
    assert data["detail"]

FastAPI returns 422 if Pydantic validation fails. It is important to test both success and validation failures.

Testing Dependencies and Overrides

FastAPI dependencies, created with Depends, are powerful. In tests, you often want to replace real dependencies with fake, in‑memory, or stub implementations. FastAPI provides app.dependency_overrides just for this.

Simple Dependency Override

Application code:

python 
# app/deps.py
from fastapi import Depends
def get_settings():
    return {"feature_x_enabled": True}
python 
# app/main.py
from fastapi import FastAPI, Depends
from .deps import get_settings
app = FastAPI()
@app.get("/feature")
def feature(settings: dict = Depends(get_settings)):
    if not settings["feature_x_enabled"]:
        return {"enabled": False}
    return {"enabled": True, "data": "Secret feature data"}

In tests:

python 
# tests/test_feature.py
from app.main import app
from fastapi.testclient import TestClient
def override_get_settings_disabled():
    return {"feature_x_enabled": False}
def test_feature_disabled():
    app.dependency_overrides.clear()
    from app.deps import get_settings
    app.dependency_overrides[get_settings] = override_get_settings_disabled
    client = TestClient(app)
    resp = client.get("/feature")
    assert resp.status_code == 200
    assert resp.json() == {"enabled": False}
def test_feature_enabled_default():
    app.dependency_overrides.clear()  # remove overrides
    client = TestClient(app)
    resp = client.get("/feature")
    assert resp.status_code == 200
    assert resp.json()["enabled"] is True

Always clear app.dependency_overrides between tests.
If you forget, overrides from one test can leak into another test and cause confusing failures.

Overriding Database Dependencies

You will often have something like this:

python 
# app/deps.py
from sqlalchemy.orm import Session
from .database import SessionLocal
def get_db() -> Session:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Route:

python 
# app/main.py
from fastapi import Depends
from .deps import get_db
from . import models
@app.get("/users")
def list_users(db = Depends(get_db)):
    return db.query(models.User).all()

In tests, you can override get_db to use a test database or even an in‑memory database. You will see this in more detail in the next section.

Testing with a Test Database

To test routes that use a real database, you should avoid using your production or development database directly. Instead, create a separate test database, usually in memory or in a separate schema.

Below is a common pattern with SQLAlchemy and SQLite.

Test Database Setup Example

Application database module:

python 
# app/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
SQLALCHEMY_DATABASE_URL = "sqlite:///./app.db"
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

A simple model:

python 
# app/models.py
from sqlalchemy import Column, Integer, String
from .database import Base
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)

Route using the database:

python 
# app/main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from .database import Base, engine
from .models import User
from .deps import get_db
Base.metadata.create_all(bind=engine)
app = FastAPI()
@app.post("/users", status_code=201)
def create_user(email: str, db: Session = Depends(get_db)):
    if db.query(User).filter(User.email == email).first():
        raise HTTPException(status_code=400, detail="Email already registered")
    user = User(email=email)
    db.add(user)
    db.commit()
    db.refresh(user)
    return user
@app.get("/users")
def list_users(db: Session = Depends(get_db)):
    return db.query(User).all()

Creating a Test Database and Overriding `get_db`

Test configuration:

python 
# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.database import Base
from app.main import app
from app.deps import get_db
# Use a separate SQLite file or in-memory DB for tests
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
# For in-memory: "sqlite:///:memory:"
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Create tables
Base.metadata.create_all(bind=engine)
def override_get_db():
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()
app.dependency_overrides[get_db] = override_get_db
@pytest.fixture
def client():
    with TestClient(app) as c:
        yield c

Now all routes that depend on get_db will receive a session bound to the test engine.

Testing Database‑Using Endpoints

python 
# tests/test_users.py
from sqlalchemy.orm import Session
from app.models import User
from app.database import Base
def test_create_user_success(client):
    resp = client.post("/users", params={"email": "alice@example.com"})
    assert resp.status_code == 201
    data = resp.json()
    assert data["email"] == "alice@example.com"
    assert "id" in data
def test_create_user_duplicate_email(client):
    # First create user
    client.post("/users", params={"email": "bob@example.com"})
    # Try duplicate
    resp = client.post("/users", params={"email": "bob@example.com"})
    assert resp.status_code == 400
    assert resp.json() == {"detail": "Email already registered"}
def test_list_users(client):
    # Start with a clean DB or ensure known state
    client.post("/users", params={"email": "c1@example.com"})
    client.post("/users", params={"email": "c2@example.com"})
    resp = client.get("/users")
    assert resp.status_code == 200
    users = resp.json()
    emails = {u["email"] for u in users}
    assert {"c1@example.com", "c2@example.com"} <= emails

Make sure your test database is isolated from development or production.
Never point tests at your production database.

For even better isolation, you can:

Testing Authentication and Protected Endpoints

Authentication logic frequently lives in dependencies, for example:

python 
# app/auth.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def get_current_user(token: str = Depends(oauth2_scheme)):
    if token != "secrettoken":
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
    return {"username": "testuser"}

Protected route:

python 
# app/main.py
from fastapi import Depends
from .auth import get_current_user
@app.get("/profile")
def read_profile(current_user: dict = Depends(get_current_user)):
    return {"username": current_user["username"], "bio": "Hello"}

Testing Without Authentication

python 
# tests/test_auth_profile.py
def test_profile_requires_auth(client):
    resp = client.get("/profile")
    assert resp.status_code == 401
    # FastAPI OAuth2PasswordBearer returns this structured error
    assert resp.json()["detail"] == "Not authenticated"

Testing With a Valid Token

python 
def test_profile_with_valid_token(client):
    headers = {"Authorization": "Bearer secrettoken"}
    resp = client.get("/profile", headers=headers)
    assert resp.status_code == 200
    assert resp.json()["username"] == "testuser"

Overriding `get_current_user` in Tests

Often, your real get_current_user will:

For simple tests, this can be unnecessary or slow. You can override get_current_user to return a fake user.

python 
# tests/conftest.py (additions)
from app.auth import get_current_user
def override_get_current_user():
    return {"username": "fakeuser", "is_admin": True}
app.dependency_overrides[get_current_user] = override_get_current_user

Then your tests do not need to send a real token, but the route sees a user object as usual.

Test:

python 
# tests/test_profile_override.py
def test_profile_with_overridden_user(client):
    resp = client.get("/profile")
    assert resp.status_code == 200
    assert resp.json()["username"] == "fakeuser"

This is a common pattern when testing authorization rules. You override a dependency to simulate:

Then you verify the endpoint behavior.

Testing Error Handling and Custom Exceptions

FastAPI allows you to define exception handlers and raise HTTPException or custom exceptions. You must ensure that:

Application code:

python 
# app/errors.py
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
class ItemNotFoundError(Exception):
    def __init__(self, item_id: int):
        self.item_id = item_id
def register_exception_handlers(app: FastAPI):
    @app.exception_handler(ItemNotFoundError)
    async def item_not_found_handler(request: Request, exc: ItemNotFoundError):
        return JSONResponse(
            status_code=404,
            content={"detail": f"Item {exc.item_id} not found"},
        )
python 
# app/main.py
from fastapi import FastAPI
from .errors import ItemNotFoundError, register_exception_handlers
app = FastAPI()
register_exception_handlers(app)
fake_items = {1: "Apple"}
@app.get("/items/{item_id}")
def read_item(item_id: int):
    if item_id not in fake_items:
        raise ItemNotFoundError(item_id)
    return {"id": item_id, "name": fake_items[item_id]}

Test:

python 
# tests/test_errors.py
def test_item_not_found_custom_handler(client):
    resp = client.get("/items/999")
    assert resp.status_code == 404
    assert resp.json() == {"detail": "Item 999 not found"}
def test_item_found(client):
    resp = client.get("/items/1")
    assert resp.status_code == 200
    assert resp.json() == {"id": 1, "name": "Apple"}

You can also test global exception handlers for 500 errors or validation errors if you customize them.

Testing Middleware and Request/Response Hooks

Middleware wraps your application and can log, modify, or reject requests and responses. You should test that it:

Example middleware that adds a custom header:

python 
# app/main.py
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    response = await call_next(request)
    response.headers["X-App-Version"] = "1.0.0"
    return response
@app.get("/ping")
def ping():
    return {"status": "ok"}

Test:

python 
# tests/test_middleware.py
def test_custom_header_added(client):
    resp = client.get("/ping")
    assert resp.status_code == 200
    assert resp.headers["X-App-Version"] == "1.0.0"

If your middleware uses state, for example increments a counter, you can:

Testing Async Endpoints

FastAPI endpoints can be defined as async def. TestClient handles them automatically. You can still write plain synchronous tests.

Example:

python 
# app/main.py
import asyncio
from fastapi import FastAPI
app = FastAPI()
@app.get("/async-hello")
async def async_hello():
    await asyncio.sleep(0.01)
    return {"message": "Hello from async"}

Test:

python 
# tests/test_async.py
def test_async_hello(client):
    resp = client.get("/async-hello")
    assert resp.status_code == 200
    assert resp.json() == {"message": "Hello from async"}

You do not need pytest-asyncio just to test FastAPI async endpoints through TestClient. FastAPI and Starlette manage the event loop internally.

If you want to test standalone async functions that are not called via HTTP, then you might use async test support from pytest.

Structuring Tests in a FastAPI Project

How you organize tests can make them easier to navigate. One simple pattern:

DirectoryPurpose
tests/Root test directory
tests/test_api_*.pyTests for API endpoints
tests/test_auth_*.pyAuth and security tests
tests/test_db_*.pyDatabase related tests
tests/test_integration_*.pyIntegration tests
tests/conftest.pyShared fixtures like client, DB

Example layout:

text 
app/
    main.py
    auth.py
    deps.py
    database.py
    models.py
tests/
    conftest.py
    test_items.py
    test_users.py
    test_auth.py
    test_errors.py

Some guidelines:

Putting It All Together: A Small Example

Here is a tiny FastAPI app with:

Then a set of tests that cover the main flows.

Application pieces (simplified):

python 
# app/schemas.py
from pydantic import BaseModel, EmailStr
class UserCreate(BaseModel):
    email: EmailStr
    password: str
class UserOut(BaseModel):
    id: int
    email: EmailStr
    class Config:
        orm_mode = True
python 
# app/auth.py
from fastapi import Depends, HTTPException, status
def get_current_user_id(auth_header: str | None = None):
    # Fake example: header is "User <id>"
    if not auth_header or not auth_header.startswith("User "):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid auth")
    try:
        return int(auth_header.split()[1])
    except ValueError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid auth")
python 
# app/main.py
from fastapi import FastAPI, Depends, Header, HTTPException, status
from sqlalchemy.orm import Session
from .database import Base, engine, SessionLocal
from .models import User
from .schemas import UserCreate, UserOut
from .auth import get_current_user_id
Base.metadata.create_all(bind=engine)
app = FastAPI()
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
@app.post("/users", response_model=UserOut, status_code=201)
def register(user_in: UserCreate, db: Session = Depends(get_db)):
    if db.query(User).filter(User.email == user_in.email).first():
        raise HTTPException(status_code=400, detail="Email taken")
    user = User(email=user_in.email, hashed_password="fakehash")
    db.add(user)
    db.commit()
    db.refresh(user)
    return user
@app.get("/me", response_model=UserOut)
def read_me(
    x_user_id: int = Depends(get_current_user_id),
    db: Session = Depends(get_db),
):
    user = db.query(User).filter(User.id == x_user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

Test configuration:

python 
# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.database import Base
from app.main import app, get_db
SQLALCHEMY_DATABASE_URL = "sqlite:///./test_fastapi.db"
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base.metadata.drop_all(bind=engine)
Base.metadata.create_all(bind=engine)
def override_get_db():
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()
app.dependency_overrides[get_db] = override_get_db
@pytest.fixture
def client():
    with TestClient(app) as c:
        yield c

Tests:

python 
# tests/test_user_flow.py
def test_register_user(client):
    body = {"email": "user@example.com", "password": "secret"}
    resp = client.post("/users", json=body)
    assert resp.status_code == 201
    data = resp.json()
    assert data["email"] == "user@example.com"
    assert "id" in data
def test_register_duplicate_email(client):
    body = {"email": "dup@example.com", "password": "secret"}
    client.post("/users", json=body)
    resp = client.post("/users", json=body)
    assert resp.status_code == 400
    assert resp.json()["detail"] == "Email taken"
def test_read_me_requires_auth(client):
    resp = client.get("/me")
    assert resp.status_code == 401
def test_read_me_success(client):
    # Create a user
    resp = client.post("/users", json={"email": "me@example.com", "password": "pwd"})
    user = resp.json()
    user_id = user["id"]
    # Send fake auth header that get_current_user_id understands
    headers = {"Authorization": f"User {user_id}"}
    resp2 = client.get("/me", headers=headers)
    assert resp2.status_code == 200
    assert resp2.json()["email"] == "me@example.com"

These tests exercise:

This combination is typical in real FastAPI projects.

Summary

In this chapter you learned how to test FastAPI applications effectively:

These patterns form the core of Testing FastAPI. Combined with the general testing practices from previous chapters, you can now build FastAPI apps with confidence that they work correctly.

Views: 8

Comments

Please login to add a comment.

Don't have an account? Register now!

© 2025 Kahibaro
Visitors: 3442 / 24h
Impressum / Imprint
Datenschutz / Data Policy