Proyecto del Módulo: Test Pyramid Completa

Descripción del proyecto

Has aprendido la estrategia (test pyramid y trade-offs), la técnica de integration testing (FastAPI TestClient), database testing (fixtures), y E2E testing (flujos completos). Ahora vas a integrar todo construyendo una test pyramid completa para una API REST de gestión de tareas.

La API ya está implementada — tu trabajo es crear tests en los tres niveles con Claude Code: unit tests para la lógica de negocio, integration tests para los endpoints, y E2E tests para flujos completos. Cada nivel usa prompts diferenciados y verifica aspectos diferentes del sistema.

Este proyecto cierra Phase 1 de la guía. Al completarlo, tendrás dominio de los tres niveles de testing y sabrás cuándo usar cada uno. Es el template exacto que replicarás a mayor escala en el proyecto final (Módulo 8).

---

Objetivo del Proyecto

Construir una test pyramid completa con tests en los tres niveles para una API REST, usando Claude Code como generador y tu criterio estratégico para decidir qué testear en cada nivel.

Al completar este proyecto:

• ✅ Tendrás tests en los 3 niveles: unit, integration, E2E
• ✅ Habrás usado prompts diferenciados para cada nivel con Claude Code
• ✅ La pyramid seguirá las proporciones recomendadas (~70% unit, ~20% integration, ~10% E2E)
• ✅ Los tests serán independientes, organizados por nivel, y con naming descriptivo

---

Especificaciones Técnicas

Stack Tecnológico

• Lenguaje: Python 3.10+
• Framework: FastAPI
• Testing: pytest + httpx (para TestClient)
• AI: Claude Code
• Dependencias: fastapi, uvicorn, httpx, pytest

Setup Inicial

mkdir test-pyramid-project
cd test-pyramid-project

python -m venv venv
source venv/bin/activate

pip install fastapi uvicorn httpx pytest

Estructura del Proyecto

test-pyramid-project/
├── main.py                ← API REST (dado)
├── models.py              ← Modelos y lógica de negocio (dado)
├── database.py            ← Base de datos en memoria (dado)
├── tests/
│   ├── __init__.py
│   ├── conftest.py        ← Fixtures compartidas (tú creas)
│   ├── unit/
│   │   ├── __init__.py
│   │   └── test_models.py ← Unit tests (tú generas con Claude Code)
│   ├── integration/
│   │   ├── __init__.py
│   │   └── test_endpoints.py ← Integration tests (tú generas)
│   └── e2e/
│       ├── __init__.py
│       └── test_flows.py  ← E2E tests (tú generas)
├── requirements.txt
└── venv/

---

El Código a Testear

database.py

# database.py
"""Simple in-memory database for testing."""

_db: dict[str, list[dict]] = {"tasks": []}
_id_counter: dict[str, int] = {"tasks": 0}


def get_db():
    return _db


def reset_db():
    _db["tasks"] = []
    _id_counter["tasks"] = 0


def next_id(collection: str) -> int:
    _id_counter[collection] += 1
    return _id_counter[collection]

models.py

# models.py
"""Business logic and validation for tasks."""

from datetime import datetime
from typing import Optional


VALID_PRIORITIES = ["low", "medium", "high"]
VALID_STATUSES = ["pending", "in_progress", "completed"]


def validate_task(title: str, priority: str = "medium") -> dict:
    """Validate task data. Returns dict with is_valid and errors."""
    errors = []
    
    if not isinstance(title, str):
        errors.append("Title must be a string")
    elif not title.strip():
        errors.append("Title cannot be empty")
    elif len(title) > 200:
        errors.append("Title cannot exceed 200 characters")
    
    if priority not in VALID_PRIORITIES:
        errors.append(f"Priority must be one of: {', '.join(VALID_PRIORITIES)}")
    
    return {"is_valid": len(errors) == 0, "errors": errors}


