Testing por Capa
Pirámide de tests en DDD: unit, integration y e2e con sus estrategias de aislamiento.
La pirámide de tests no es una metáfora decorativa: es una estrategia de inversión de tiempo y dinero. Los tests unitarios son baratos, rápidos y deterministas porque no tocan I/O. Los tests de integración validan que los adaptadores se conectan correctamente a la infraestructura real, son más lentos y requieren entorno. Los tests e2e verifican flujos completos pero son los más frágiles y costosos. La pirámide dice que debes tener muchos de los primeros y pocos de los últimos, no que los últimos no valgan.
En DDD, la capa de dominio es la más rica en lógica y la más fácil de testear: no depende de nada externo. Un test unitario de un Aggregate solo necesita instanciar el objeto, llamar un método y verificar el estado resultante o los domain events emitidos. No hay base de datos, no hay red, no hay framework. Si necesitas un mock para testear el dominio, es una señal de que el dominio tiene una dependencia que no debería tener.
El patrón Fake Object (también llamado In-Memory Repository) es la herramienta central de aislamiento en tests unitarios de la capa de aplicación. Un `InMemory<Aggregate>Repository` implementa el mismo puerto ABC que la versión SQL, pero guarda los datos en un diccionario en memoria. Esto permite testear los Command Handlers completos —incluyendo la lógica de orquestación— sin tocar la base de datos y sin mocks frágiles que verifican llamadas en lugar de comportamiento.
Los tests de integración prueban los adaptadores contra la infraestructura real: `Sql<Aggregate>Repository` contra una base de datos real (normalmente SQLite en memoria o Postgres en Docker), `<Bus>EventBus` contra un broker real. Estos tests son más lentos y deben ejecutarse en CI pero no en cada pre-commit. Su valor es verificar que el mapeo ORM, las transacciones y las queries SQL realmente funcionan, lo que los fakes no pueden validar.
Los tests e2e lanzan la aplicación completa (FastAPI con sus dependencias reales) y hacen llamadas HTTP usando un cliente de test como `httpx.AsyncClient`. Prueban que el flujo completo funciona: HTTP request → command → domain → repository → response. Son los más cercanos a cómo el usuario final usa el sistema, pero los más difíciles de mantener porque son sensibles a cambios en contratos HTTP, esquemas de base de datos y configuración del entorno.
Una regla práctica: si un test falla con `ImportError: No module named 'sqlalchemy'`, está en el directorio equivocado. Los tests en `tests/unit/` no deberían necesitar ninguna dependencia de infraestructura. Organizar los tests en directorios que reflejan las capas hace que este error sea obvio y fácil de corregir antes de que se acumule deuda.
# tests/unit/ — pure domain logic, no I/O, uses in-memory fakes
# test_<aggregate>_<action>.py
# In-memory fake: implements the same port as the real adapter
class InMemory<Aggregate>Repository(<Aggregate>Repository):
def __init__(self) -> None:
self._store: dict[<AggregateId>, <Aggregate>] = {}
def find_by_id(self, id: <AggregateId>) -> <Aggregate> | None:
return self._store.get(id)
def save(self, aggregate: <Aggregate>) -> None:
self._store[aggregate.id] = aggregate
# Unit test: tests the Command Handler behavior (not implementation details)
def test_<do_something>_handler_<expected_outcome>():
repo = InMemory<Aggregate>Repository()
handler = <DoSomething>Handler(repo=repo)
command = <DoSomething>Command(<param>=<valid_value>)
handler.handle(command)
saved = repo.find_by_id(<valid_id>)
assert saved is not None
assert saved.<attribute> == <expected_value>
# tests/integration/ — adapters against real infra (DB, bus)
# test_sql_<aggregate>_repository.py
def test_sql_<aggregate>_repository_saves_and_retrieves(db_session):
repo = Sql<Aggregate>Repository(session=db_session)
aggregate = <Aggregate>Factory.create(<param>=<value>)
repo.save(aggregate)
db_session.flush()
retrieved = repo.find_by_id(aggregate.id)
assert retrieved is not None
assert retrieved.<attribute> == aggregate.<attribute>
# tests/e2e/ — full API flow with httpx.AsyncClient
# test_<endpoint>_e2e.py
async def test_post_<aggregate>_returns_201(async_client: AsyncClient):
payload = {"<field>": "<value>"}
response = await async_client.post("/<aggregates>", json=payload)
assert response.status_code == 201
assert "id" in response.json()Debugging lab
Detecta y corrige el error o la violación de diseño.
- 8.2.5.1
# BAD: unit test does real I/O — violates the unit test contract # tests/unit/<bounded_context>/application/test_<do_something>_handler.py import pytest from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from ...infrastructure.persistence.sql_<aggregate>_repository import Sql<Aggregate>Repository from ...application.commands.<do_something>.handler import <DoSomething>Handler from ...application.commands.<do_something>.command import <DoSomething>Command def test_handler_saves_aggregate(): engine = create_engine("sqlite:///:memory:") Session = sessionmaker(bind=engine) repo = Sql<Aggregate>Repository(session=Session()) # <-- real DB in a unit test! handler = <DoSomething>Handler(repo=repo) handler.handle(<DoSomething>Command(<param>=<value>))
- 8.2.5.2
# BAD: test verifies implementation details (how), not behavior (what) # tests/unit/<bounded_context>/application/test_<do_something>_handler.py from unittest.mock import MagicMock, patch def test_handler_calls_repository_save(): mock_repo = MagicMock() handler = <DoSomething>Handler(repo=mock_repo) handler.handle(<DoSomething>Command(<param>=<value>)) # This only verifies that save() was called — not that it worked mock_repo.save.assert_called_once()
- 8.2.5.3
# BAD: domain aggregate test asserts on internal structure, not behavior # tests/unit/<bounded_context>/domain/test_<aggregate>.py def test_<aggregate>_internal_state_after_action(): aggregate = <Aggregate>(id=<id>, <attribute>=<initial_value>) aggregate.<perform_action>(<new_value>) # Testing private implementation details assert aggregate._<attribute> == <new_value> # <-- accesses private field assert len(aggregate._events) == 1 # <-- accesses private list
- 8.2.5.4
# BAD: integration test belongs in unit/ because it uses a fake # tests/integration/<bounded_context>/test_<do_something>_handler_integration.py # This file is in tests/integration/ but uses no real infrastructure from tests.unit.fakes.in_memory_<aggregate>_repository import InMemory<Aggregate>Repository def test_handler_integration_with_fake_repo(): repo = InMemory<Aggregate>Repository() handler = <DoSomething>Handler(repo=repo) handler.handle(<DoSomething>Command(<param>=<value>)) assert repo.find_by_id(<expected_id>) is not None
- 8.2.5.5
# BAD: e2e test shares state between test cases — flaky test risk # tests/e2e/test_<aggregate>_api.py # Global shared state — one test can pollute another created_id = None async def test_create_<aggregate>(async_client: AsyncClient): global created_id response = await async_client.post("/<aggregates>", json={"<field>": "<value>"}) created_id = response.json()["id"] assert response.status_code == 201 async def test_get_<aggregate>(async_client: AsyncClient): # Depends on test above running first — order-dependent! response = await async_client.get(f"/<aggregates>/{created_id}") assert response.status_code == 200