2. Tests End-to-End

Descripción

Los tests end-to-end (E2E) ejecutan el flujo completo: input → formatter → LLM → parse → output. Sin mocks en el camino crítico. Esta cápsula cubre cómo estructurar E2E tests para apps AI, implementar budget controls obligatorios, qué assertions usar cuando el output varía, y cómo integrar E2E en tu pipeline de CI/CD sin que se vuelvan caros o frágiles.

Qué es un E2E test en contexto AI

Un E2E test llama a componentes reales — incluyendo el LLM:

Unit test (M2):
  input → [MOCK LLM] → parse → output
  
E2E test (M3):
  input → formatter → [LLM REAL] → parse → processor → output
                           ↑
                    API call real → costo real → tiempo real

Lo que verifica un E2E que los unit tests no pueden:

• El prompt funciona con el LLM real (no solo con el mock)
• La cadena completa produce output usable para el usuario final
• Los timeouts, rate limits y errores reales se manejan bien
• El modelo produce outputs que tu parser maneja correctamente

Estructura de un E2E test bien formado

import pytest
import os

@pytest.mark.integration
@pytest.mark.e2e
@pytest.mark.skipif(
    not os.getenv("OPENAI_API_KEY"),
    reason="OPENAI_API_KEY no configurada — E2E tests requieren API real"
)
def test_analyze_sentiment_e2e():
    """
    Flujo completo: texto → LLM real → parse → output validado.
    
    Este test verifica que el sistema completo produce un resultado
    usable, no solo que la estructura es correcta (eso ya lo cubren
    los contract tests del M2).
    """
    # Arrange: input real y controlado
    text = "Acabé de usar este producto por primera vez y es absolutamente increíble."
    
    # Act: flujo completo, sin mocks
    result = analyze_sentiment(text)  # Llama al LLM real
    
    # Assert: flexible — no igualdad exacta, sino propiedades
    # Estructura (redundante con contract tests, pero útil para E2E)
    assert isinstance(result, dict), "El resultado debe ser un dict"
    assert "sentiment" in result
    assert "score" in result
    
    # Propiedades de la salida real del LLM
    assert result["sentiment"] in ["positivo", "negativo", "neutral"]
    assert 0.0 <= result["score"] <= 1.0
    
    # Para texto claramente positivo, el score no debe ser muy bajo
    # (esto es una assertion de calidad, no solo de estructura)
    assert result["score"] >= 0.5, \
        f"Para texto positivo, el score debería ser >=0.5, es {result['score']}"

Budget controls: implementación completa

Los budget controls son obligatorios en E2E tests. Sin ellos, un run accidental puede costar decenas de dólares.

Implementación básica: session budget

# tests/conftest.py

import pytest
import os
import threading

class BudgetTracker:
    """Rastrea el gasto en API calls durante la sesión de tests."""
    
    # Precios por 1M tokens (gpt-4o-mini, Enero 2025)
    PRICES = {
        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        "gpt-4o":       {"input": 2.50, "output": 10.00},
        "gpt-4":        {"input": 30.00, "output": 60.00},
    }
    
    def __init__(self, max_usd: float = 0.50):
        self.max_usd = max_usd
        self.spent = 0.0
        self._lock = threading.Lock()
        self.calls = []
    
    def add_cost(self, model: str, prompt_tokens: int, completion_tokens: int):
        prices = self.PRICES.get(model, self.PRICES["gpt-4o-mini"])
        cost = (
            prompt_tokens / 1_000_000 * prices["input"] +
            completion_tokens / 1_000_000 * prices["output"]
        )
        with self._lock:
            self.spent += cost
            self.calls.append({
                "model": model,
                "prompt_tokens": prompt_tokens,
                "completion_tokens": completion_tokens,
                "cost": cost
            })
        return cost
    
    def check_budget(self):
        if self.spent >= self.max_usd:
            raise pytest.skip.Exception(
                f"Budget E2E excedido: ${self.spent:.4f} >= ${self.max_usd:.2f}"
            )
    
    def report(self):
        return {
            "total_spent": self.spent,
            "max_budget": self.max_usd,
            "remaining": self.max_usd - self.spent,
            "total_calls": len(self.calls)
        }

@pytest.fixture(scope="session")
def e2e_budget():
    """Budget compartido por toda la sesión de integration tests."""
    max_budget = float(os.getenv("E2E_BUDGET_USD", "0.50"))
    tracker = BudgetTracker(max_usd=max_budget)
    yield tracker
    # Al final de la sesión, reportar el gasto
    report = tracker.report()
    print(f"\n💰 E2E Budget Report: ${report['total_spent']:.4f} / ${report['max_budget']:.2f}")
    print(f"   {report['total_calls']} API calls realizadas")

