Testing con override de dependencias y base de datos

Override de dependencias: el superpoder del testing

Una de las características más poderosas de FastAPI para el testing es la capacidad de sobreescribir dependencias en tiempo de test. El atributo app.dependency_overrides es un diccionario donde puedes registrar sustitutos para cualquier dependencia, y FastAPI los usará automáticamente en todos los tests.

Aplicación de ejemplo con dependencias

Partimos de una API con dependencias reales:

# app/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

DATABASE_URL = "postgresql://usuario:contrasena@localhost/mi_bd"

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)

class Base(DeclarativeBase):
    pass

def get_db():
    """Dependencia que proporciona una sesión de base de datos."""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# app/main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from app.database import get_db, Base
from app.models import Producto
from app.schemas import ProductoCrear, ProductoRespuesta

app = FastAPI()

@app.post("/productos", response_model=ProductoRespuesta, status_code=201)
def crear_producto(producto: ProductoCrear, db: Session = Depends(get_db)):
    db_producto = Producto(**producto.model_dump())
    db.add(db_producto)
    db.commit()
    db.refresh(db_producto)
    return db_producto

@app.get("/productos/{producto_id}", response_model=ProductoRespuesta)
def obtener_producto(producto_id: int, db: Session = Depends(get_db)):
    producto = db.query(Producto).filter(Producto.id == producto_id).first()
    if not producto:
        raise HTTPException(status_code=404, detail="Producto no encontrado")
    return producto

Configurar la base de datos de testing

Para los tests, usamos SQLite en memoria, que es más rápida y no requiere un servidor:

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

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

# Base de datos SQLite en memoria para tests
SQLALCHEMY_DATABASE_URL = "sqlite://"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,  # Misma conexión para todos los threads
)

TestingSessionLocal = sessionmaker(bind=engine)

# Crear todas las tablas en la BD de testing
Base.metadata.create_all(bind=engine)

def override_get_db():
    """Sustituto de get_db que usa SQLite en memoria."""
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()

@pytest.fixture
def client():
    """Fixture que proporciona TestClient con BD de testing."""
    # Registrar el override ANTES de crear el cliente
    app.dependency_overrides[get_db] = override_get_db

    with TestClient(app) as test_client:
        yield test_client

    # Limpiar después del test
    app.dependency_overrides.clear()
    # Resetear las tablas entre tests
    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)

Ahora los tests usan SQLite en memoria en lugar de PostgreSQL real:

# tests/test_productos.py
def test_crear_producto(client):
    """Test que verifica la creación de un producto en la BD de testing."""
    datos = {"nombre": "Laptop", "precio": 999.99, "stock": 10}
    respuesta = client.post("/productos", json=datos)

    assert respuesta.status_code == 201
    producto = respuesta.json()
    assert producto["nombre"] == "Laptop"
    assert producto["precio"] == 999.99
    assert "id" in producto

def test_obtener_producto_creado(client):
    """Test que crea un producto y luego lo obtiene."""
    # Crear
    crear = client.post("/productos", json={"nombre": "Mouse", "precio": 29.99, "stock": 50})
    producto_id = crear.json()["id"]

    # Obtener
    respuesta = client.get(f"/productos/{producto_id}")
    assert respuesta.status_code == 200
    assert respuesta.json()["nombre"] == "Mouse"

Override de autenticación

Testear endpoints protegidos requiere simular el usuario autenticado:

# app/auth.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

class UsuarioActual(BaseModel):
    id: int
    username: str
    es_admin: bool = False

def get_current_user(token: str = Depends(oauth2_scheme)) -> UsuarioActual:
    """Dependencia real que valida el JWT."""
    # Aquí iría la lógica real de validación del token
    ...

# tests/conftest.py (ampliado)
from app.auth import get_current_user, UsuarioActual

# Usuario simulado para tests
USUARIO_TEST = UsuarioActual(id=1, username="usuario_test", es_admin=False)
ADMIN_TEST = UsuarioActual(id=2, username="admin_test", es_admin=True)

@pytest.fixture
def client_autenticado():
    """Fixture con usuario normal autenticado."""
    app.dependency_overrides[get_db] = override_get_db
    app.dependency_overrides[get_current_user] = lambda: USUARIO_TEST

    with TestClient(app) as test_client:
        yield test_client

    app.dependency_overrides.clear()
    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)