def create_task_dict(
    task_id: int,
    title: str,
    priority: str = "medium",
    status: str = "pending",
) -> dict:
    """Create a task dictionary with all fields."""
    return {
        "id": task_id,
        "title": title.strip(),
        "priority": priority,
        "status": status,
        "created_at": datetime.utcnow().isoformat(),
        "updated_at": None,
    }


def can_transition_status(current: str, new: str) -> bool:
    """Check if status transition is valid."""
    allowed = {
        "pending": ["in_progress"],
        "in_progress": ["completed", "pending"],
        "completed": [],
    }
    return new in allowed.get(current, [])


def filter_tasks(
    tasks: list[dict],
    status: Optional[str] = None,
    priority: Optional[str] = None,
) -> list[dict]:
    """Filter tasks by status and/or priority."""
    result = tasks
    
    if status:
        if status not in VALID_STATUSES:
            raise ValueError(f"Invalid status: {status}")
        result = [t for t in result if t["status"] == status]
    
    if priority:
        if priority not in VALID_PRIORITIES:
            raise ValueError(f"Invalid priority: {priority}")
        result = [t for t in result if t["priority"] == priority]
    
    return result


def sort_tasks(tasks: list[dict], by: str = "created_at", reverse: bool = False) -> list[dict]:
    """Sort tasks by field."""
    valid_fields = ["created_at", "priority", "title"]
    if by not in valid_fields:
        raise ValueError(f"Cannot sort by '{by}'. Use: {', '.join(valid_fields)}")
    
    priority_order = {"high": 0, "medium": 1, "low": 2}
    
    if by == "priority":
        return sorted(tasks, key=lambda t: priority_order.get(t["priority"], 99), reverse=reverse)
    
    return sorted(tasks, key=lambda t: t.get(by, ""), reverse=reverse)

main.py

# main.py
"""FastAPI REST API for task management."""

from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel, Field
from typing import Optional

from database import get_db, next_id, reset_db
from models import (
    validate_task, create_task_dict, can_transition_status,
    filter_tasks, sort_tasks, VALID_PRIORITIES, VALID_STATUSES,
)

app = FastAPI(title="Task Manager API")


class TaskCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    priority: str = Field(default="medium")


class TaskUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=1, max_length=200)
    priority: Optional[str] = None
    status: Optional[str] = None


@app.get("/health")
def health_check():
    return {"status": "healthy", "service": "task-manager"}


@app.get("/tasks")
def list_tasks(
    status: Optional[str] = Query(None),
    priority: Optional[str] = Query(None),
    sort_by: str = Query("created_at"),
):
    db = get_db()
    tasks = db["tasks"]
    
    try:
        if status or priority:
            tasks = filter_tasks(tasks, status=status, priority=priority)
        tasks = sort_tasks(tasks, by=sort_by)
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    
    return tasks


@app.get("/tasks/{task_id}")
def get_task(task_id: int):
    db = get_db()
    task = next((t for t in db["tasks"] if t["id"] == task_id), None)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task


@app.post("/tasks", status_code=201)
def create_task(task: TaskCreate):
    validation = validate_task(task.title, task.priority)
    if not validation["is_valid"]:
        raise HTTPException(status_code=400, detail=validation["errors"])
    
    db = get_db()
    task_id = next_id("tasks")
    new_task = create_task_dict(task_id, task.title, task.priority)
    db["tasks"].append(new_task)
    return new_task


@app.put("/tasks/{task_id}")
def update_task(task_id: int, updates: TaskUpdate):
    db = get_db()
    task = next((t for t in db["tasks"] if t["id"] == task_id), None)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    
    if updates.title is not None:
        task["title"] = updates.title.strip()
    
    if updates.priority is not None:
        if updates.priority not in VALID_PRIORITIES:
            raise HTTPException(status_code=400, detail=f"Invalid priority: {updates.priority}")
        task["priority"] = updates.priority
    
    if updates.status is not None:
        if not can_transition_status(task["status"], updates.status):
            raise HTTPException(
                status_code=400,
                detail=f"Cannot transition from '{task['status']}' to '{updates.status}'"
            )
        task["status"] = updates.status
    
    from datetime import datetime
    task["updated_at"] = datetime.utcnow().isoformat()
    return task


