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

  1. pytest-mock Documentation - Plugin de mocking para pytest
  2. unittest.mock - Referencia oficial
  3. pytest Fixtures - Fixtures avanzadas
  4. pytest-cov - Coverage con pytest
  5. 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