I've shipped CitizenApp with nine AI features in production, and I can tell you with absolute certainty: bad tests in FastAPI will destroy your confidence faster than any outage. I'm not talking about unit tests that mock everything into oblivion—those are security theater. I'm talking about real tests that hit PostgreSQL, verify transactional behavior, and catch the subtle data corruption bugs that only appear under load.
Here's what actually works, and why most FastAPI test guides miss the point entirely.
Most tutorials show you how to mock the database. That's fine for toy projects. In production, I've watched carefully mocked unit tests pass while the actual endpoint corrupted customer data because of transaction isolation issues, cascading deletes, or constraint violations that only emerge with real schema interactions.
The problem: when you mock everything, you're testing your assumptions, not your code. Your assumptions are usually wrong.
I prefer integration tests that use a real PostgreSQL database (or a container) because they catch the bugs that matter. Yes, they're slower. No, you don't run them on every keystroke. But they're the difference between shipping with confidence and shipping with prayers.
Start with the right fixtures. Here's my baseline setup:
import os
import pytest
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.pool import StaticPool
from app.db import Base
from app.main import app
from fastapi.testclient import TestClient
DATABASE_URL = os.getenv(
"TEST_DATABASE_URL",
"postgresql://postgres:password@localhost:5432/test_db"
)
engine = create_engine(
DATABASE_URL,
echo=False, # Set to True to see SQL statements
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
@pytest.fixture(scope="session")
def db_engine():
"""Create tables once per test session."""
Base.metadata.create_all(bind=engine)
yield engine
Base.metadata.drop_all(bind=engine)
@pytest.fixture
def db_session(db_engine):
"""
Create a new database session for each test.
Rollback after the test to keep the database clean.
"""
connection = db_engine.connect()
transaction = connection.begin()
session = TestingSessionLocal(bind=connection)
yield session
session.close()
transaction.rollback()
connection.close()
@pytest.fixture
def client(db_session: Session):
"""Override the dependency injection to use our test session."""
def override_get_db():
yield db_session
from app.main import get_db
app.dependency_overrides[get_db] = override_get_db
yield TestClient(app)
app.dependency_overrides.clear()
Why this structure? The transaction rollback is critical. Each test runs inside a transaction that gets rolled back after the test completes. This keeps your test database clean without dropping tables between tests, and it's fast.
Here's a real endpoint from CitizenApp—a multi-tenant API that creates workspaces:
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.db import get_db
from app.models import Workspace, User
from app.schemas import WorkspaceCreate, WorkspaceResponse
router = APIRouter()
@router.post("/workspaces", response_model=WorkspaceResponse)
async def create_workspace(
workspace: WorkspaceCreate,
user: User = Depends(get_current_user),
db: Session = Depends(get_db),
):
"""Create a workspace and assign the user as owner."""
existing = db.query(Workspace).filter(
Workspace.owner_id == user.id,
Workspace.name == workspace.name,
).first()
if existing:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Workspace name already exists",
)
new_workspace = Workspace(
name=workspace.name,
owner_id=user.id,
)
db.add(new_workspace)
db.commit()
db.refresh(new_workspace)
return new_workspace
Now, here's the test that catches real bugs:
import pytest
from fastapi import status
from app.models import User, Workspace
from app.schemas import WorkspaceCreate
@pytest.fixture
def test_user(db_session):
"""Create a test user."""
user = User(email="test@example.com", hashed_password="fake_hash")
db_session.add(user)
db_session.commit()
db_session.refresh(user)
return user
@pytest.fixture
def auth_header(test_user):
"""Return auth header for test user."""
return {"Authorization": f"Bearer {test_user.id}"}
def test_create_workspace_success(client, test_user, auth_header):
"""Happy path: create a workspace."""
response = client.post(
"/workspaces",
json={"name": "My First Workspace"},
headers=auth_header,
)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert data["name"] == "My First Workspace"
assert data["owner_id"] == test_user.id
def test_create_duplicate_workspace_name(client, test_user, auth_header, db_session):
"""This test catches a real bug: duplicate name in same user's workspace."""
workspace1 = Workspace(name="Duplicate", owner_id=test_user.id)
db_session.add(workspace1)
db_session.commit()
response = client.post(
"/workspaces",
json={"name": "Duplicate"},
headers=auth_header,
)
assert response.status_code == status.HTTP_409_CONFLICT
def test_different_users_can_have_same_workspace_name(
client, db_session, auth_header
):
"""Multi-tenant isolation: user A and user B can both have 'Workspace'."""
user_b = User(email="user_b@example.com", hashed_password="hash")
db_session.add(user_b)
db_session.commit()
response_a = client.post(
"/workspaces",
json={"name": "My Workspace"},
headers=auth_header,
)
assert response_a.status_code == status.HTTP_200_OK
auth_header_b = {"Authorization": f"Bearer {user_b.id}"}
response_b = client.post(
"/workspaces",
json={"name": "My Workspace"},
headers=auth_header_b,
)
assert response_b.status_code == status.HTTP_200_OK
assert response_a.json()["id"] != response_b.json()["id"]
When you have 9 AI features, you need to test variations efficiently:
@pytest.mark.parametrize(
"workspace_name,expected_status",
[
("Valid Workspace", 200),
("", 422), # Empty name
("x" * 256, 200), # Very long name (assuming your schema allows it)
(None, 422), # Missing field
],
)
def test_workspace_name_validation(
client, test_user, auth_header, workspace_name, expected_status
):
"""Test name validation with multiple inputs."""
response = client.post(
"/workspaces",
json={"name": workspace_name} if workspace_name is not None else {},
headers=auth_header,
)
assert response.status_code == expected_status
Here's where I burned an hour: if you're using async endpoints with asyncio
in tests, the transaction rollback might not work as expected. FastAPI's TestClient runs async in a special way.
My workaround:
@pytest.fixture
def client(db_session: Session):
"""For async endpoints, ensure the session is thread-safe."""
import asyncio
from sqlalchemy.orm import scoped_session
scoped = scoped_session(TestingSessionLocal)
def override_get_db():
try:
yield scoped()
finally:
scoped.remove()
from app.main import get_db
app.dependency_overrides[get_db] = override_get_db
yield TestClient(app)
app.dependency_overrides.clear()
Use GitHub Actions with a Postgres container:
yaml
name: Tests
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-latest
services:
postgres:
image: postgres:15
env:
POSTGRES_PASSWORD: postgres
POSTGRES_DB: test_db
options: >-
--health-cmd pg_isready
--health-interval 10s
--health-timeout 5s
--health-retries 5
ports:
- 5432:5432
steps:
- uses: actions/checkout@v3
- uses: actions/setup-python@v4
with:
python