@app.delete("/tasks/{task_id}")
def delete_task(task_id: int):
    db = get_db()
    task = next((t for t in db["tasks"] if t["id"] == task_id), None)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    
    db["tasks"] = [t for t in db["tasks"] if t["id"] != task_id]
    return {"message": "Task deleted", "id": task_id}

---

Qué Testear en Cada Nivel

Unit tests (tests/unit/test_models.py)

Testea las funciones de models.py aisladas — sin HTTP, sin database:

• validate_task: títulos válidos/inválidos, prioridades, strings vacíos, límite de caracteres
• create_task_dict: campos correctos, strip de título, valores default
• can_transition_status: transiciones válidas e inválidas entre estados
• filter_tasks: filtrado por status, priority, ambos, valores inválidos
• sort_tasks: ordenar por diferentes campos, reverse, campo inválido

Target: 15-20 unit tests

Integration tests (tests/integration/test_endpoints.py)

Testea los endpoints HTTP con TestClient — status codes, response bodies:

• GET /health: retorna 200 con status healthy
• GET /tasks: retorna lista, funciona con filtros
• GET /tasks/{id}: retorna task específica, 404 si no existe
• POST /tasks: crea task (201), falla con datos inválidos (400/422)
• PUT /tasks/{id}: actualiza campos, valida transiciones de status
• DELETE /tasks/{id}: elimina task, 404 si no existe

Target: 8-12 integration tests

E2E tests (tests/e2e/test_flows.py)

Testea flujos completos que simulan un usuario real:

• Flow 1: Task lifecycle — crear → obtener → actualizar → completar → eliminar
• Flow 2: Filtrado — crear varias tasks → filtrar por status → filtrar por priority
• Flow 3: Status transitions — crear → start (in_progress) → complete → verificar que no puede volver a pending

Target: 3-5 E2E tests

---

Proceso Paso a Paso

Paso 1: Crear conftest.py

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from main import app
from database import reset_db


@pytest.fixture(autouse=True)
def clean_database():
    """Reset database before each test."""
    reset_db()
    yield
    reset_db()


@pytest.fixture
def client():
    """FastAPI TestClient."""
    return TestClient(app)


@pytest.fixture
def sample_task(client):
    """Create and return a sample task."""
    response = client.post("/tasks", json={"title": "Sample Task", "priority": "medium"})
    return response.json()

Paso 2: Generar unit tests con Claude Code

Prompt:

Genera unit tests profesionales para las funciones en models.py.
Cubrir: happy path, edge cases (vacíos, límites), error handling.
Usa parametrize para validate_task y can_transition_status.
Patrón AAA. Naming descriptivo.

Paso 3: Generar integration tests

Prompt:

Genera integration tests para los endpoints en main.py usando
FastAPI TestClient. Testear status codes, response bodies.
Incluir: 200/201 success, 404 not found, 400 bad request, 422 validation.
Usa la fixture client de conftest.py.

Paso 4: Generar E2E tests

Prompt:

Genera E2E tests que validen flujos completos de la Task Manager API.
Flow 1: Lifecycle completo (create→read→update→delete).
Flow 2: Crear varias tasks y filtrar por status y priority.
Flow 3: Status transitions válidas e inválidas.
Cada test simula un usuario real usando la API de principio a fin.

Paso 5: Evaluar, refinar, ejecutar

# Ejecutar todos los tests
pytest tests/ -v

# Ejecutar por nivel
pytest tests/unit/ -v
pytest tests/integration/ -v
pytest tests/e2e/ -v

# Verificar conteo
pytest tests/ --co -q

---

Criterios de Éxito

Tu proyecto está completo cuando:

• ✅ pytest tests/ -v → todos green
• ✅ 15-20 unit tests en tests/unit/
• ✅ 8-12 integration tests en tests/integration/
• ✅ 3-5 E2E tests en tests/e2e/
• ✅ Proporciones de pyramid respetadas (~70% unit, ~20% integration, ~10% E2E)
• ✅ Cada test es independiente (autouse fixture resetea DB)
• ✅ Naming descriptivo en toda la suite

---

Rúbrica de Evaluación (100 puntos)

Unit Tests (35 puntos)

• (15 pts) Todas las funciones de models.py testeadas
• (10 pts) Edge cases cubiertos (vacíos, límites, tipos inválidos)
• (5 pts) Parametrize usado para validate_task y can_transition_status
• (5 pts) Tests enfocados (un assert por test)

Integration Tests (30 puntos)

• (15 pts) Todos los endpoints testeados (GET, POST, PUT, DELETE)
• (10 pts) Status codes y response bodies verificados
• (5 pts) Error cases: 404, 400, 422

E2E Tests (20 puntos)

• (10 pts) Al menos 2 flujos completos (lifecycle + filtrado)
• (5 pts) Status transitions testeadas end-to-end
• (5 pts) Flujos independientes y reproducibles

Organización (15 puntos)

• (5 pts) Estructura de directorios correcta (unit/, integration/, e2e/)
• (5 pts) conftest.py con fixtures compartidas
• (5 pts) Naming descriptivo en toda la suite

Extra Credit (hasta +10 puntos)

• (+5 pts) Test de concurrent access (crear y eliminar en rápida sucesión)
• (+5 pts) Prompt adversarial que descubra un edge case no cubierto

---

Errores Comunes

Error 1: Tests dependientes por estado de database

Causa: No usar autouse fixture que resetea la DB.

Solución: La fixture clean_database en conftest.py con autouse=True garantiza que cada test empieza con DB limpia.

Error 2: E2E tests que son realmente integration tests

Causa: Un test que hace un solo POST y verifica el response es integration, no E2E.

Solución: E2E implica múltiples requests secuenciales que forman un flujo: create → read → update → delete. Si es un solo request, es integration.

Error 3: Unit tests que importan FastAPI

Causa: Unit tests no deben tocar HTTP ni la app. Solo testean funciones puras de models.py.

Solución: test_models.py importa from models import ..., nunca from main import app.

Error 4: No testear error cases en integration

Causa: Solo testear happy path (200, 201).

Solución: Cada endpoint necesita al menos un test de error: 404 para resources que no existen, 400/422 para inputs inválidos.

Error 5: Demasiados E2E tests

Causa: Escribir 15 E2E tests en vez de 3-5.

Solución: E2E tests son para flujos críticos. Si tienes más de 5, probablemente algunos deberían ser integration tests.

---

Recursos para el Proyecto

1. FastAPI Testing - Documentación oficial de testing
2. pytest Fixtures - Referencia de fixtures
3. Martin Fowler: Test Pyramid - La referencia original
4. httpx Documentation - Cliente HTTP usado por TestClient
5. Ham Vocke: Practical Test Pyramid - Guía práctica

---

Conexión con Siguiente Módulo

Lo que construiste hoy se usa directamente en Phase 2:

• Módulo 4 (TDD Workflow): Usarás el ciclo red-green-refactor con tests en los 3 niveles
• Módulo 5 (Coverage): Medirás coverage de tu pyramid y descubrirás gaps
• Módulo 6 (Mocking): Aprenderás a mockear servicios externos en integration tests
• Módulo 8 (Proyecto Final): Construirás una pyramid completa a mayor escala

Completaste Phase 1. Tienes spec-first, unit tests, integration tests, y E2E tests. Phase 2 los integra en workflows profesionales.

---

Módulo 3, Cápsula 06 — Testing with Claude Code Guide Tu primera test pyramid completa — de unit a E2E