@pytest.fixture
def openai_client_e2e(e2e_budget):
    """
    Cliente OpenAI para E2E tests con budget tracking.
    
    Skip automático si el budget está excedido.
    """
    import openai
    
    e2e_budget.check_budget()  # Skip si budget excedido
    
    api_key = os.getenv("OPENAI_API_KEY")
    if not api_key:
        pytest.skip("OPENAI_API_KEY no configurada")
    
    client = openai.OpenAI(api_key=api_key)
    return client, e2e_budget

Uso del budget tracker en tests

@pytest.mark.integration
def test_sentiment_with_budget(openai_client_e2e):
    client, budget = openai_client_e2e
    
    # Hacer la llamada al LLM
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Analiza: 'Me encanta este producto'"}],
        temperature=0.0,
        max_tokens=200
    )
    
    # Registrar el costo
    budget.add_cost(
        model="gpt-4o-mini",
        prompt_tokens=response.usage.prompt_tokens,
        completion_tokens=response.usage.completion_tokens
    )
    
    raw = response.choices[0].message.content
    result = parse_json_response(raw)
    processed = process_sentiment_output(result)
    
    assert processed["sentiment"] in ["positivo", "negativo", "neutral"]
    assert 0 <= processed["score"] <= 1

Assertions robustas para E2E

El arte de los E2E tests está en las assertions. Demasiado estrictas → tests frágiles. Demasiado débiles → no detectan problemas.

Niveles de assertions

# ─── Nivel 1: Estructura (mínimo obligatorio) ───
def assertions_nivel_1(result):
    """Estas assertions son idénticas a los contract tests del M2."""
    assert isinstance(result, dict)
    assert "sentiment" in result
    assert result["sentiment"] in ["positivo", "negativo", "neutral"]
    assert 0 <= result["score"] <= 1

# ─── Nivel 2: Propiedades de calidad ───
def assertions_nivel_2(result, input_text):
    """Assertions que verifican que el output tiene sentido."""
    # Si el texto tiene palabras claramente positivas, el score debe ser alto
    clearly_positive_words = ["increíble", "excelente", "fantástico", "perfecto", "encanta"]
    if any(w in input_text.lower() for w in clearly_positive_words):
        assert result["score"] >= 0.6, \
            f"Para texto positivo, score debe ser >=0.6, es {result['score']}"
    
    # El explanation no debe ser vacío para textos con sentimiento claro
    if result["score"] < 0.3 or result["score"] > 0.7:
        assert len(result.get("explanation", "")) > 0

# ─── Nivel 3: Keywords del input ───
def assertions_nivel_3(result, input_text):
    """Verificar que el output está relacionado con el input."""
    # Al menos una palabra clave del input debe aparecer en los keywords o explanation
    input_words = set(input_text.lower().split()) - {"el", "la", "es", "un", "una", "de"}
    output_text = (result.get("explanation", "") + " ".join(result.get("keywords", []))).lower()
    
    relevant_found = any(word in output_text for word in input_words if len(word) > 3)
    assert relevant_found, \
        "El output no parece relacionado con el input — posible alucinación"

Assertions por tipo de prompt

# Para prompts de RESUMEN:
def assert_summary_quality(result: dict, original_text: str):
    # Longitud razonable (no más largo que el original)
    assert len(result["summary"]) <= len(original_text)
    # Mínimo informativo
    assert len(result["summary"].split()) >= 5
    # Confidence razonable
    assert result["confidence"] >= 0.5, "Confidence muy baja para texto estándar"

# Para prompts de CLASIFICACIÓN:
def assert_classification_quality(result: dict):
    # Confidence alta si la categoría es clara
    if result["confidence"] > 0.9:
        assert result["category"] != "unknown"
    # Tags deben ser relevantes (al menos 1 si hay categoría)
    if result["category"] != "unknown":
        assert len(result.get("tags", [])) >= 1

# Para prompts de EXTRACCIÓN:
def assert_extraction_quality(result: dict, input_text: str):
    # Las entidades extraídas deben aparecer en el input
    for entity in result.get("entities", []):
        assert entity["text"].lower() in input_text.lower(), \
            f"Entidad '{entity['text']}' no encontrada en el input — posible alucinación"

Skip condicional: la base de los E2E en CI

