API Testing Patterns
Comprehensive API testing with pytest and FastAPI
PRINCIPLES
- •Test Isolation: Each test independent, no shared state
- •Fixture Composition: Build complex fixtures from simple ones
- •Async Testing: Use pytest-asyncio for async code
- •Realistic Data: Use factories for test data
- •Coverage Goals: Test happy paths, edge cases, and errors
PYTEST SETUP
Configuration
python
# pyproject.toml
[tool.pytest.ini_options]
minversion = "7.0"
addopts = "-ra -q --strict-markers"
testpaths = ["tests"]
asyncio_mode = "auto"
filterwarnings = [
"ignore::DeprecationWarning",
]
[tool.coverage.run]
source = ["app"]
omit = ["*/tests/*", "*/__init__.py"]
branch = true
[tool.coverage.report]
exclude_lines = [
"pragma: no cover",
"if TYPE_CHECKING:",
"raise NotImplementedError",
]
CONFTEST FIXTURES
Base Test Configuration
python
# tests/conftest.py
import pytest
from typing import AsyncGenerator
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from app.main import app
from app.db.session import get_db
from app.db.base import Base
TEST_DATABASE_URL = "postgresql+asyncpg://test:test@localhost/test_db"
@pytest.fixture(scope="session")
def anyio_backend():
return "asyncio"
@pytest.fixture(scope="session")
async def engine():
engine = create_async_engine(TEST_DATABASE_URL, echo=True)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield engine
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
await engine.dispose()
@pytest.fixture
async def db_session(engine) -> AsyncGenerator[AsyncSession, None]:
async_session_maker = async_sessionmaker(
engine,
class_=AsyncSession,
expire_on_commit=False,
)
async with async_session_maker() as session:
yield session
await session.rollback()
@pytest.fixture
async def client(db_session: AsyncSession) -> AsyncGenerator[AsyncClient, None]:
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 client:
yield client
app.dependency_overrides.clear()
@pytest.fixture
async def authenticated_client(client: AsyncClient, test_user) -> AsyncClient:
response = await client.post(
"/auth/token",
data={"username": test_user.email, "password": "testpassword"},
)
token = response.json()["access_token"]
client.headers["Authorization"] = f"Bearer {token}"
return client
TEST FACTORIES
Data Generation
python
# tests/factories.py
from typing import Any
import factory
from faker import Faker
from app.models import User, Post
from passlib.hash import bcrypt
fake = Faker()
class UserFactory:
"""Factory for creating test users."""
@staticmethod
def build(**kwargs) -> dict[str, Any]:
defaults = {
"email": fake.email(),
"full_name": fake.name(),
"hashed_password": bcrypt.hash("testpassword"),
"is_active": True,
}
defaults.update(kwargs)
return defaults
@staticmethod
async def create(db: AsyncSession, **kwargs) -> User:
data = UserFactory.build(**kwargs)
user = User(**data)
db.add(user)
await db.flush()
await db.refresh(user)
return user
class PostFactory:
"""Factory for creating test posts."""
@staticmethod
def build(**kwargs) -> dict[str, Any]:
defaults = {
"title": fake.sentence(),
"content": fake.text(max_nb_chars=500),
}
defaults.update(kwargs)
return defaults
@staticmethod
async def create(db: AsyncSession, author: User, **kwargs) -> Post:
data = PostFactory.build(author_id=author.id, **kwargs)
post = Post(**data)
db.add(post)
await db.flush()
await db.refresh(post)
return post
# Fixtures using factories
@pytest.fixture
async def test_user(db_session: AsyncSession) -> User:
return await UserFactory.create(
db_session,
email="test@example.com",
)
@pytest.fixture
async def test_posts(db_session: AsyncSession, test_user: User) -> list[Post]:
posts = []
for _ in range(5):
post = await PostFactory.create(db_session, author=test_user)
posts.append(post)
return posts
ENDPOINT TESTING
CRUD Test Patterns
python
# tests/api/test_users.py
import pytest
from httpx import AsyncClient
class TestUserEndpoints:
"""Test user API endpoints."""
async def test_create_user(self, client: AsyncClient):
response = await client.post(
"/api/v1/users/",
json={
"email": "newuser@example.com",
"password": "SecurePass123",
"full_name": "New User",
},
)
assert response.status_code == 201
data = response.json()
assert data["email"] == "newuser@example.com"
assert "id" in data
assert "password" not in data # Never expose password
async def test_create_user_duplicate_email(
self,
client: AsyncClient,
test_user,
):
response = await client.post(
"/api/v1/users/",
json={
"email": test_user.email, # Already exists
"password": "SecurePass123",
},
)
assert response.status_code == 400
assert "already exists" in response.json()["detail"]
async def test_get_user(
self,
client: AsyncClient,
test_user,
):
response = await client.get(f"/api/v1/users/{test_user.id}")
assert response.status_code == 200
assert response.json()["email"] == test_user.email
async def test_get_user_not_found(self, client: AsyncClient):
response = await client.get("/api/v1/users/99999")
assert response.status_code == 404
async def test_list_users_pagination(
self,
client: AsyncClient,
db_session,
):
# Create multiple users
for i in range(15):
await UserFactory.create(db_session)
response = await client.get("/api/v1/users/?skip=0&limit=10")
assert response.status_code == 200
assert len(response.json()) == 10
response = await client.get("/api/v1/users/?skip=10&limit=10")
assert len(response.json()) == 5
async def test_update_user_authenticated(
self,
authenticated_client: AsyncClient,
test_user,
):
response = await authenticated_client.patch(
f"/api/v1/users/{test_user.id}",
json={"full_name": "Updated Name"},
)
assert response.status_code == 200
assert response.json()["full_name"] == "Updated Name"
async def test_delete_user_unauthorized(
self,
client: AsyncClient,
test_user,
):
response = await client.delete(f"/api/v1/users/{test_user.id}")
assert response.status_code == 401
MOCKING
External Dependencies
python
from unittest.mock import AsyncMock, patch
import pytest
class TestExternalServices:
async def test_send_email(self, client: AsyncClient):
with patch("app.services.email.send_email") as mock_send:
mock_send.return_value = True
response = await client.post(
"/api/v1/users/forgot-password",
json={"email": "test@example.com"},
)
assert response.status_code == 200
mock_send.assert_called_once()
async def test_external_api_call(self, client: AsyncClient):
mock_response = {"status": "success", "data": {"id": 123}}
with patch("httpx.AsyncClient.get", new_callable=AsyncMock) as mock_get:
mock_get.return_value.json.return_value = mock_response
mock_get.return_value.status_code = 200
response = await client.get("/api/v1/external-data")
assert response.status_code == 200
assert response.json()["external_id"] == 123
VALIDATION TESTING
Input Validation
python
@pytest.mark.parametrize(
"payload,expected_status,error_field",
[
({"email": "invalid"}, 422, "email"),
({"email": "test@example.com", "password": "short"}, 422, "password"),
({"password": "ValidPass123"}, 422, "email"),
({}, 422, "email"),
],
)
async def test_create_user_validation(
client: AsyncClient,
payload: dict,
expected_status: int,
error_field: str,
):
response = await client.post("/api/v1/users/", json=payload)
assert response.status_code == expected_status
errors = response.json()["detail"]
assert any(error_field in str(e) for e in errors)
ANTI-PATTERNS
❌ AVOID:
python
# Shared mutable state
users = []
def test_one():
users.append(create_user())
def test_two():
assert len(users) == 0 # Fails!
✅ PREFER:
python
# Fixture isolation
@pytest.fixture
async def users(db_session):
return [await UserFactory.create(db_session)]
async def test_one(users):
assert len(users) == 1
async def test_two(users): # Fresh fixture
assert len(users) == 1