Testing in FastAPI: PyTest, Fixtures e Async

pytest + FastAPI + async SQLAlchemy: setup corretto, isolamento del database, dependency injection nei test

---

⏱️ 14 minuti di lettura | 💻 Codice completo su GitHub → | Livello: Intermedio

---

TL;DR

Il problema: FastAPI + async + SQLAlchemy async ha quirk specifici che rendono il setup dei test non ovvio.

La soluzione:

• pytest-asyncio con asyncio_mode = auto — zero boilerplate
• Database di test in-memory (SQLite) o PostgreSQL dedicato per isolamento
• app.dependency_overrides per sostituire le dipendenze FastAPI
• httpx.AsyncClient + ASGITransport per test di integrazione HTTP
• Scope delle fixture: session per il database, function per la sessione (con rollback)

Quando usarlo:

• ✅ API FastAPI con SQLAlchemy async
• ✅ Endpoint con business logic non banale
• ✅ Prima di ogni deploy

Quando NON basta questo approccio:

• ❌ Microservizi con molte dipendenze esterne (serve contract testing)
• ❌ Performance testing (usa locust o k6)

Setup base | Database di test | Dependency injection | Errori comuni

---

Il Problema: Test Verdi, Produzione Rotta

I test erano tutti verdi. Deploy in produzione: endpoint critico rotto su un edge case che non avevamo coperto.

Il problema non era la mancanza di test. Era la struttura: mockavamo troppo. La sessione del database era condivisa tra test. Le dipendenze FastAPI non venivano sostituite correttamente. I test passavano ma non stavano testando il comportamento reale.

Con FastAPI async la combinazione è non banale:

• pytest-asyncio ha quirk sulle fixture async che causano errori criptici
• SQLAlchemy async richiede gestione esplicita delle transazioni nei test
• Il sistema di Dependency Injection di FastAPI va sostituito per i test, non mockato a caso
• TestClient di Starlette è sincrono — con async SQLAlchemy causa deadlock

Questo articolo copre il setup corretto, dalla prima fixture fino ai test di integrazione HTTP.

---

Architettura di Riferimento

Gli esempi usano la stessa struttura dell’articolo sul Repository Pattern: router → service → repository → database.

app/
├── main.py
├── database.py          # Engine, Base, get_db
├── models.py            # SQLAlchemy models
├── repositories/
│   └── order_repository.py
├── services/
│   └── order_service.py
└── routers/
    └── orders.py

tests/
├── conftest.py          # Fixture condivise
├── test_orders_router.py   # Test integrazione HTTP
├── test_order_service.py   # Test unit service
└── test_order_repository.py

---

Setup Base

Dipendenze

pip install pytest pytest-asyncio httpx aiosqlite

# requirements-dev.txt
pytest==8.0.0
pytest-asyncio==0.23.0
httpx==0.27.0
aiosqlite==0.20.0        # SQLite async driver per i test

pytest.ini

[pytest]
asyncio_mode = auto
asyncio_default_fixture_loop_scope = session

asyncio_mode = auto elimina il bisogno di decorare ogni test con @pytest.mark.asyncio. Tutti i test async vengono gestiti automaticamente.

asyncio_default_fixture_loop_scope = session è necessario se usi fixture async con scope="session" (come l’engine). Senza questa riga, pytest-asyncio >= 0.23 emette un DeprecationWarning criptico e le fixture session-scoped potrebbero girare su loop diversi da quello dei test.

Struttura del database (app/database.py)

# app/database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/mydb"

engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

---

Database di Test

Opzione 1: SQLite In-Memory

Veloce, zero setup, nessuna dipendenza esterna. Limite: non supporta tutte le feature PostgreSQL (JSONB, array types, ecc.).

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

from app.database import Base

TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"

@pytest_asyncio.fixture(scope="session")
async def engine():
    """Create test engine once per test session."""
    engine = create_async_engine(TEST_DATABASE_URL, echo=False)

    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_asyncio.fixture(scope="function")
async def db_session(engine):
    """Provide a fresh database session for each test, with rollback."""
    async_session = async_sessionmaker(engine, expire_on_commit=False)

    async with async_session() as session:
        yield session
        await session.rollback()  # Clean up after each test

Scope session per l’engine (si crea una volta sola), scope function per la sessione con rollback automatico. Ogni test parte da un database pulito.

Opzione 2: PostgreSQL di Test