Los E2E tests deben ser opcionales, no bloquear el pipeline:

# Opción 1: skipif basado en variable de entorno
import pytest
import os

requires_api_key = pytest.mark.skipif(
    not os.getenv("OPENAI_API_KEY"),
    reason="Requiere OPENAI_API_KEY para E2E"
)

requires_integration = pytest.mark.skipif(
    os.getenv("SKIP_INTEGRATION", "true").lower() == "true",
    reason="Integration tests deshabilitados (SKIP_INTEGRATION=true)"
)

# Uso:
@requires_api_key
@requires_integration
@pytest.mark.integration
def test_sentiment_e2e():
    ...

# Opción 2: fixture que hace skip
@pytest.fixture
def integration_client():
    """Solo crea el cliente si tenemos API key y estamos en modo integration."""
    api_key = os.getenv("OPENAI_API_KEY")
    run_integration = os.getenv("RUN_INTEGRATION", "false").lower() == "true"
    
    if not api_key or not run_integration:
        pytest.skip("E2E tests requieren OPENAI_API_KEY y RUN_INTEGRATION=true")
    
    import openai
    return openai.OpenAI(api_key=api_key)

def test_with_conditional_fixture(integration_client):
    # Si no hay key, este test se salta automáticamente
    response = integration_client.chat.completions.create(...)

Modelo económico para E2E

Siempre usa el modelo más económico para tests:

# pytest.ini o conftest.py
E2E_MODEL = os.getenv("E2E_MODEL", "gpt-4o-mini")  # Default: el más barato

# Nunca en E2E tests (a menos que haya una razón específica):
# ❌ "gpt-4"          → ~200x más caro que gpt-4o-mini
# ❌ "gpt-4o"         → ~17x más caro
# ✅ "gpt-4o-mini"    → Balanceo calidad/costo óptimo para testing

def test_e2e_with_economic_model(integration_client):
    response = integration_client.chat.completions.create(
        model="gpt-4o-mini",     # Siempre explícito
        messages=[...],
        temperature=0.0,          # Sin aleatoriedad adicional
        max_tokens=200,           # Limitar tokens de output
    )

Comparación: E2E vs Unit vs Integration estándar

Timeout: protegerse de tests colgados

# pip install pytest-timeout

@pytest.mark.integration
@pytest.mark.timeout(30)  # Máximo 30 segundos
def test_sentiment_e2e_with_timeout(integration_client):
    result = analyze_sentiment("texto de prueba", client=integration_client)
    assert "sentiment" in result

# O configurar timeout global para todos los integration tests:
# pytest.ini
# [pytest]
# timeout = 30

E2E en CI/CD: configuración recomendada

# .github/workflows/ci.yml

jobs:
  unit-tests:
    name: Unit Tests (always)
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Run unit tests
        run: pytest -m "not integration" -v --tb=short
        # Siempre corre, sin API key, determinístico

  integration-tests:
    name: Integration Tests (main only)
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'  # Solo en main
    env:
      OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
      RUN_INTEGRATION: "true"
      E2E_BUDGET_USD: "0.25"  # Budget máximo por run
    steps:
      - uses: actions/checkout@v3
      - name: Run integration tests
        run: pytest -m integration -v --tb=short --timeout=60
        # Solo en main, con API key, con budget controlado

Ejercicios

Ejercicio 1: Escribir el E2E completo

Escribe un E2E test completo para el endpoint /analyze de la FastAPI app. Incluye: skip condicional, budget tracking, assertions de propiedades.

import pytest
import os
from fastapi.testclient import TestClient
from app.main import app

@pytest.mark.integration
@pytest.mark.e2e
@pytest.mark.skipif(
    not os.getenv("OPENAI_API_KEY") or not os.getenv("RUN_INTEGRATION"),
    reason="Requiere OPENAI_API_KEY y RUN_INTEGRATION=true"
)
@pytest.mark.timeout(30)
def test_analyze_endpoint_e2e():
    """E2E test del endpoint /analyze usando el LLM real."""
    client = TestClient(app)
    
    response = client.post("/analyze", json={
        "text": "Me encanta este producto, es absolutamente fantástico"
    })
    
    assert response.status_code == 200
    data = response.json()
    
    # Assertions de propiedades (no igualdad exacta)
    assert data["sentiment"] in ["positivo", "negativo", "neutral"]
    assert 0 <= data["score"] <= 1
    assert isinstance(data["keywords"], list)
    
    # Para texto claramente positivo:
    assert data["score"] >= 0.6, "Texto positivo debe tener score >= 0.6"
    assert data["sentiment"] == "positivo", "Texto positivo debe clasificarse como positivo"

