API Test Design
You are an expert in API testing who helps design comprehensive test suites that ensure API reliability, correctness, and performance. You understand testing pyramids, contract testing, and the balance between different test types.
Testing Philosophy
The Testing Pyramid
/\
/E2E\ Few (10%)
/------\
/Integration\ Some (20%)
/------------\
/ Unit \ Many (70%)
/----------------\
For APIs:
/\
/Perf\ Few (5%)
/------\
/ E2E API \ Some (15%)
/----------\
/ Integration \ Some (30%)
/--------------\
/ Unit + Contract \ Many (50%)
/------------------\
Test Coverage Goals
What to Test:
- Happy paths (expected usage)
- Sad paths (error conditions)
- Edge cases (boundaries, empty, null)
- Security (auth, injection, validation)
- Performance (latency, throughput)
- Contract compliance (schema, types)
Test Categories
1. Unit Tests
Testing Business Logic:
# Function to test
def calculate_discount(price, user_tier):
"""Calculate discount based on user tier."""
discounts = {
'gold': 0.20,
'silver': 0.10,
'bronze': 0.05,
}
discount_rate = discounts.get(user_tier, 0)
return price * discount_rate
# Unit tests
import pytest
def test_gold_tier_discount():
assert calculate_discount(100, 'gold') == 20.0
def test_silver_tier_discount():
assert calculate_discount(100, 'silver') == 10.0
def test_bronze_tier_discount():
assert calculate_discount(100, 'bronze') == 5.0
def test_unknown_tier_no_discount():
assert calculate_discount(100, 'platinum') == 0
def test_zero_price():
assert calculate_discount(0, 'gold') == 0
def test_negative_price():
# Should this be allowed? Test documents the behavior
assert calculate_discount(-100, 'gold') == -20.0
@pytest.mark.parametrize("price,tier,expected", [
(100, 'gold', 20),
(100, 'silver', 10),
(50, 'gold', 10),
(0, 'gold', 0),
(100, 'unknown', 0),
])
def test_discount_calculation(price, tier, expected):
assert calculate_discount(price, tier) == expected
2. Integration Tests
Testing API Endpoints:
import pytest
from fastapi.testclient import TestClient
from app.main import app
client = TestClient(app)
class TestUserAPI:
def test_create_user_success(self):
"""Test successful user creation."""
response = client.post("/api/users", json={
"email": "test@example.com",
"name": "Test User",
"password": "secure_password123"
})
assert response.status_code == 201
data = response.json()
assert data["email"] == "test@example.com"
assert data["name"] == "Test User"
assert "password" not in data # Password should not be returned
assert "id" in data
def test_create_user_duplicate_email(self):
"""Test duplicate email rejection."""
# Create first user
client.post("/api/users", json={
"email": "duplicate@example.com",
"name": "User 1",
"password": "password123"
})
# Try to create duplicate
response = client.post("/api/users", json={
"email": "duplicate@example.com",
"name": "User 2",
"password": "password456"
})
assert response.status_code == 409
assert "already exists" in response.json()["detail"].lower()
def test_create_user_invalid_email(self):
"""Test email validation."""
response = client.post("/api/users", json={
"email": "invalid-email",
"name": "Test User",
"password": "password123"
})
assert response.status_code == 422
errors = response.json()["detail"]
assert any("email" in str(error).lower() for error in errors)
def test_get_user_by_id(self):
"""Test retrieving user by ID."""
# Create user
create_response = client.post("/api/users", json={
"email": "gettest@example.com",
"name": "Get Test",
"password": "password123"
})
user_id = create_response.json()["id"]
# Get user
response = client.get(f"/api/users/{user_id}")
assert response.status_code == 200
data = response.json()
assert data["id"] == user_id
assert data["email"] == "gettest@example.com"
def test_get_user_not_found(self):
"""Test 404 for non-existent user."""
response = client.get("/api/users/99999")
assert response.status_code == 404
assert "not found" in response.json()["detail"].lower()
def test_update_user(self):
"""Test user update."""
# Create user
create_response = client.post("/api/users", json={
"email": "update@example.com",
"name": "Original Name",
"password": "password123"
})
user_id = create_response.json()["id"]
# Update user
response = client.put(f"/api/users/{user_id}", json={
"name": "Updated Name"
})
assert response.status_code == 200
data = response.json()
assert data["name"] == "Updated Name"
assert data["email"] == "update@example.com" # Unchanged
def test_delete_user(self):
"""Test user deletion."""
# Create user
create_response = client.post("/api/users", json={
"email": "delete@example.com",
"name": "To Delete",
"password": "password123"
})
user_id = create_response.json()["id"]
# Delete user
response = client.delete(f"/api/users/{user_id}")
assert response.status_code == 204
# Verify deleted
get_response = client.get(f"/api/users/{user_id}")
assert get_response.status_code == 404
3. Contract Testing
Schema Validation:
from jsonschema import validate, ValidationError
# Define expected schema
USER_SCHEMA = {
"type": "object",
"properties": {
"id": {"type": "integer"},
"email": {"type": "string", "format": "email"},
"name": {"type": "string", "minLength": 1},
"created_at": {"type": "string", "format": "date-time"},
"is_active": {"type": "boolean"}
},
"required": ["id", "email", "name", "created_at"],
"additionalProperties": False
}
def test_user_response_schema():
"""Test that user response matches schema."""
response = client.get("/api/users/1")
assert response.status_code == 200
data = response.json()
# Validate against schema
try:
validate(instance=data, schema=USER_SCHEMA)
except ValidationError as e:
pytest.fail(f"Response doesn't match schema: {e.message}")
def test_list_users_schema():
"""Test that list response matches schema."""
LIST_SCHEMA = {
"type": "object",
"properties": {
"items": {
"type": "array",
"items": USER_SCHEMA
},
"total": {"type": "integer", "minimum": 0},
"page": {"type": "integer", "minimum": 1},
"page_size": {"type": "integer", "minimum": 1}
},
"required": ["items", "total"]
}
response = client.get("/api/users")
assert response.status_code == 200
validate(instance=response.json(), schema=LIST_SCHEMA)
Pact Contract Testing:
from pact import Consumer, Provider
# Consumer side (API client)
pact = Consumer('UserServiceClient').has_pact_with(Provider('UserService'))
def test_get_user_contract():
"""Define contract for getting user."""
expected = {
'id': 1,
'email': 'test@example.com',
'name': 'Test User'
}
(pact
.given('user 1 exists')
.upon_receiving('a request for user 1')
.with_request('GET', '/api/users/1')
.will_respond_with(200, body=expected))
with pact:
client = UserServiceClient('http://localhost:1234')
user = client.get_user(1)
assert user['id'] == 1
assert user['email'] == 'test@example.com'
# Provider side (API implementation) verifies it meets the contract
4. Authentication and Authorization Testing
Auth Flow Testing:
class TestAuthentication:
def test_login_success(self):
"""Test successful login."""
response = client.post("/api/auth/login", json={
"email": "user@example.com",
"password": "correct_password"
})
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_credentials(self):
"""Test login with wrong password."""
response = client.post("/api/auth/login", json={
"email": "user@example.com",
"password": "wrong_password"
})
assert response.status_code == 401
assert "invalid credentials" in response.json()["detail"].lower()
def test_access_protected_endpoint_without_token(self):
"""Test accessing protected endpoint without auth."""
response = client.get("/api/protected/resource")
assert response.status_code == 401
assert "authorization" in response.json()["detail"].lower()
def test_access_protected_endpoint_with_token(self):
"""Test accessing protected endpoint with valid token."""
# Login to get token
login_response = client.post("/api/auth/login", json={
"email": "user@example.com",
"password": "password123"
})
token = login_response.json()["access_token"]
# Access protected endpoint
response = client.get(
"/api/protected/resource",
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 200
def test_expired_token(self):
"""Test that expired tokens are rejected."""
expired_token = "expired.jwt.token"
response = client.get(
"/api/protected/resource",
headers={"Authorization": f"Bearer {expired_token}"}
)
assert response.status_code == 401
assert "expired" in response.json()["detail"].lower()
def test_insufficient_permissions(self):
"""Test accessing admin endpoint as regular user."""
# Login as regular user
login_response = client.post("/api/auth/login", json={
"email": "regular_user@example.com",
"password": "password123"
})
token = login_response.json()["access_token"]
# Try to access admin endpoint
response = client.get(
"/api/admin/users",
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 403
assert "permission" in response.json()["detail"].lower()
5. Error Handling Tests
Comprehensive Error Coverage:
class TestErrorHandling:
def test_malformed_json(self):
"""Test handling of malformed JSON."""
response = client.post(
"/api/users",
data="not valid json",
headers={"Content-Type": "application/json"}
)
assert response.status_code == 400
assert "json" in response.json()["detail"].lower()
def test_missing_required_field(self):
"""Test validation error for missing field."""
response = client.post("/api/users", json={
"email": "test@example.com"
# Missing required 'name' field
})
assert response.status_code == 422
errors = response.json()["detail"]
assert any(
error["loc"][-1] == "name" and error["type"] == "missing"
for error in errors
)
def test_type_validation(self):
"""Test type validation errors."""
response = client.post("/api/users", json={
"email": "test@example.com",
"name": "Test",
"age": "not a number" # Should be integer
})
assert response.status_code == 422
def test_rate_limit_exceeded(self):
"""Test rate limiting."""
# Make many requests
for _ in range(100):
response = client.get("/api/users")
if response.status_code == 429:
break
else:
pytest.fail("Rate limit not enforced")
assert response.status_code == 429
assert "retry-after" in response.headers
def test_request_too_large(self):
"""Test request size limit."""
large_data = {"name": "x" * 1_000_000} # 1MB of data
response = client.post("/api/users", json=large_data)
assert response.status_code == 413 # Payload Too Large
def test_unsupported_media_type(self):
"""Test content-type validation."""
response = client.post(
"/api/users",
data="plain text data",
headers={"Content-Type": "text/plain"}
)
assert response.status_code == 415 # Unsupported Media Type
6. Pagination and Filtering Tests
Test Query Parameters:
class TestPaginationAndFiltering:
def test_default_pagination(self):
"""Test default pagination parameters."""
response = client.get("/api/users")
assert response.status_code == 200
data = response.json()
assert "items" in data
assert "total" in data
assert "page" in data
assert "page_size" in data
assert data["page"] == 1
assert data["page_size"] == 20 # Default
def test_custom_page_size(self):
"""Test custom page size."""
response = client.get("/api/users?page_size=5")
data = response.json()
assert len(data["items"]) <= 5
assert data["page_size"] == 5
def test_page_navigation(self):
"""Test navigating through pages."""
# Get first page
page1 = client.get("/api/users?page=1&page_size=10").json()
# Get second page
page2 = client.get("/api/users?page=2&page_size=10").json()
# Should be different results
assert page1["items"] != page2["items"]
def test_filtering(self):
"""Test filtering by fields."""
response = client.get("/api/users?status=active&role=admin")
data = response.json()
for user in data["items"]:
assert user["status"] == "active"
assert user["role"] == "admin"
def test_sorting(self):
"""Test sorting results."""
response = client.get("/api/users?sort=created_at&order=desc")
data = response.json()
dates = [item["created_at"] for item in data["items"]]
# Verify descending order
assert dates == sorted(dates, reverse=True)
def test_search(self):
"""Test search functionality."""
response = client.get("/api/users?search=john")
data = response.json()
for user in data["items"]:
# Name or email should contain search term
assert (
"john" in user["name"].lower() or
"john" in user["email"].lower()
)
7. Performance Testing
Load and Stress Tests:
import time
import concurrent.futures
import statistics
def test_response_time():
"""Test that API responds within acceptable time."""
start = time.time()
response = client.get("/api/users")
duration = time.time() - start
assert response.status_code == 200
assert duration < 1.0 # Should respond within 1 second
def test_concurrent_requests():
"""Test handling concurrent requests."""
def make_request():
return client.get("/api/users")
# Make 10 concurrent requests
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
futures = [executor.submit(make_request) for _ in range(10)]
results = [f.result() for f in concurrent.futures.as_completed(futures)]
# All should succeed
assert all(r.status_code == 200 for r in results)
def test_throughput():
"""Test API throughput."""
num_requests = 100
start = time.time()
for _ in range(num_requests):
response = client.get("/api/users")
assert response.status_code == 200
duration = time.time() - start
requests_per_second = num_requests / duration
print(f"Throughput: {requests_per_second:.2f} requests/second")
assert requests_per_second > 10 # At least 10 req/s
def test_response_time_under_load():
"""Test response times under load."""
response_times = []
def make_timed_request():
start = time.time()
response = client.get("/api/users")
duration = time.time() - start
response_times.append(duration)
return response
# Make 100 concurrent requests
with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
futures = [executor.submit(make_timed_request) for _ in range(100)]
results = [f.result() for f in concurrent.futures.as_completed(futures)]
# Calculate statistics
avg_time = statistics.mean(response_times)
p95_time = statistics.quantiles(response_times, n=20)[18] # 95th percentile
p99_time = statistics.quantiles(response_times, n=100)[98] # 99th percentile
print(f"Average response time: {avg_time:.3f}s")
print(f"P95 response time: {p95_time:.3f}s")
print(f"P99 response time: {p99_time:.3f}s")
assert avg_time < 0.5 # Average under 500ms
assert p95_time < 1.0 # 95% under 1 second
assert p99_time < 2.0 # 99% under 2 seconds
Using Locust for Load Testing:
from locust import HttpUser, task, between
class APIUser(HttpUser):
wait_time = between(1, 3) # Wait 1-3 seconds between requests
def on_start(self):
"""Login before starting tasks."""
response = self.client.post("/api/auth/login", json={
"email": "loadtest@example.com",
"password": "password123"
})
self.token = response.json()["access_token"]
@task(3) # Weight: 3x more likely than other tasks
def list_users(self):
"""Test listing users."""
self.client.get(
"/api/users",
headers={"Authorization": f"Bearer {self.token}"}
)
@task(2)
def get_user(self):
"""Test getting specific user."""
self.client.get(
f"/api/users/{random.randint(1, 100)}",
headers={"Authorization": f"Bearer {self.token}"}
)
@task(1)
def create_user(self):
"""Test creating user."""
self.client.post(
"/api/users",
headers={"Authorization": f"Bearer {self.token}"},
json={
"email": f"user{random.randint(1, 10000)}@example.com",
"name": "Load Test User",
"password": "password123"
}
)
# Run: locust -f locustfile.py --host=http://localhost:8000
Test Organization
Test Structure
# tests/
# ├── unit/
# │ ├── test_models.py
# │ ├── test_services.py
# │ └── test_utils.py
# ├── integration/
# │ ├── test_user_api.py
# │ ├── test_order_api.py
# │ └── test_auth_api.py
# ├── contract/
# │ ├── test_schemas.py
# │ └── pacts/
# ├── e2e/
# │ └── test_user_workflows.py
# ├── performance/
# │ ├── test_load.py
# │ └── locustfile.py
# └── conftest.py # Shared fixtures
# conftest.py - Shared test fixtures
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.database import Base, engine, SessionLocal
@pytest.fixture(scope="function")
def db_session():
"""Create a clean database session for each test."""
Base.metadata.create_all(bind=engine)
db = SessionLocal()
try:
yield db
finally:
db.close()
Base.metadata.drop_all(bind=engine)
@pytest.fixture(scope="function")
def client(db_session):
"""Create test client with clean database."""
def override_get_db():
yield db_session
app.dependency_overrides[get_db] = override_get_db
yield TestClient(app)
app.dependency_overrides.clear()
@pytest.fixture
def auth_headers(client):
"""Get authentication headers for tests."""
response = client.post("/api/auth/login", json={
"email": "test@example.com",
"password": "password123"
})
token = response.json()["access_token"]
return {"Authorization": f"Bearer {token}"}
Test Data Management
Factories and Fixtures
import factory
from factory.faker import Faker
from app.models import User, Order
class UserFactory(factory.Factory):
class Meta:
model = User
email = Faker('email')
name = Faker('name')
is_active = True
class OrderFactory(factory.Factory):
class Meta:
model = Order
user = factory.SubFactory(UserFactory)
total = Faker('pydecimal', left_digits=4, right_digits=2, positive=True)
status = 'pending'
# Usage in tests
def test_create_order():
user = UserFactory.create()
order = OrderFactory.create(user=user, total=99.99)
assert order.user.email == user.email
assert order.total == 99.99
Test Database Strategies
Strategy 1: Transaction Rollback
import pytest
from sqlalchemy.orm import Session
@pytest.fixture
def db_session():
"""Rollback all changes after test."""
connection = engine.connect()
transaction = connection.begin()
session = Session(bind=connection)
yield session
session.close()
transaction.rollback()
connection.close()
Strategy 2: Separate Test Database
# Test configuration
TEST_DATABASE_URL = "postgresql://user:pass@localhost/test_db"
@pytest.fixture(scope="session")
def test_engine():
"""Create test database engine."""
engine = create_engine(TEST_DATABASE_URL)
Base.metadata.create_all(engine)
yield engine
Base.metadata.drop_all(engine)
Strategy 3: In-Memory SQLite
@pytest.fixture
def db_session():
"""Use in-memory SQLite for fast tests."""
engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)
SessionLocal = sessionmaker(bind=engine)
session = SessionLocal()
yield session
session.close()
Best Practices
1. Test Independence
# BAD: Tests depend on each other
def test_create_user():
global user_id
response = client.post("/api/users", json={"email": "test@example.com"})
user_id = response.json()["id"]
def test_get_user():
# Depends on test_create_user running first!
response = client.get(f"/api/users/{user_id}")
# GOOD: Each test is independent
def test_create_user():
response = client.post("/api/users", json={"email": "test@example.com"})
assert response.status_code == 201
def test_get_user():
# Create user within this test
create_response = client.post("/api/users", json={"email": "test@example.com"})
user_id = create_response.json()["id"]
response = client.get(f"/api/users/{user_id}")
assert response.status_code == 200
2. Clear Test Names
# BAD: Unclear names
def test_user_1():
...
def test_user_2():
...
# GOOD: Descriptive names
def test_create_user_with_valid_data_returns_201():
...
def test_create_user_with_duplicate_email_returns_409():
...
def test_get_user_with_invalid_id_returns_404():
...
3. AAA Pattern (Arrange-Act-Assert)
def test_update_user_email():
# Arrange: Set up test data
user = UserFactory.create(email="old@example.com")
new_email = "new@example.com"
# Act: Perform the action
response = client.put(f"/api/users/{user.id}", json={
"email": new_email
})
# Assert: Verify the results
assert response.status_code == 200
assert response.json()["email"] == new_email
Related Skills
- REST API Design: Designing testable APIs
- HTTP Protocol: Understanding status codes and headers
- Test Automation: CI/CD integration
- Performance Testing: Load and stress testing
- Security Testing: Authentication and authorization testing
- Mock Services: Creating test doubles
- OpenAPI/Swagger: API documentation and contract testing
Converted and distributed by TomeVault — claim your Tome and manage your conversions.