Necessario quando i test usano JSONB, array types, o funzioni specifiche di PostgreSQL.

# tests/conftest.py
import os

TEST_DATABASE_URL = os.getenv(
    "TEST_DATABASE_URL",
    "postgresql+asyncpg://postgres:postgres@localhost/test_db"
)

@pytest_asyncio.fixture(scope="session")
async def engine():
    engine = create_async_engine(TEST_DATABASE_URL)

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)   # Reset state
        await conn.run_sync(Base.metadata.create_all)

    yield engine
    await engine.dispose()

Con PostgreSQL serve il drop_all all’inizio per garantire uno stato pulito anche se il test precedente è crashato senza cleanup.

---

Dependency Injection nei Test

FastAPI usa un sistema di DI basato su Depends(). Nei test dobbiamo sostituire get_db con la nostra sessione di test.

app.dependency_overrides

# tests/conftest.py
from httpx import AsyncClient, ASGITransport
from app.main import app
from app.database import get_db

@pytest_asyncio.fixture(scope="function")
async def client(db_session):
    """HTTP client that uses the test database 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 ac:
        yield ac

    app.dependency_overrides.clear()

app.dependency_overrides è un dict {dependency: override_function}. FastAPI lo controlla prima di risolvere ogni dipendenza. Il .clear() al termine è importante: senza di esso gli override rimangono attivi per i test successivi.

Perché httpx.AsyncClient e non TestClient

starlette.testclient.TestClient è sincrono. Internamente gira l’app in un thread separato. Con SQLAlchemy async questo causa deadlock perché la sessione async viene usata da due loop diversi.

httpx.AsyncClient con ASGITransport esegue tutto nello stesso event loop: nessun deadlock.

# ❌ Causa deadlock con async SQLAlchemy
from starlette.testclient import TestClient
client = TestClient(app)

# ✅ Corretto
from httpx import AsyncClient, ASGITransport
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
    response = await client.get("/orders")

---

Test di Integrazione HTTP

Con il client configurato, i test HTTP sono diretti:

# tests/test_orders_router.py
import pytest
from uuid import uuid4

async def test_create_order(client):
    response = await client.post("/orders", json={
        "product_id": str(uuid4()),
        "quantity": 2
    })
    assert response.status_code == 201
    data = response.json()
    assert data["quantity"] == 2
    assert "id" in data

async def test_get_order_not_found(client):
    response = await client.get(f"/orders/{uuid4()}")
    assert response.status_code == 404

async def test_create_order_invalid_quantity(client):
    response = await client.post("/orders", json={
        "product_id": str(uuid4()),
        "quantity": 0  # Pydantic validation: quantity >= 1
    })
    assert response.status_code == 422

async def test_order_roundtrip(client):
    """Create an order, then fetch it and verify the data."""
    product_id = str(uuid4())

    create_response = await client.post("/orders", json={
        "product_id": product_id,
        "quantity": 3
    })
    assert create_response.status_code == 201
    order_id = create_response.json()["id"]

    get_response = await client.get(f"/orders/{order_id}")
    assert get_response.status_code == 200
    assert get_response.json()["product_id"] == product_id
    assert get_response.json()["quantity"] == 3

Questi test passano per l’intero stack: router → service → repository → database di test. Se cambia qualcosa in uno dei layer, i test lo rilevano.

---

Test del Service Layer con Mock

Per testare la business logic in isolamento, si mockano i repository (non il database). Vedi l’articolo sul Repository Pattern per il setup dei repository.

# tests/test_order_service.py
import pytest
from unittest.mock import AsyncMock
from uuid import uuid4

from app.services.order_service import OrderService
from app.models import Order

@pytest.fixture
def mock_order_repo():
    return AsyncMock()

@pytest.fixture
def order_service(mock_order_repo):
    return OrderService(order_repo=mock_order_repo)

async def test_create_order_success(order_service, mock_order_repo):
    """Business logic: order is created and repo.create is called."""
    product_id = uuid4()
    mock_order_repo.create.return_value = Order(
        id=uuid4(),
        product_id=product_id,
        quantity=2,
        status="pending"
    )

    result = await order_service.create_order(
        product_id=product_id,
        quantity=2
    )

    assert result.status == "pending"
    mock_order_repo.create.assert_called_once()

async def test_create_order_zero_quantity(order_service):
    """Business logic: zero quantity raises ValueError."""
    with pytest.raises(ValueError, match="quantity must be positive"):
        await order_service.create_order(product_id=uuid4(), quantity=0)

async def test_cancel_order_already_shipped(order_service, mock_order_repo):
    """Business logic: cannot cancel a shipped order."""
    order_id = uuid4()
    mock_order_repo.get.return_value = Order(
        id=order_id,
        status="shipped"
    )

    with pytest.raises(ValueError, match="cannot cancel"):
        await order_service.cancel_order(order_id=order_id)

Questi test sono veloci (nessun database), coprono casi limite, e falliscono precisamente quando la business logic cambia. Sono complementari ai test di integrazione, non sostitutivi.

AsyncMock: errori comuni

# ❌ MagicMock non funziona con await
mock_repo = MagicMock()
result = await mock_repo.get(uuid4())  # TypeError: object MagicMock can't be used in await

# ✅ AsyncMock gestisce await
mock_repo = AsyncMock()
result = await mock_repo.get(uuid4())  # OK, ritorna un AsyncMock per default

# Specificare il return value
mock_repo.get.return_value = Order(id=uuid4(), status="pending")
result = await mock_repo.get(uuid4())
assert result.status == "pending"

# Simulare un'eccezione
mock_repo.get.side_effect = Exception("DB connection lost")
with pytest.raises(Exception):
    await mock_repo.get(uuid4())

---

Fixture Parametrizzate

Per testare più scenari sullo stesso endpoint senza duplicare il codice:

# tests/test_orders_router.py
import pytest

@pytest.mark.parametrize("quantity, expected_status", [
    (1,    201),   # Minimum valid quantity
    (100,  201),   # Large but valid quantity
    (0,    422),   # Zero: validation error
    (-1,   422),   # Negative: validation error
    (None, 422),   # Missing field: validation error
])
async def test_create_order_quantity_validation(client, quantity, expected_status):
    payload = {"product_id": str(uuid4())}
    if quantity is not None:
        payload["quantity"] = quantity

    response = await client.post("/orders", json=payload)
    assert response.status_code == expected_status

Per test sul service layer con mock, pytest.fixture con params è ancora più pulito perché non richiede database:

# tests/test_order_service.py
@pytest.fixture(params=["pending", "confirmed", "processing"])
def cancellable_status(request):
    return request.param

@pytest.fixture(params=["shipped", "delivered", "cancelled"])
def non_cancellable_status(request):
    return request.param

async def test_cancel_valid_statuses(order_service, mock_order_repo, cancellable_status):
    """Runs 3 times, once per cancellable status."""
    order_id = uuid4()
    mock_order_repo.get.return_value = Order(id=order_id, status=cancellable_status)
    mock_order_repo.update_status.return_value = Order(id=order_id, status="cancelled")

    result = await order_service.cancel_order(order_id=order_id)

    assert result.status == "cancelled"
    mock_order_repo.update_status.assert_called_once_with(order_id, "cancelled")

async def test_cancel_invalid_statuses(order_service, mock_order_repo, non_cancellable_status):
    """Runs 3 times, once per non-cancellable status."""
    mock_order_repo.get.return_value = Order(id=uuid4(), status=non_cancellable_status)

    with pytest.raises(ValueError, match="cannot cancel"):
        await order_service.cancel_order(order_id=uuid4())

Sei test scritti con due fixture e due funzioni. Se si aggiunge uno stato, basta aggiungere un elemento a params.

---

Errori Comuni

1. Scope sbagliato sulle fixture del database

# ❌ Scope "session" sulla sessione DB: i dati persistono tra i test
@pytest_asyncio.fixture(scope="session")
async def db_session(engine):
    async with AsyncSession(engine) as session:
        yield session
    # Nessun rollback: test A crea un record, test B lo trova già lì

# ✅ Scope "function" con rollback
@pytest_asyncio.fixture(scope="function")
async def db_session(engine):
    async with AsyncSession(engine) as session:
        yield session
        await session.rollback()

L’engine può avere scope session (si crea una volta). La sessione deve avere scope function con rollback per isolare ogni test.

2. Dimenticare await nei test async

# ❌ Il test passa sempre, non esegue niente
async def test_get_order():
    result = service.get_order(uuid4())  # Missing await!
    assert result is not None            # result è una coroutine, non None → True

# ✅ Corretto
async def test_get_order():
    result = await service.get_order(uuid4())
    assert result is not None

Con asyncio_mode = auto, pytest non avvisa se dimentichi await su una coroutine. Il test passa silenziosamente. Se hai dubbi, aggiungi import asyncio; assert not asyncio.iscoroutine(result) temporaneamente.

3. Mockare troppo

# ❌ Mock di tutto: il test non testa niente di reale
async def test_create_order(mock_service, mock_validator, mock_repo):
    mock_validator.validate.return_value = True
    mock_service.create.return_value = {"id": "123", "status": "ok"}
    response = await client.post("/orders", json={...})
    assert response.status_code == 201
    # Hai testato solo che il router chiama mock_service.create

# ✅ Mock solo delle dipendenze esterne (email, payment gateway, ecc.)
# Il database di test è reale, la business logic è reale
async def test_create_order(client):
    response = await client.post("/orders", json={
        "product_id": str(uuid4()), "quantity": 2
    })
    assert response.status_code == 201
    assert response.json()["status"] == "pending"

La regola: mocka le dipendenze che escono dal sistema (email, pagamenti, API esterne). Non mockare il database — usa un database di test reale.

4. Non pulire dependency_overrides

# ❌ Gli override rimangono attivi per i test successivi
@pytest_asyncio.fixture
async def client(db_session):
    app.dependency_overrides[get_db] = lambda: db_session
    async with AsyncClient(...) as ac:
        yield ac
    # Manca il cleanup!

# ✅ Cleanup garantito
@pytest_asyncio.fixture
async def client(db_session):
    async def override_get_db():
        yield db_session

    app.dependency_overrides[get_db] = override_get_db

    async with AsyncClient(...) as ac:
        yield ac

    app.dependency_overrides.clear()  # Always runs, even if test fails

5. @pytest.fixture invece di @pytest_asyncio.fixture su fixture async

# ❌ pytest.fixture non gestisce coroutine — la fixture restituisce una coroutine
# invece di eseguirla, e il test riceve un oggetto coroutine invece della sessione
@pytest.fixture(scope="function")
async def db_session(engine):
    async with AsyncSession(engine) as session:
        yield session
        await session.rollback()
# Errore a runtime: il test riceve un oggetto generator, non una AsyncSession

# ✅ pytest_asyncio.fixture per fixture async
@pytest_asyncio.fixture(scope="function")
async def db_session(engine):
    async with AsyncSession(engine) as session:
        yield session
        await session.rollback()

L’errore è sottile perché il test non fallisce con un messaggio chiaro — spesso si vede un AttributeError sul tipo sbagliato, non un errore sulla fixture in sé. Regola: qualsiasi fixture che usa async def o yield su operazioni async va decorata con @pytest_asyncio.fixture.

---

Esempio Completo: conftest.py

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

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

TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"

@pytest_asyncio.fixture(scope="session")
async def engine():
    """Single engine for the entire test session."""
    engine = create_async_engine(TEST_DATABASE_URL, echo=False)

    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_asyncio.fixture(scope="function")
async def db_session(engine):
    """Fresh session with rollback for each test."""
    async_session = async_sessionmaker(engine, expire_on_commit=False)

    async with async_session() as session:
        yield session
        await session.rollback()

@pytest_asyncio.fixture(scope="function")
async def client(db_session):
    """HTTP client wired to the test database."""
    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 ac:
        yield ac

    app.dependency_overrides.clear()

---

Conclusione

Next Steps

Prova subito:

git clone https://github.com/layerbylayer-blog/fastapi-testing-guide
cd fastapi-testing-guide
pip install -r requirements-dev.txt
pytest -v

Nel tuo progetto:

1. Aggiungi asyncio_mode = auto al pytest.ini
2. Crea conftest.py con le fixture engine, db_session, client
3. Scrivi prima i test degli endpoint (integrazione), poi i test del service (unit) per la business logic complessa
4. Aggiungi pytest alla CI — se non gira automaticamente, non viene aggiornato

Articolo correlato: Repository Pattern in FastAPI — il pattern architetturale che rende i test del service layer puliti e veloci.

---

Risorse

GitHub: fastapi-testing-guide

Include:

• ✅ conftest.py completo (SQLite + PostgreSQL)
• ✅ Esempi di test integrazione, unit, parametrizzati
• ✅ docker-compose.yml con PostgreSQL di test
• ✅ CI GitHub Actions

Documentazione:

• pytest-asyncio
• httpx AsyncClient
• SQLAlchemy async testing