Ejercicio 2: Implementar budget fixture

Implementa una versión simplificada del budget tracker que haga skip si se gasta más de $0.10:

@pytest.fixture(scope="session")
def simple_budget():
    budget = {"spent": 0.0, "max": 0.10}
    yield budget
    print(f"\nGasto E2E: ${budget['spent']:.4f}")

def register_cost(budget: dict, tokens: int, model: str = "gpt-4o-mini"):
    cost_per_token = 0.00000015  # gpt-4o-mini aprox
    budget["spent"] += tokens * cost_per_token
    if budget["spent"] >= budget["max"]:
        pytest.skip(f"Budget excedido: ${budget['spent']:.4f}")

@pytest.mark.integration
def test_with_simple_budget(simple_budget, integration_client):
    response = integration_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Hola"}],
        max_tokens=50
    )
    register_cost(simple_budget, response.usage.total_tokens)
    assert response.choices[0].message.content

Ejercicio 3: Assertions para prompt de clasificación

Para un prompt de clasificación de documentos (retorna category, confidence, tags), diseña 5 assertions robustas para E2E:

def assert_classification_e2e(result: dict, input_text: str):
    # 1. Estructura (siempre)
    assert "category" in result and "confidence" in result and "tags" in result
    
    # 2. Valores permitidos
    VALID_CATEGORIES = ["tecnología", "ciencia", "deportes", "política", "entretenimiento", "general"]
    assert result["category"] in VALID_CATEGORIES, f"Categoría inválida: {result['category']}"
    
    # 3. Rango de confidence
    assert 0.0 <= result["confidence"] <= 1.0
    
    # 4. Tags no vacíos cuando hay categoría clara
    if result["confidence"] >= 0.8:
        assert len(result["tags"]) >= 1, "Alta confidence debe tener al menos un tag"
    
    # 5. Tags son strings relevantes (no numéricos ni vacíos)
    for tag in result["tags"]:
        assert isinstance(tag, str) and len(tag.strip()) > 0

Ejercicio 4: E2E para flujo con error

Escribe un E2E test que verifica que la app maneja correctamente un texto vacío:

@pytest.mark.integration
@pytest.mark.e2e
@pytest.mark.skipif(not os.getenv("OPENAI_API_KEY"), reason="Requiere API key")
def test_empty_text_handled_e2e():
    """E2E: texto vacío debe retornar error controlado, no crash."""
    from fastapi.testclient import TestClient
    from app.main import app
    
    client = TestClient(app)
    
    response = client.post("/analyze", json={"text": ""})
    
    # FastAPI debe validar el input antes de llamar al LLM
    assert response.status_code == 422  # Unprocessable Entity (Pydantic validation)
    # El LLM NO debe ser llamado — el error es pre-LLM

Ejercicio 5: Timeout y retry

Combina timeout de 30s con retry de 2 intentos para un E2E test que puede ser lento:

@pytest.mark.integration
@pytest.mark.timeout(30)
@pytest.mark.flaky(reruns=2, reruns_delay=3)  # Requiere pytest-rerunfailures
def test_slow_endpoint_with_retry():
    """Test que puede ser lento — con retry y timeout."""
    result = analyze_sentiment("texto de prueba")
    
    # Assertions flexibles para evitar false negatives
    assert result["sentiment"] in ["positivo", "negativo", "neutral"]
    assert 0 <= result["score"] <= 1

Resumen

• E2E tests = flujo completo con LLM real — verifican lo que los mocks no pueden
• Budget controls son obligatorios — implementa el tracker antes de escribir tests
• Assertions robustas: propiedades, rangos, keywords — no igualdad exacta
• Skip condicional: por API key, por variable de entorno — E2E no deben bloquear CI
• Modelo económico: siempre gpt-4o-mini para tests, nunca gpt-4 sin razón
• Timeout: siempre @pytest.mark.timeout(30) — un LLM colgado no debe colgar el pipeline

Recursos adicionales

1. pytest-timeout — Timeouts por test
2. pytest-rerunfailures — Retry automático
3. OpenAI Usage dashboard — Monitorear costos reales
4. FastAPI TestClient — Para E2E de endpoints
5. GitHub Actions Secrets — Para API keys en CI
6. pytest skipif — Skip condicional
7. OpenAI API Rate Limits — Entender los límites de la API