@pytest.fixture
def client_admin():
    """Fixture con usuario administrador autenticado."""
    app.dependency_overrides[get_db] = override_get_db
    app.dependency_overrides[get_current_user] = lambda: ADMIN_TEST

    with TestClient(app) as test_client:
        yield test_client

    app.dependency_overrides.clear()
    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)

Tests con autenticación:

# tests/test_auth.py
def test_endpoint_sin_token(client):
    """Verifica que un endpoint protegido rechaza peticiones sin token."""
    respuesta = client.get("/perfil")
    assert respuesta.status_code == 401

def test_endpoint_con_usuario(client_autenticado):
    """Verifica que un endpoint protegido acepta peticiones con usuario válido."""
    respuesta = client_autenticado.get("/perfil")
    assert respuesta.status_code == 200
    assert respuesta.json()["username"] == "usuario_test"

def test_endpoint_admin_rechaza_usuario_normal(client_autenticado):
    """Verifica que solo admins acceden a rutas de administración."""
    respuesta = client_autenticado.get("/admin/usuarios")
    assert respuesta.status_code == 403

def test_endpoint_admin_acepta_admin(client_admin):
    """Verifica que un administrador sí puede acceder a rutas de admin."""
    respuesta = client_admin.get("/admin/usuarios")
    assert respuesta.status_code == 200

Tests con transacciones para aislamiento

Para garantizar que los tests son completamente independientes, podemos usar transacciones que se revierten al final:

# tests/conftest.py (con transacciones)
import pytest
from sqlalchemy import event

@pytest.fixture
def db_session():
    """Fixture que proporciona una sesión con transacción revertible."""
    connection = engine.connect()
    transaction = connection.begin()
    session = TestingSessionLocal(bind=connection)

    yield session

    session.close()
    transaction.rollback()
    connection.close()

@pytest.fixture
def client_con_transaccion(db_session):
    """Fixture que usa la sesión de transacción para aislar los tests."""
    def override_get_db_transaccion():
        yield db_session

    app.dependency_overrides[get_db] = override_get_db_transaccion

    with TestClient(app) as test_client:
        yield test_client

    app.dependency_overrides.clear()

Con este patrón, cada test opera dentro de una transacción que se revierte al final, dejando la base de datos exactamente igual que al principio, sin importar qué datos se hayan insertado, modificado o eliminado.

Tests asíncronos con pytest-asyncio

Para testear funciones y endpoints async, necesitas pytest-asyncio:

pip install pytest-asyncio

Configuración en pyproject.toml:

[tool.pytest.ini_options]
asyncio_mode = "auto"

Tests asíncronos:

# tests/test_async.py
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app

@pytest.mark.asyncio
async def test_raiz_asincrono():
    """Test asíncrono usando AsyncClient de httpx."""
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as ac:
        respuesta = await ac.get("/")

    assert respuesta.status_code == 200
    assert respuesta.json() == {"mensaje": "API de Tareas"}

@pytest.mark.asyncio
async def test_flujo_completo_asincrono():
    """Test asíncrono que verifica un flujo completo de CRUD."""
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as ac:
        # Crear
        crear = await ac.post("/tareas", json={"titulo": "Async Test"})
        assert crear.status_code == 201
        tarea_id = crear.json()["id"]

        # Obtener
        obtener = await ac.get(f"/tareas/{tarea_id}")
        assert obtener.status_code == 200

        # Eliminar
        eliminar = await ac.delete(f"/tareas/{tarea_id}")
        assert eliminar.status_code == 204

Resumen de buenas prácticas

✓ Un test, una responsabilidad: cada test verifica una sola cosa
✓ Tests independientes: no deben depender del orden de ejecución
✓ Nombres descriptivos: test_crear_producto_sin_nombre_devuelve_422
✓ Cubrir casos de éxito Y casos de error
✓ Usar fixtures para evitar duplicación de código de setup
✓ Medir cobertura con pytest-cov y aspirar al 80%+ en código de negocio
✓ Limpiar la BD entre tests para garantizar aislamiento