Testing FastAPI SaaS Applications
Untested code breaks at 3 AM in production. A SaaS backend handling real payments, user data, and multi-tenant isolation cannot afford bugs. This guide implements a testing strategy for FastAPI using pytest: unit tests for business logic, integration tests for database interactions, and fixtures for mocking dependencies. You'll achieve 80%+ coverage, catch bugs before deploy, and refactor with confidence.
Testing Pyramid: Unit, Integration, E2E
A testing pyramid has three layers (bottom to top): unit tests (70%), integration tests (20%), end-to-end tests (10%). Unit tests are fast and catch logic bugs. Integration tests verify database and API interactions. E2E tests are slow but catch real user flows:
/\
/ \ E2E: 10% (slow, real browser/curl, whole system)
/____\
/ \
/ \ Integration: 20% (database, external APIs mocked)
/________\
/ \
/ \ Unit: 70% (business logic, dependencies mocked)
/_____________\
This article focuses on unit and integration; E2E is typically manual or with tools like Playwright.
Setting Up Pytest
Install testing dependencies:
pip install pytest pytest-asyncio pytest-cov httpx
Create a conftest.py at the project root with shared fixtures:
# conftest.py
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from app.main import app
from app.models import Base
from app.database import get_db
# Use in-memory SQLite for tests (fast, isolated)
@pytest.fixture(scope="function")
def test_db():
"""Create a fresh test database for each test."""
engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)
SessionLocal = sessionmaker(bind=engine)
session = SessionLocal()
yield session
session.close()
engine.dispose()
@pytest.fixture
def client(test_db: Session):
"""
FastAPI TestClient with dependencies mocked.
Overrides get_db to use test database.
"""
def override_get_db():
yield test_db
app.dependency_overrides[get_db] = override_get_db
yield TestClient(app)
app.dependency_overrides.clear()
@pytest.fixture
def authenticated_client(client: TestClient, test_db: Session):
"""TestClient with an authenticated user (JWT token in header)."""
from app.models import Tenant, User
from app.auth import create_access_token
# Create a test tenant and user
tenant = Tenant(id=1, slug="test", name="Test Tenant")
test_db.add(tenant)
test_db.flush()
user = User(
id=1,
tenant_id=1,
email="[email protected]",
password_hash="hashed_password",
first_name="Test",
last_name="User",
role="admin"
)
test_db.add(user)
test_db.commit()
# Create JWT token
token = create_access_token(
user_id=user.id,
tenant_id=tenant.id,
role=user.role
)
# Add token to client headers
client.headers = {"Authorization": f"Bearer {token}"}
return client
Unit Tests: Testing Business Logic
Unit tests verify individual functions in isolation with mocked dependencies:
# tests/test_auth.py
import pytest
from app.auth import hash_password, verify_password, create_access_token, decode_token
class TestPasswordHashing:
"""Test password hashing and verification."""
def test_hash_password_creates_different_hash(self):
"""Hashing the same password produces different hashes (salt)."""
password = "securePassword123"
hash1 = hash_password(password)
hash2 = hash_password(password)
assert hash1 != hash2
assert verify_password(password, hash1)
assert verify_password(password, hash2)
def test_verify_password_rejects_wrong_password(self):
"""Wrong password fails verification."""
hashed = hash_password("correctPassword")
assert not verify_password("wrongPassword", hashed)
def test_hash_prevents_dictionary_attacks(self):
"""Hashing is slow by design (bcrypt has cost factor)."""
import time
password = "testPassword"
start = time.time()
hashed = hash_password(password)
elapsed = time.time() - start
assert elapsed > 0.1 # Should take at least 100ms
assert verify_password(password, hashed)
class TestTokens:
"""Test JWT token generation and validation."""
def test_create_access_token(self):
"""Create a valid JWT token with claims."""
token = create_access_token(
user_id=42,
tenant_id=1,
role="admin"
)
token_data = decode_token(token)
assert token_data.user_id == 42
assert token_data.tenant_id == 1
assert token_data.role == "admin"
def test_decode_invalid_token_returns_none(self):
"""Invalid or tampered token returns None."""
invalid_token = "not.a.valid.token"
assert decode_token(invalid_token) is None
def test_expired_token_returns_none(self):
"""Token past expiration returns None."""
from datetime import timedelta
token = create_access_token(
user_id=1,
tenant_id=1,
role="user",
expires_delta=timedelta(seconds=-1) # Expired 1 second ago
)
assert decode_token(token) is None
Integration Tests: Testing API Endpoints
Integration tests verify FastAPI endpoints work with the database:
# tests/test_endpoints.py
import pytest
from fastapi.testclient import TestClient
class TestAuth:
"""Test authentication endpoints."""
def test_login_success(self, client: TestClient, test_db: Session):
"""Successful login returns tokens."""
from app.models import Tenant, User
from app.auth import hash_password
# Create test user
tenant = Tenant(id=1, slug="test", name="Test Tenant")
test_db.add(tenant)
test_db.flush()
user = User(
id=1,
tenant_id=1,
email="[email protected]",
password_hash=hash_password("password123"),
first_name="John",
last_name="Doe"
)
test_db.add(user)
test_db.commit()
# Login
response = client.post("/auth/login", json={
"email": "[email protected]",
"password": "password123"
})
assert response.status_code == 200
data = response.json()
assert "access_token" in data
assert "refresh_token" in data
assert data["token_type"] == "bearer"
def test_login_invalid_password(self, client: TestClient, test_db: Session):
"""Login with wrong password fails."""
from app.models import Tenant, User
from app.auth import hash_password
tenant = Tenant(id=1, slug="test", name="Test")
test_db.add(tenant)
test_db.flush()
user = User(
id=1,
tenant_id=1,
email="[email protected]",
password_hash=hash_password("correctPassword"),
first_name="John",
last_name="Doe"
)
test_db.add(user)
test_db.commit()
response = client.post("/auth/login", json={
"email": "[email protected]",
"password": "wrongPassword"
})
assert response.status_code == 401
assert "Invalid" in response.json()["detail"]
class TestUsers:
"""Test user management endpoints."""
def test_create_user_requires_auth(self, client: TestClient):
"""Creating a user without auth token fails."""
response = client.post("/users", json={
"email": "[email protected]",
"first_name": "Jane",
"last_name": "Smith"
})
assert response.status_code == 403 # Forbidden
def test_create_user_authenticated(self, authenticated_client: TestClient, test_db: Session):
"""Authenticated user can create a new user in their tenant."""
response = authenticated_client.post("/users", json={
"email": "[email protected]",
"first_name": "Jane",
"last_name": "Smith"
})
assert response.status_code == 201
data = response.json()
assert data["email"] == "[email protected]"
assert data["tenant_id"] == 1 # Same tenant as authenticated user
def test_list_users_shows_only_current_tenant(self, authenticated_client: TestClient, test_db: Session):
"""Users list is filtered by tenant (multi-tenant isolation)."""
from app.models import User
from app.auth import hash_password
# Add another user to same tenant
user2 = User(
id=2,
tenant_id=1,
email="[email protected]",
password_hash=hash_password("pass"),
first_name="Jane",
last_name="Smith"
)
test_db.add(user2)
test_db.commit()
response = authenticated_client.get("/users")
assert response.status_code == 200
users = response.json()
assert len(users) == 2
assert all(u["tenant_id"] == 1 for u in users)
Testing Async Functions
Use pytest-asyncio for async tests:
# tests/test_async.py
import pytest
import httpx
@pytest.mark.asyncio
async def test_async_http_client(authenticated_client):
"""Test async HTTP operations."""
async def fetch_profile(user_id: int):
async with httpx.AsyncClient() as client:
response = await client.get(f"https://api.example.com/users/{user_id}")
return response.json()
# Mock the external API
from unittest.mock import AsyncMock, patch
with patch("httpx.AsyncClient.get") as mock_get:
mock_response = AsyncMock()
mock_response.json = AsyncMock(return_value={"id": 1, "name": "John"})
mock_get.return_value = mock_response
result = await fetch_profile(1)
assert result["name"] == "John"
Mocking External Dependencies
Mock Stripe, emails, and external APIs:
# tests/test_billing.py
from unittest.mock import patch, MagicMock
class TestBilling:
"""Test Stripe integration."""
@patch("stripe.Customer.create")
def test_create_stripe_customer(self, mock_stripe_create, authenticated_client, test_db):
"""Verify Stripe customer creation is called."""
mock_stripe_create.return_value = MagicMock(id="cus_12345")
response = authenticated_client.post("/billing/create-customer")
assert response.status_code == 200
assert response.json()["stripe_customer_id"] == "cus_12345"
mock_stripe_create.assert_called_once()
@patch("tasks.send_email.delay")
def test_send_welcome_email_queues_task(self, mock_celery_task, authenticated_client):
"""Verify email task is queued (not sent directly)."""
mock_celery_task.return_value = MagicMock(id="task_123")
response = authenticated_client.post(
"/users/1/send-welcome-email"
)
assert response.status_code == 200
mock_celery_task.assert_called_once()
Parametrized Tests
Test multiple inputs with a single test function:
# tests/test_validation.py
import pytest
class TestUserValidation:
"""Test user input validation."""
@pytest.mark.parametrize("email", [
"[email protected]",
"[email protected]",
"[email protected]"
])
def test_valid_emails_accepted(self, authenticated_client, email):
"""Valid emails are accepted."""
response = authenticated_client.post("/users", json={
"email": email,
"first_name": "Test",
"last_name": "User"
})
assert response.status_code == 201
@pytest.mark.parametrize("email", [
"invalid",
"@example.com",
"[email protected]",
"user @example.com" # Space
])
def test_invalid_emails_rejected(self, authenticated_client, email):
"""Invalid emails are rejected."""
response = authenticated_client.post("/users", json={
"email": email,
"first_name": "Test",
"last_name": "User"
})
assert response.status_code == 422 # Validation error
Coverage Reporting
Measure test coverage:
# Generate coverage report
pytest --cov=app --cov-report=html tests/
# View HTML report
open htmlcov/index.html
Target 80%+ coverage. Exclude obvious code (e.g., pass, data models). Add CI check to fail if coverage drops:
# .github/workflows/tests.yml
- name: Run tests with coverage
run: pytest --cov=app --cov-fail-under=80 tests/
Key Takeaways
- Use pytest fixtures to set up test databases and authenticated clients; avoid repeating setup code.
- Unit tests verify business logic (auth, hashing); integration tests verify endpoints and database interactions.
- Mock external dependencies (Stripe, email) to avoid side effects during testing.
- Parametrize tests to check multiple inputs with a single test function.
- Target 80%+ coverage; use CI to enforce coverage gates before merging.
Frequently Asked Questions
Should I test private methods?
No. Test public APIs and behavior. Private methods are implementation details; if they need testing, consider making them public or refactoring. Test behavior, not implementation.
How do I test database migrations?
Run migrations on test database, then verify schema: alembic upgrade head during test setup. Use fixtures to revert migrations between tests.
Should I use real Redis in tests or mock it?
For unit tests, mock Redis. For integration tests, use a test instance (Docker or ephemeral) or mock. In CI, use a Docker service (testcontainers or GitHub Actions services).
Can I test FastAPI WebSocket endpoints?
Yes, use TestClient with with client.websocket_connect("/ws") as ws: to simulate WebSocket connections. Send/receive messages and assert on responses.
How do I test Celery tasks?
Use CELERY_TASK_ALWAYS_EAGER = True in test config to run tasks synchronously. Tasks execute immediately, making tests deterministic.