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.
Why Your Current Tests Probably Suck
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.
Setting Up pytest with FastAPI and PostgreSQL
Start with the right fixtures. Here's my baseline setup:
# tests/conftest.py
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
# Use an in-memory SQLite for speed, OR a real Postgres test database
# I prefer Postgres because SQLite doesn't catch all constraint violations
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.
Writing Tests That Actually Matter
Here's a real endpoint from CitizenApp—a multi-tenant API that creates workspaces:
# app/api/workspaces.py
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."""
# Check for duplicate names within user's workspaces
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:
# tests/test_workspaces.py
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."""
# In reality, you'd generate a valid JWT
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."""
# Create first workspace
workspace1 = Workspace(name="Duplicate", owner_id=test_user.id)
db_session.add(workspace1)
db_session.commit()
# Try to create duplicate
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()
# User A creates "My Workspace"
response_a = client.post(
"/workspaces",
json={"name": "My Workspace"},
headers=auth_header,
)
assert response_a.status_code == status.HTTP_200_OK
# User B also creates "My Workspace" (different owner_id)
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"]
Parametrized Tests for Edge Cases
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
Gotcha: Transaction Isolation and Async
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()
Running Tests in CI
Use GitHub Actions with a Postgres container:
yaml
# .github/workflows/test.yml
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
Top comments (0)