Introduction#
FastAPI’s design makes it highly testable. This post covers the testing pyramid for FastAPI applications: unit tests for business logic, integration tests for the full HTTP stack with a real database, and contract tests for API compatibility.
Test Client Setup#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
| # conftest.py
import asyncio
import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from app.main import app
from app.database import Base, get_db
# Use a separate test database
TEST_DB_URL = "postgresql+asyncpg://test:test@localhost:5432/test_db"
@pytest.fixture(scope="session")
def event_loop():
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
@pytest_asyncio.fixture(scope="function")
async def db_session():
engine = create_async_engine(TEST_DB_URL, echo=False)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async_session = sessionmaker(engine, class_=AsyncSession)
async with async_session() as session:
yield session
await session.rollback() # roll back after each test
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
@pytest_asyncio.fixture
async def client(db_session):
# Override the database dependency with the test session
async def override_get_db():
yield db_session
app.dependency_overrides[get_db] = override_get_db
async with AsyncClient(
transport=ASGITransport(app=app),
base_url="http://test"
) as c:
yield c
app.dependency_overrides.clear()
|
Unit Testing Business Logic#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
| # app/services/order_service.py
from decimal import Decimal
from app.models import Order, OrderItem
def calculate_order_total(items: list[OrderItem], discount_pct: float = 0.0) -> Decimal:
subtotal = sum(item.price * item.quantity for item in items)
discount = subtotal * Decimal(str(discount_pct))
return subtotal - discount
# tests/unit/test_order_service.py
from decimal import Decimal
import pytest
from app.services.order_service import calculate_order_total
from app.models import OrderItem
def test_calculate_order_total_no_discount():
items = [
OrderItem(price=Decimal("10.00"), quantity=2),
OrderItem(price=Decimal("5.50"), quantity=1),
]
result = calculate_order_total(items)
assert result == Decimal("25.50")
def test_calculate_order_total_with_discount():
items = [OrderItem(price=Decimal("100.00"), quantity=1)]
result = calculate_order_total(items, discount_pct=0.10)
assert result == Decimal("90.00")
def test_calculate_order_total_empty():
assert calculate_order_total([]) == Decimal("0")
|
Integration Tests: HTTP + Database#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
| # tests/integration/test_orders.py
import pytest
import pytest_asyncio
@pytest.mark.asyncio
async def test_create_order_success(client, db_session):
payload = {
"user_id": 1,
"items": [
{"product_id": 1, "quantity": 2, "price": 14.99}
]
}
response = await client.post("/api/v1/orders", json=payload)
assert response.status_code == 201
data = response.json()
assert data["user_id"] == 1
assert data["status"] == "pending"
assert abs(data["total"] - 29.98) < 0.01
assert "id" in data
@pytest.mark.asyncio
async def test_create_order_invalid_quantity(client):
payload = {
"user_id": 1,
"items": [{"product_id": 1, "quantity": -1, "price": 14.99}]
}
response = await client.post("/api/v1/orders", json=payload)
assert response.status_code == 422
@pytest.mark.asyncio
async def test_get_order_not_found(client):
response = await client.get("/api/v1/orders/99999")
assert response.status_code == 404
@pytest.mark.asyncio
async def test_list_orders_pagination(client, db_session):
# Create 15 orders
for i in range(15):
await client.post("/api/v1/orders", json={
"user_id": 1,
"items": [{"product_id": i, "quantity": 1, "price": 1.00}]
})
response = await client.get("/api/v1/orders?page=1&size=10")
assert response.status_code == 200
data = response.json()
assert len(data["items"]) == 10
assert data["total"] == 15
assert data["pages"] == 2
|
Mocking External Services#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
| # tests/integration/test_payment.py
from unittest.mock import AsyncMock, patch
import pytest
@pytest.mark.asyncio
async def test_process_payment_success(client):
with patch("app.services.payment.stripe_client.charge") as mock_charge:
mock_charge.return_value = AsyncMock(
return_value={"id": "ch_123", "status": "succeeded"}
)()
response = await client.post("/api/v1/payments", json={
"order_id": 1,
"card_token": "tok_visa",
"amount": 99.99
})
assert response.status_code == 200
mock_charge.assert_called_once()
@pytest.mark.asyncio
async def test_process_payment_card_declined(client):
import stripe
with patch("app.services.payment.stripe_client.charge") as mock_charge:
mock_charge.side_effect = stripe.CardError(
"Card declined", "card_declined", "card_declined"
)
response = await client.post("/api/v1/payments", json={
"order_id": 1,
"card_token": "tok_chargeDeclined",
"amount": 99.99
})
assert response.status_code == 402
assert "declined" in response.json()["detail"].lower()
|
Authentication in Tests#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
| # conftest.py: provide an authenticated client
import jwt
from datetime import datetime, timedelta
def create_test_token(user_id: int, roles: list[str] = None) -> str:
payload = {
"sub": str(user_id),
"roles": roles or ["user"],
"exp": datetime.utcnow() + timedelta(hours=1),
}
return jwt.encode(payload, "test-secret", algorithm="HS256")
@pytest_asyncio.fixture
async def auth_client(client):
token = create_test_token(user_id=1, roles=["admin"])
client.headers["Authorization"] = f"Bearer {token}"
return client
@pytest_asyncio.fixture
async def user_client(client):
token = create_test_token(user_id=42, roles=["user"])
client.headers["Authorization"] = f"Bearer {token}"
return client
# In tests:
async def test_delete_order_requires_admin(user_client):
response = await user_client.delete("/api/v1/orders/1")
assert response.status_code == 403
|
Snapshot Testing for JSON Responses#
1
2
3
4
5
6
7
8
| # Using syrupy for snapshot tests
from syrupy.assertion import SnapshotAssertion
async def test_order_response_shape(client, snapshot: SnapshotAssertion):
response = await client.post("/api/v1/orders", json={...})
assert response.json() == snapshot
# First run: creates __snapshots__/test_orders.ambr
# Subsequent runs: compares against saved snapshot
|
Conclusion#
Use httpx.AsyncClient with ASGI transport for realistic HTTP integration tests. Override database dependencies per test with a transaction that rolls back after each test — this provides isolation without recreating the schema each time. Mock external services (Stripe, email) at the client level, not deep in the business logic. Test error paths (400, 404, 403) as rigorously as happy paths.