Testing — End-to-End Quality Assurance with pytest
Testing is like a safety net—you can't see it while walking the tightrope (writing code), but if you slip up (encounter a bug), it's your only lifeline. A performance without a safety net is bound to end in an accident sooner or later.
1. What You'll Learn
TestClient(httpx) Basics: Synchronous/Asynchronous Test Client- Test database isolation: Use a separate database for each test, managed via fixtures
- Dependency Override:
app.dependency_overridesReplace DB/Authentication Dependencies - Testing Pyramid: A Layered Strategy of Unit → Integration → E2E Testing
- Alice Scenario: PriceTracker's Complete Test Suite
2. Alice's True Story
(1) Pain Point: Discovering Bugs Only After Launch
After Alice deployed PriceTracker, she discovered that the JWT token expiration time was set to 1 second, batch imports were skipping all validations, and rate limiting for Pro users was the same as for Free users. Every time she fixed a bug, she introduced a new one, and Bob complained, "The feature that was working last week isn't working again this week." Alice didn't have automated tests; she relied entirely on manually clicking through Swagger UI, and each regression test took two hours.
(2) Solutions for pytest Automated Testing
pytest + httpx TestClient ensures that every API endpoint has automated tests; dependency_overrides replaces the production database with a test database; and every time git push is run, all tests are automatically executed, detecting all regression issues within 2 minutes.
def test_create_product(client):
response = client.post("/api/v1/products", json={"name": "Widget", "price": 9.99})
assert response.status_code == 201
assert response.json()["name"] == "Widget"
(3) Revenue
Regression testing went from a 2-hour manual process to a 2-minute automated one; a JWT expiration bug was caught during testing in the development phase; and after deployment, the number of bugs dropped from 5 per week to 1 per month.
3. TestClient Basics
(1) The Testing Pyramid
graph TD
E2E[E2E Tests - Few] --> INT[Integration Tests - Medium]
INT --> UNIT[Unit Tests - Many]
UNIT --- U1[Pydantic Model Validation]
UNIT --- U2[Repository Functions]
UNIT --- U3[Service Logic]
INT --- I1[API Endpoint + DB]
INT --- I2[Auth Flow]
INT --- I3[CRUD Operations]
E2E --- E1[Full User Journey]
E2E --- E2[WebSocket + API]
| Level | Quantity | Speed | Dependencies |
|---|---|---|---|
| Unit Tests | Many (100+) | Fast (< 1 ms) | No External Dependencies |
| Integration Testing | Medium (30-50) | Medium (10-100 ms) | Test Database |
| E2E Testing | Few (5-10) | Slow (1-5 s) | Full Environment |
(1) ▶ Example: TestClient Basic Fixture
# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.core.deps import get_db
# Override database dependency
def get_test_db():
# Use in-memory SQLite for testing
engine = create_async_engine("sqlite+aiosqlite:///test.db")
# ... session setup
yield session
# ... cleanup
@pytest.fixture
def client():
app.dependency_overrides[get_db] = get_test_db
with TestClient(app) as c:
yield c
app.dependency_overrides.clear()
Output:
# Function defined successfully
(2) ▶ Example: Basic Endpoint Testing
# tests/test_health.py
def test_health_check(client):
response = client.get("/health")
assert response.status_code == 200
data = response.json()
assert data["status"] == "healthy"
assert "service" in data
def test_openapi_docs_available(client):
response = client.get("/docs")
assert response.status_code == 200
Output:
# Function defined successfully
4. Testing Database Isolation
(1) A separate database for each test
(1) ▶ Example: Test database fixture
# tests/conftest.py
import pytest
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from app.models import Base
TEST_DATABASE_URL = "sqlite+aiosqlite:///test_pricetracker.db"
@pytest.fixture(scope="function")
async def test_db():
# Create fresh test database for each test
engine = create_async_engine(TEST_DATABASE_URL, echo=False)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
TestSession = async_sessionmaker(engine, expire_on_commit=False)
async with TestSession() as session:
yield session
# Drop all tables after test
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
await engine.dispose()
Output:
# Function defined successfully
(2) ▶ Example: Asynchronous Test Client
# tests/conftest.py
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app
@pytest.fixture
async def async_client():
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as ac:
yield ac
Output:
# Function defined successfully
(2) Comparison of Quarantine Strategies
| Strategy | Speed | Isolation | Applicability |
|---|---|---|---|
| Create a new table for each test | Slow | Fully isolated | Data integrity test |
| Transaction Rollback | Fast | Good | Most integration tests |
| In-Memory SQLite | Fast | Good | Tests that do not rely on PostgreSQL features |
5. Techniques for Overriding Dependencies
(1) Replace the authentication and database
(1) ▶ Example: Overriding Authentication Dependencies
# tests/conftest.py
from app.core.deps import get_current_user, get_db
def get_test_user():
"""Mock authenticated user for testing"""
return {"id": 1, "email": "alice@test.com", "role": "admin", "subscription": "pro"}
@pytest.fixture
def auth_client(client):
# Override authentication - no real JWT needed
app.dependency_overrides[get_current_user] = get_test_user
yield client
app.dependency_overrides.clear()
Output:
# Function defined successfully
(2) ▶ Example: Testing Protected Endpoints
# tests/test_products.py
def test_list_products_unauthorized(client):
"""Without auth token, should return 401"""
response = client.get("/api/v1/products")
assert response.status_code == 401
def test_list_products_authorized(auth_client):
"""With mocked auth, should return 200"""
response = auth_client.get("/api/v1/products")
assert response.status_code == 200
def test_create_product(auth_client):
response = auth_client.post(
"/api/v1/products",
json={"name": "Widget", "category": "electronics", "base_price": 29.99},
)
assert response.status_code == 201
assert response.json()["name"] == "Widget"
Output:
# Function defined successfully
(3) ▶ Example: Overriding Subscription-Level Test Permissions
def get_free_user():
return {"id": 2, "email": "free@test.com", "role": "user", "subscription": "free"}
def get_pro_user():
return {"id": 3, "email": "pro@test.com", "role": "user", "subscription": "pro"}
def test_bulk_import_free_user_limited(client):
"""Free users can import max 1000 prices"""
app.dependency_overrides[get_current_user] = get_free_user
prices = [{"product_id": i, "price": 9.99} for i in range(1500)]
response = client.post("/api/v1/prices/bulk", json=prices)
assert response.status_code == 403
assert "limit" in response.json()["detail"].lower()
app.dependency_overrides.clear()
def test_bulk_import_pro_user(client):
"""Pro users can import up to 100000 prices"""
app.dependency_overrides[get_current_user] = get_pro_user
prices = [{"product_id": i, "price": 9.99} for i in range(5000)]
response = client.post("/api/v1/prices/bulk", json=prices)
assert response.status_code == 201
app.dependency_overrides.clear()
Output:
# Function defined successfully
6. Example of a Complete Test Suite
(1) ▶ Example: Unit Testing Pydantic Models
# tests/test_models.py
import pytest
from pydantic import ValidationError
from app.schemas import ProductCreate, PriceCreate
def test_product_create_valid():
p = ProductCreate(name="Widget", category="electronics", base_price=29.99)
assert p.name == "Widget"
assert p.base_price == 29.99
def test_product_create_negative_price():
with pytest.raises(ValidationError) as exc:
ProductCreate(name="Widget", category="electronics", base_price=-1)
assert "greater than 0" in str(exc.value)
def test_product_create_empty_name():
with pytest.raises(ValidationError):
ProductCreate(name="", category="electronics", base_price=9.99)
def test_price_create_rounds_precision():
p = PriceCreate(product_id=1, price=9.999, currency="USD", source="test")
assert p.price == 10.0 # Rounded to 2 decimal places
def test_price_create_invalid_currency():
with pytest.raises(ValidationError):
PriceCreate(product_id=1, price=9.99, currency="XYZ", source="test")
Output:
# Function defined successfully
(2) ▶ Example: Integration Testing for CRUD Operations
# tests/test_crud.py
import pytest
from fastapi.testclient import TestClient
def test_product_crud_lifecycle(auth_client):
# Create
create_resp = auth_client.post(
"/api/v1/products",
json={"name": "Test Widget", "category": "electronics", "base_price": 19.99},
)
assert create_resp.status_code == 201
product_id = create_resp.json()["id"]
# Read
get_resp = auth_client.get(f"/api/v1/products/{product_id}")
assert get_resp.status_code == 200
assert get_resp.json()["name"] == "Test Widget"
# Update
update_resp = auth_client.put(
f"/api/v1/products/{product_id}",
json={"name": "Updated Widget", "base_price": 24.99},
)
assert update_resp.status_code == 200
assert update_resp.json()["base_price"] == 24.99
# Delete
delete_resp = auth_client.delete(f"/api/v1/products/{product_id}")
assert delete_resp.status_code == 200
# Verify deleted
get_resp2 = auth_client.get(f"/api/v1/products/{product_id}")
assert get_resp2.status_code == 404
Output:
# Function defined successfully
❓ FAQ
function (recreated for each test) is the safest; session (shared across the entire test session) is the fastest but offers poor isolation. Use function for the database and function for TestClient.AsyncClient.websocket_connect() to establish a connection, and send and receive messages to verify its behavior.pytest-xdist for parallel execution (pytest -n auto), use transaction rollback instead of creating tables, and use unit tests instead of integration tests.dependency_overrides affect other tests?yield statement in the fixture app.dependency_overrides.clear().📖 Summary
- Testing Pyramid: Unit Testing (Frequent and Fast) → Integration Testing (Moderate) → E2E (Less Frequent and Slower)
- TestClient is used to test simple synchronous endpoints, while AsyncClient is used to test asynchronous logic (such as WebSockets)
- Each test uses a separate database; fixtures manage creation and cleanup
app.dependency_overridesReplaces authentication and database dependencies; no real JWT or PG required- Comprehensive test coverage: Pydantic validation (unit), CRUD workflows (integration), access control (integration)
📝 Exercises
- Basic Exercise (Difficulty: ⭐): Use TestClient to write a test for the
/healthendpoint to verify that it returns a 200 status code and the correct JSON structure. Hint:TestClient(app)+client.get("/health") - Advanced Problem (Difficulty ⭐⭐): Write CRUD lifecycle tests—create, query, update, and delete products—and verify the status codes and return data for each step. Use
dependency_overridesto skip authentication. Hint:app.dependency_overrides[get_current_user] = mock_fn - Challenge (Difficulty: ⭐⭐⭐): Implement a complete test suite: Pydantic validation tests (invalid data raises a
ValidationError), authentication tests (401 for no token, 403 for Free users, 200 for Pro users), and pagination tests (correct return ofskipandlimitvalues). Hint:pytest.raises(ValidationError)+ multipledependency_overridesfixtures
---|



