fastapi-tdd

name	fastapi-tdd
description	FastAPI testing with TDD — pytest-asyncio, httpx AsyncClient, async SQLAlchemy fixtures, polyfactory, mocking external services, and coverage targets.
origin	local

FastAPI Testing with TDD

Test-driven development for FastAPI applications using pytest-asyncio, httpx, and async SQLAlchemy.

When to Activate

• Writing new FastAPI endpoints or services
• Setting up testing infrastructure for a FastAPI project
• Testing async routes, dependencies, and database interactions
• Implementing TDD on FastAPI + Alembic + SQLAlchemy projects

TDD Workflow

Red-Green-Refactor for FastAPI

# Step 1: RED — Write failing test first
async def test_create_product_returns_201(client, auth_headers):
    response = await client.post(
        "/api/v1/products/",
        json={"name": "Widget", "price": "9.99", "stock": 100},
        headers=auth_headers,
    )
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "Widget"
    assert data["id"] is not None

# Step 2: GREEN — Implement the route + service to make it pass
# Step 3: REFACTOR — Improve while keeping tests green

Setup

Dependencies

# pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
python_files = ["test_*.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
addopts = [
    "--cov=app",
    "--cov-report=term-missing",
    "--cov-report=html",
    "--cov-fail-under=80",
    "--strict-markers",
]
markers = [
    "slow: marks tests as slow (deselect with -m 'not slow')",
    "integration: marks integration tests",
]

[tool.coverage.run]
omit = ["tests/*", "alembic/*", "app/main.py"]

# Install test dependencies
uv add --dev pytest pytest-asyncio httpx pytest-cov anyio polyfactory

Async Test Database

# tests/conftest.py
import pytest
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from app.main import app
from app.database import Base, get_db

TEST_DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/test_db"

test_engine = create_async_engine(TEST_DATABASE_URL, echo=False)
TestAsyncSessionLocal = async_sessionmaker(
    bind=test_engine, expire_on_commit=False, autoflush=False
)


@pytest.fixture(scope="session", autouse=True)
async def setup_database():
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await test_engine.dispose()


@pytest.fixture
async def db() -> AsyncSession:
    async with TestAsyncSessionLocal() as session:
        yield session
        await session.rollback()  # Isolate each test


@pytest.fixture
def override_db(db: AsyncSession):
    
      ():
         db

    app.dependency_overrides[get_db] = _override
    
    app.dependency_overrides.clear()



  () -> AsyncClient:
      AsyncClient(
        transport=ASGITransport(app=app), base_url=
    )  c:
         c

User + Auth Fixtures

# tests/conftest.py (continued)
import pytest
from app.auth.passwords import hash_password
from app.auth.tokens import create_access_token
from app.models.user import User


@pytest.fixture
async def user(db: AsyncSession) -> User:
    u = User(
        email="[e-mail]",
        hashed_password=hash_password("TestPass123"),
        is_active=True,
    )
    db.add(u)
    await db.flush()
    await db.refresh(u)
    return u


@pytest.fixture
async def admin_user(db: AsyncSession) -> User:
    u = User(
        email="[e-mail]",
        hashed_password=hash_password("AdminPass123"),
        is_active=True,
        is_superuser=True,
        role="admin",
    )
    db.add(u)
    await db.flush()
    await db.refresh(u)
    return u


@pytest.fixture
def auth_headers(user: User) -> dict[str, str]:
    token = create_access_token(subject=user.id)
    return {"Authorization": f"Bearer {token}"}


@pytest.fixture
def () -> [, ]:
    token = create_access_token(subject=admin_user.)
     {: }

Polyfactory for Test Data

# tests/factories.py
from decimal import Decimal
from polyfactory.factories.sqlalchemy_factory import SQLAlchemyFactory
from app.models.user import User
from app.models.product import Product


class UserFactory(SQLAlchemyFactory):
    __model__ = User
    __set_relationships__ = False

    email = SQLAlchemyFactory.faker.email
    hashed_password = "$2b$12$fakehash"  # pre-hashed placeholder
    is_active = True
    is_superuser = False


class ProductFactory(SQLAlchemyFactory):
    __model__ = Product
    __set_relationships__ = False

    name = SQLAlchemyFactory.faker.sentence(nb_words=3)
    description = SQLAlchemyFactory.faker.text(max_nb_chars=200)
    price = Decimal("49.99")
    stock = 50
    is_active = True


# Usage in tests
async def test_something(db):
    product = ProductFactory.build()          # in-memory, no DB
    product = await ProductFactory.create_async(db)  # persisted
    products = await ProductFactory.create_batch_async(5, db)

Model / Service Testing

# tests/test_services/test_product_service.py
import pytest
from decimal import Decimal
from fastapi import HTTPException

from app.services.product_service import ProductService
from app.schemas.product import ProductCreate, ProductUpdate
from tests.factories import ProductFactory, UserFactory


class TestProductService:

    async def test_create_product(self, db, user):
        service = ProductService(db)
        data = ProductCreate(name="Widget", price=Decimal("9.99"), stock=10)

        product = await service.create(data, owner_id=user.id)

        assert product.id is not None
        assert product.name == "Widget"
        assert product.owner_id == user.id

    async def test_get_or_404_raises_for_missing(self, db):
        service = ProductService(db)

        with pytest.raises(HTTPException) as exc_info:
            await service.get_or_404(99999)

        assert exc_info.value.status_code == 404

    async def test_update_partial(self, db, user):
        product =  ProductFactory.create_async(db, owner_id=user.)
        service = ProductService(db)
        data = ProductUpdate(name=)

        updated =  service.update(product, data)

         updated.name == 
         updated.price == product.price  

      ():
         ProductFactory.create_batch_async(, db, is_active=)
         ProductFactory.create_batch_async(, db, is_active=)
        service = ProductService(db)

        result =  service.()

         result.total == 

      ():
         ProductFactory.create_async(db, name=, is_active=)
         ProductFactory.create_async(db, name=, is_active=)
        service = ProductService(db)

        result =  service.(search=)

         result.total == 
         result.items[].name == 

      ():
         ProductFactory.create_batch_async(, db, is_active=)
        service = ProductService(db)

        page1 =  service.(page=, size=)
        page2 =  service.(page=, size=)

         (page1.items) == 
         (page2.items) == 
         page1.pages == 

API Endpoint Testing

# tests/test_routers/test_products.py
import pytest
from httpx import AsyncClient


class TestProductEndpoints:

    async def test_list_products_unauthenticated(self, client: AsyncClient):
        response = await client.get("/api/v1/products/")
        assert response.status_code == 200  # public endpoint

    async def test_create_product_requires_auth(self, client: AsyncClient):
        response = await client.post(
            "/api/v1/products/",
            json={"name": "Widget", "price": "9.99", "stock": 10},
        )
        assert response.status_code == 401

    async def test_create_product(self, client: AsyncClient, auth_headers):
        response = await client.post(
            "/api/v1/products/",
            json={"name": "Widget", "price": "9.99", "stock": 10},
            headers=auth_headers,
        )
        assert response.status_code == 201
        data = response.json()
        assert data["name"] == "Widget"
           data

      ():
        response =  client.post(
            ,
            json={: , : , : },
            headers=auth_headers,
        )
         response.status_code == 

      ():
        response =  client.get()
         response.status_code == 

      ():
        other_user =  UserFactory.create_async(db)
        product =  ProductFactory.create_async(db, owner_id=other_user.)

        response =  client.patch(
            ,
            json={: },
            headers=auth_headers,
        )
         response.status_code == 

      ():
        product =  ProductFactory.create_async(db, owner_id=user.)

        response =  client.delete(
            , headers=auth_headers
        )
         response.status_code == 

      ():
        response =  client.get()
        data = response.json()
           data
           data
           data

Auth Endpoint Testing

# tests/test_routers/test_auth.py
class TestAuth:

    async def test_login_success(self, client: AsyncClient, user):
        response = await client.post(
            "/api/v1/auth/token",
            data={"username": user.email, "password": "TestPass123"},
        )
        assert response.status_code == 200
        data = response.json()
        assert "access_token" in data
        assert data["token_type"] == "bearer"

    async def test_login_wrong_password(self, client: AsyncClient, user):
        response = await client.post(
            "/api/v1/auth/token",
            data={"username": user.email, "password": "WrongPass"},
        )
        assert response.status_code == 401

    async def test_login_unknown_user(self, client: AsyncClient):
        response = await client.post(
            "/api/v1/auth/token",
            data={"username": "[e-mail]", "password": "Pass123"},
        )
        assert response.status_code == 401

    async def test_protected_endpoint_with_invalid_token():
        response =  client.get(
            ,
            headers={: },
        )
         response.status_code == 

Mocking External Services

# tests/test_services/test_email.py
from unittest.mock import AsyncMock, patch
import pytest


async def test_order_confirmation_email_sent(db, user):
    with patch("app.services.email_service.send_email", new_callable=AsyncMock) as mock_send:
        from app.services.order_service import OrderService
        service = OrderService(db)
        order = await service.create(...)

        mock_send.assert_called_once_with(
            to=user.email,
            subject="Order Confirmation",
        )


# Override entire dependency for a test
async def test_with_mock_payment_gateway(client, auth_headers):
    with patch("app.routers.orders.PaymentGateway.charge") as mock_charge:
        mock_charge.return_value = {"status": "succeeded", "id": "ch_123"}

        response = await client.post("/api/v1/orders/", ..., headers=auth_headers)

        assert response.status_code == 201
        mock_charge.assert_called_once()

Testing Best Practices

DO

• asyncio_mode = "auto" — no need to mark every test @pytest.mark.asyncio
• Roll back after each test — use await session.rollback() in the db fixture
• Override get_db via dependency_overrides — inject test session into app
• Test the HTTP contract — status codes, response shape, error messages
• Test authorization explicitly — unauthenticated, wrong user, missing role
• Use polyfactory for realistic but isolated test data
• Mock at the boundary — mock external HTTP calls, email clients, S3, etc.

DON'T

• Don't use the production DB — always a separate test DB
• Don't share state between tests — each test rolls back
• Don't test Pydantic / SQLAlchemy internals — trust the libraries
• Don't over-mock — real DB queries catch real bugs
• Don't hardcode IDs — use fixture-created objects

Coverage Targets

Component	Target
Routers (endpoints)	85%+
Services (business logic)	90%+
Auth / security	90%+
Schemas (validators)	80%+
Overall	80%+

# Run all tests with coverage
pytest

# Run only unit tests (skip slow/integration)
pytest -m "not slow and not integration"

# Run a specific file
pytest tests/test_routers/test_products.py -v

# Show which lines are missing coverage
pytest --cov=app --cov-report=term-missing

Quick Reference

Pattern	Usage
asyncio_mode = "auto"	Auto-async tests in pytest.ini
ASGITransport(app=app)	Mount FastAPI into httpx
app.dependency_overrides	Swap Depends for tests
await session.rollback()	Isolate DB state per test
SQLAlchemyFactory	Generate ORM model instances
patch("module.func", AsyncMock)	Mock async external calls
auth_headers fixture	Reuse JWT headers across tests
pytest.raises(HTTPException)	Assert service-layer errors

Remember: Tests are the first consumer of your API. If they're painful to write, the design needs work.