Módulo 6: Mocking, Fixtures y Validation Loops
Proyecto del Módulo: Validation Pipeline
Proyecto del Módulo: Validation Pipeline
Descripción del proyecto
Has aprendido mocking para reemplazar servicios externos, fixtures avanzadas para organizar datos de test, y validation loops para automatizar el ciclo de corrección con Claude Code. Ahora vas a integrar todo construyendo un pipeline de procesamiento de texto que depende de una API de AI y una database — completamente testeado con mocks, fixtures, y al menos un validation loop.
El pipeline recibe texto, lo envía a una API de AI para análisis de sentimiento, almacena el resultado en database, y retorna un resumen. En producción dependería de servicios reales. En tests, todo está mockeado: la API retorna respuestas controladas, la database es un dict en memoria, y los tests son deterministas y rápidos.
Objetivo del Proyecto
Construir un pipeline con servicios externos mockeados, fixtures organizadas, y validation loops para iterar automáticamente.
Al completar este proyecto:
- ✅ Tendrás mocks para 2 servicios externos (API de AI + database)
- ✅ Tendrás fixtures organizadas en conftest.py (global + por directorio)
- ✅ Habrás ejecutado al menos 1 validation loop completo
- ✅ Los tests serán deterministas, rápidos, e independientes
- ✅ Tendrás ≥85% coverage en el pipeline
Especificaciones Técnicas
Stack Tecnológico
- Lenguaje: Python 3.10+
- Testing: pytest, pytest-mock, pytest-cov
- AI: Claude Code
- Dependencias: pytest, pytest-mock, pytest-cov
Setup Inicial
mkdir validation-pipeline
cd validation-pipeline
python -m venv venv
source venv/bin/activate
pip install pytest pytest-mock pytest-cov
Estructura del Proyecto
validation-pipeline/
├── pipeline/
│ ├── __init__.py
│ ├── analyzer.py ← Llama a API de AI (tú creas con TDD + mocks)
│ ├── storage.py ← Interactúa con database (tú creas con TDD + mocks)
│ ├── processor.py ← Orquesta el pipeline (tú creas con validation loop)
│ └── models.py ← Modelos de datos (dado)
├── tests/
│ ├── __init__.py
│ ├── conftest.py ← Fixtures globales
│ ├── unit/
│ │ ├── __init__.py
│ │ ├── conftest.py ← Fixtures de unit
│ │ ├── test_analyzer.py
│ │ └── test_storage.py
│ └── integration/
│ ├── __init__.py
│ ├── conftest.py ← Fixtures de integration
│ └── test_processor.py
├── pyproject.toml
└── requirements.txt
El Código Base
models.py (dado)
"""Data models for the pipeline."""
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
@dataclass
class AnalysisRequest:
text: str
language: str = "es"
max_length: int = 1000
def __post_init__(self):
if not self.text.strip():
raise ValueError("Text cannot be empty")
if len(self.text) > self.max_length:
raise ValueError(f"Text exceeds max length of {self.max_length}")
@dataclass
class SentimentResult:
sentiment: str # "positive", "negative", "neutral"
confidence: float # 0.0 - 1.0
keywords: list[str] = field(default_factory=list)
def __post_init__(self):
if self.sentiment not in ("positive", "negative", "neutral"):
raise ValueError(f"Invalid sentiment: {self.sentiment}")
if not 0.0 <= self.confidence <= 1.0:
raise ValueError(f"Confidence must be 0.0-1.0, got {self.confidence}")
@dataclass
class AnalysisRecord:
id: Optional[int] = None
text: str = ""
sentiment: str = ""
confidence: float = 0.0
keywords: list[str] = field(default_factory=list)
created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat())
Qué Construir
1. analyzer.py — Servicio de AI (con mocks)
class SentimentAnalyzer:
def __init__(self, api_key: str, base_url: str = "https://api.ai-service.com"):
self.api_key = api_key
self.base_url = base_url
def analyze(self, request: AnalysisRequest) -> SentimentResult:
"""Llama a la API de AI para análisis de sentimiento."""
# En producción: hace HTTP POST a self.base_url
# En tests: mockeado
...
def batch_analyze(self, requests: list[AnalysisRequest]) -> list[SentimentResult]:
"""Analiza múltiples textos."""
...
Tests con mocks:
- Mock de la API HTTP (success, error, timeout)
- Mock de respuestas de AI (positive, negative, neutral)
- Test de retry logic cuando la API falla
2. storage.py — Database (con mocks)
class AnalysisStorage:
def __init__(self, connection_string: str):
self.connection_string = connection_string
def save(self, record: AnalysisRecord) -> int:
"""Guarda un resultado en database. Retorna el ID."""
...
def get(self, record_id: int) -> Optional[AnalysisRecord]:
"""Obtiene un resultado por ID."""
...
def list_by_sentiment(self, sentiment: str) -> list[AnalysisRecord]:
"""Lista resultados filtrados por sentimiento."""
...
Tests con mocks:
- Mock de database operations (insert, query, error)
- Test de connection errors
- Test de data integrity
3. processor.py — Pipeline (con validation loop)
class TextProcessor:
def __init__(self, analyzer: SentimentAnalyzer, storage: AnalysisStorage):
self.analyzer = analyzer
self.storage = storage
def process(self, text: str) -> AnalysisRecord:
"""Pipeline completo: validar → analizar → guardar → retornar."""
...
def process_batch(self, texts: list[str]) -> list[AnalysisRecord]:
"""Procesa múltiples textos."""
...
def get_stats(self) -> dict:
"""Retorna estadísticas: count por sentimiento."""
...
Usa validation loop para implementar esto:
Escribe los tests primero, luego pide a Claude Code que implemente TextProcessor usando validation loops.
Proceso Paso a Paso
Paso 1: Tests de analyzer con mocks
# tests/unit/test_analyzer.py
def test_analyze_positive_text(mocker):
mock_post = mocker.patch("pipeline.analyzer.requests.post")
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = {
"sentiment": "positive",
"confidence": 0.95,
"keywords": ["excellent", "recommended"]
}
analyzer = SentimentAnalyzer(api_key="test-key")
result = analyzer.analyze(AnalysisRequest(text="Excellent product"))
assert result.sentiment == "positive"
assert result.confidence == 0.95
Paso 2: Tests de storage con mocks
Paso 3: Tests de processor (integration) + validation loop
Escribe los tests de TextProcessor y usa un validation loop:
Implementa TextProcessor en pipeline/processor.py para que pasen
los tests en tests/integration/test_processor.py.
Después de implementar, ejecuta: pytest tests/integration/ -v
Si algún test falla, lee el error, corrige, y re-ejecuta.
Repite hasta que todos pasen.
Constraints: usa los interfaces de SentimentAnalyzer y AnalysisStorage.
Paso 4: Fixtures profesionales
# tests/conftest.py (global)
@pytest.fixture
def mock_analyzer(mocker):
analyzer = mocker.MagicMock(spec=SentimentAnalyzer)
analyzer.analyze.return_value = SentimentResult(
sentiment="positive", confidence=0.9, keywords=["test"]
)
return analyzer
@pytest.fixture
def mock_storage(mocker):
storage = mocker.MagicMock(spec=AnalysisStorage)
storage.save.return_value = 1
return storage
Paso 5: Medir coverage y cerrar gaps
pytest --cov=pipeline --cov-report=term-missing --cov-branch tests/
Rúbrica de Evaluación (100 puntos)
Mocks (30 puntos)
- (15 pts) Mocks funcionales para API de AI (success, error, timeout)
- (10 pts) Mocks funcionales para database (CRUD, connection error)
- (5 pts) Mocks verifican que se llaman correctamente (assert_called)
Fixtures (25 puntos)
- (10 pts) conftest.py global con fixtures compartidas
- (10 pts) conftest.py por directorio (unit/, integration/)
- (5 pts) Factory fixtures para datos variables
Validation Loop (25 puntos)
- (15 pts) Al menos 1 validation loop ejecutado y documentado
- (10 pts) El loop resultó en implementación funcional (tests green)
Coverage y Organización (20 puntos)
- (10 pts) ≥85% coverage del pipeline
- (5 pts) Tests organizados por nivel
- (5 pts) Naming descriptivo en toda la suite
Extra Credit (hasta +10 puntos)
- (+5 pts) Property-based test con hypothesis para el processor
- (+5 pts) Mock de retry logic con backoff exponencial
Errores Comunes
Error 1: Mockear la clase que estás testeando
Causa: Mockear TextProcessor en test_processor.py.
Solución: Mockea las dependencias (analyzer, storage), no la clase bajo test. El processor es real; sus dependencias son mocks.
Error 2: Validation loop sin tests claros
Causa: Tests ambiguos que Claude Code no puede resolver.
Solución: Cada test debe tener nombre descriptivo, assertion clara, y un solo motivo de fallo.
Error 3: Fixtures con scope incorrecto
Causa: Usar scope="session" para fixtures que necesitan reset entre tests.
Solución: Mocks deben ser scope="function" (default) para que cada test tenga mocks frescos.
Error 4: No verificar que los mocks se usaron
Causa: Los tests pasan pero el código real no llama al mock.
Solución: Usa assert_called_once(), assert_called_with() para verificar que la interacción ocurrió.
Recursos para el Proyecto
- pytest-mock Documentation - Plugin de mocking para pytest
- unittest.mock - Referencia oficial
- pytest Fixtures - Fixtures avanzadas
- pytest-cov - Coverage con pytest
- Martin Fowler: Mocks Aren't Stubs - Filosofía de mocking
Conexión con Siguiente Módulo
Lo que construiste hoy es la base del proyecto final:
- Módulo 7 (Estrategia): Definirás la estrategia de testing para una app completa
- Módulo 8 (Proyecto Final): Replicarás este patrón de mocks + fixtures + validation loops a mayor escala
Phase 2 completada. Tienes el workflow TDD, coverage, edge cases, mocking, fixtures, y validation loops. Phase 3 es donde lo juntas todo en un proyecto profesional.
Módulo 6, Cápsula 06 — Testing with Claude Code Guide Pipeline profesional con mocks, fixtures y validation loops