Módulo 3: Integration Testing & Estrategias No-Determinísticas
7. Proyecto: Integration Test Suite
Descripción
Este es el mini-proyecto del Módulo 3. Expandirás la test suite de los Módulos 1-2 con: integration tests que llaman al LLM real con budget controls, semantic similarity assertions, property-based tests con Hypothesis, flaky test management, y configuración de tres entornos (mock/sandbox/real). Al terminar tendrás una suite completa: unit tests determinísticos + integration tests que manejan non-determinism de forma robusta.
Objetivos del proyecto
Al completar este proyecto tendrás:
- Budget tracker que limita el gasto en API calls durante los tests
- E2E tests con LLM real y assertions flexibles (máximo $0.50 por run)
- Al menos 2 tests con semantic similarity assertion
- Al menos 2 tests con Hypothesis (property-based, con mock)
- Flaky management: retry en tests inestables, quarantine para los problemáticos
- Configuración de entornos: mock por defecto, integration cuando hay API key
- CI/CD básico: configuración de GitHub Actions
Estructura del proyecto
tu-proyecto/
├── src/
│ └── app/
│ ├── __init__.py
│ ├── config.py
│ ├── parsers.py
│ ├── processors.py
│ ├── sentiment.py
│ └── main.py
├── tests/
│ ├── __init__.py
│ ├── helpers.py # create_openai_chat_response, etc.
│ ├── semantic.py # assert_semantically_similar, etc.
│ ├── conftest.py # Fixtures compartidas: budget, clients
│ ├── unit/ # M1-M2: contract, parsers, regression
│ │ ├── conftest.py
│ │ ├── contracts/
│ │ ├── parsers/
│ │ └── regression/
│ └── integration/ # M3: E2E, semantic, property
│ ├── __init__.py
│ ├── conftest.py # Fixtures específicas de integration
│ ├── test_e2e.py # E2E con LLM real
│ ├── test_semantic.py # Semantic similarity assertions
│ └── test_property.py # Property-based con Hypothesis
├── .github/
│ └── workflows/
│ └── tests.yml # CI/CD config
├── pytest.ini
└── requirements.txt
Paso 1: Actualizar tests/semantic.py
# tests/semantic.py
import numpy as np
from functools import lru_cache
from typing import Optional
def cosine_similarity(a: list[float], b: list[float]) -> float:
a_arr = np.array(a, dtype=float)
b_arr = np.array(b, dtype=float)
norm_a = np.linalg.norm(a_arr)
norm_b = np.linalg.norm(b_arr)
if norm_a == 0 or norm_b == 0:
return 0.0
return float(np.dot(a_arr, b_arr) / (norm_a * norm_b))
@lru_cache(maxsize=1)
def _get_local_model():
"""Carga el modelo de sentence-transformers una vez."""
try:
from sentence_transformers import SentenceTransformer
return SentenceTransformer("paraphrase-multilingual-MiniLM-L12-v2")
except ImportError:
return None
_embedding_cache: dict[str, list[float]] = {}
def get_embedding_local(text: str) -> list[float]:
"""Embedding local con sentence-transformers. Sin costo de API."""
if text in _embedding_cache:
return _embedding_cache[text]
model = _get_local_model()
if model is None:
raise ImportError("sentence-transformers no instalado. Ejecuta: pip install sentence-transformers")
embedding = model.encode(text, normalize_embeddings=True)
_embedding_cache[text] = embedding.tolist()
return _embedding_cache[text]
def get_embedding_openai(text: str, client) -> list[float]:
"""Embedding via OpenAI API."""
if text in _embedding_cache:
return _embedding_cache[text]
response = client.embeddings.create(
model="text-embedding-3-small",
input=text.strip()
)
embedding = response.data[0].embedding
_embedding_cache[text] = embedding
return embedding
def assert_semantically_similar(
actual: str,
expected: str,
threshold: float = 0.80,
client=None,
message: str = None
) -> float:
"""
Assertion semántica: falla si actual y expected no son suficientemente similares.
Args:
actual: El output real del LLM
expected: El significado esperado (descripción del contenido esperado)
threshold: Similitud mínima aceptable (default: 0.80)
client: Cliente OpenAI (si None, usa sentence-transformers local)
message: Mensaje de error personalizado
Returns:
La similitud calculada (útil para debugging)
"""
if client is not None:
emb_actual = get_embedding_openai(actual, client)
emb_expected = get_embedding_openai(expected, client)
else:
emb_actual = get_embedding_local(actual)
emb_expected = get_embedding_local(expected)
similarity = cosine_similarity(emb_actual, emb_expected)
default_msg = (
f"Similitud semántica {similarity:.3f} < umbral {threshold}\n"
f" Actual: '{actual[:100]}'\n"
f" Expected: '{expected[:100]}'"
)
assert similarity >= threshold, message or default_msg
return similarity
Paso 2: Actualizar tests/conftest.py
# tests/conftest.py
import pytest
import os
import threading
from unittest.mock import MagicMock, AsyncMock
import json
from tests.helpers import create_openai_chat_response, create_sentiment_response
# ─── Budget Tracker ────────────────────────────────────────────────────────
class BudgetTracker:
PRICES = {
"gpt-4o-mini": {"input": 0.15, "output": 0.60},
"gpt-4o": {"input": 2.50, "output": 10.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) -> float:
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, "cost": cost})
return cost
def check_budget(self):
if self.spent >= self.max_usd:
pytest.skip(f"Budget E2E excedido: ${self.spent:.4f} >= ${self.max_usd:.2f}")
@pytest.fixture(scope="session")
def e2e_budget():
max_budget = float(os.getenv("E2E_BUDGET_USD", "0.50"))
tracker = BudgetTracker(max_usd=max_budget)
yield tracker
print(f"\n💰 E2E Budget: ${tracker.spent:.4f} / ${tracker.max_usd:.2f} ({len(tracker.calls)} calls)")
# ─── Integration Client ─────────────────────────────────────────────────────
@pytest.fixture(scope="session")
def integration_client(e2e_budget):
"""
Cliente OpenAI real para integration tests.
Skip automático si no hay API key o se excedió el budget.
"""
api_key = os.getenv("OPENAI_API_KEY")
run_integration = os.getenv("RUN_INTEGRATION", "false").lower() == "true"
if not api_key:
pytest.skip("OPENAI_API_KEY no configurada — skip integration tests")
if not run_integration and not os.getenv("FULL_VALIDATION"):
pytest.skip("RUN_INTEGRATION no habilitado — usar RUN_INTEGRATION=true para integration")
e2e_budget.check_budget()
import openai
return openai.OpenAI(api_key=api_key)
# ─── Mock Factories ─────────────────────────────────────────────────────────
@pytest.fixture
def make_sentiment_client():
def _create(sentiment="neutral", score=0.5, explanation="Análisis de prueba", keywords=None):
if keywords is None:
keywords = []
client = MagicMock()
client.chat.completions.create.return_value = create_sentiment_response(
sentiment=sentiment, score=score, explanation=explanation, keywords=keywords
)
return client
return _create
@pytest.fixture
def make_error_client():
import openai
def _create(error_type="generic"):
client = MagicMock()
errors = {
"rate_limit": openai.RateLimitError("Rate limit", response=MagicMock(status_code=429), body={}),
"timeout": openai.APITimeoutError(request=MagicMock()),
"connection": openai.APIConnectionError(request=MagicMock()),
}
if error_type == "empty_response":
client.chat.completions.create.return_value = create_openai_chat_response("")
elif error_type in errors:
client.chat.completions.create.side_effect = errors[error_type]
else:
client.chat.completions.create.side_effect = Exception(f"Error: {error_type}")
return client
return _create
Paso 3: tests/integration/conftest.py
# tests/integration/conftest.py
import pytest
import os
@pytest.fixture(autouse=True)
def check_integration_available(request):
"""
Auto-fixture: verifica automáticamente si los integration tests pueden correr.
Solo actúa en tests marcados con @pytest.mark.integration.
"""
if request.node.get_closest_marker("integration"):
api_key = os.getenv("OPENAI_API_KEY")
run_integration = (
os.getenv("RUN_INTEGRATION", "false").lower() == "true" or
os.getenv("FULL_VALIDATION", "false").lower() == "true"
)
if not api_key or not run_integration:
pytest.skip(
"Integration tests deshabilitados. Para habilitar:\n"
" export OPENAI_API_KEY=sk-...\n"
" export RUN_INTEGRATION=true"
)
Paso 4: tests/integration/test_e2e.py
# tests/integration/test_e2e.py
import pytest
import os
from app.sentiment import analyze_sentiment
@pytest.mark.integration
@pytest.mark.e2e
class TestSentimentE2E:
"""
Tests E2E del analizador de sentimiento con LLM real.
Assertions son flexibles — verifican propiedades, no valores exactos.
"""
@pytest.mark.timeout(30)
def test_positive_sentiment_detection(self, integration_client, e2e_budget):
"""Texto claramente positivo → sentiment positivo."""
text = "Me encanta este producto, es absolutamente increíble y lo recomiendo."
result = analyze_sentiment(text, client=integration_client)
# Registrar costo estimado
e2e_budget.add_cost("gpt-4o-mini", prompt_tokens=100, completion_tokens=50)
# Assertions de propiedades
assert result["sentiment"] in ["positivo", "negativo", "neutral"]
assert 0 <= result["score"] <= 1
# Para texto tan positivo, el score debe ser alto
assert result["score"] >= 0.6, \
f"Para texto positivo claro, score debe ser >=0.6. Obtenido: {result['score']}"
@pytest.mark.timeout(30)
def test_negative_sentiment_detection(self, integration_client, e2e_budget):
"""Texto claramente negativo → sentiment negativo."""
text = "Terrible experiencia, el peor producto que he comprado. Horrible calidad."
result = analyze_sentiment(text, client=integration_client)
e2e_budget.add_cost("gpt-4o-mini", prompt_tokens=100, completion_tokens=50)
assert result["sentiment"] == "negativo"
assert result["score"] <= 0.4
@pytest.mark.timeout(30)
def test_neutral_factual_text(self, integration_client, e2e_budget):
"""Texto factual sin carga emocional → sentiment neutral."""
text = "El producto llegó en caja. Tiene dimensiones de 30x20x10 cm."
result = analyze_sentiment(text, client=integration_client)
e2e_budget.add_cost("gpt-4o-mini", prompt_tokens=80, completion_tokens=50)
assert result["sentiment"] in ["neutral", "positivo"] # Puede ir a cualquiera
assert isinstance(result["keywords"], list)
@pytest.mark.timeout(30)
def test_full_pipeline_no_crash(self, integration_client, e2e_budget):
"""El pipeline completo no crashea para cualquier texto válido."""
texts = [
"texto normal",
"¿¡Qué bueno!",
"12345 números",
"text in english is also fine",
"Texto corto"
]
for text in texts:
result = analyze_sentiment(text, client=integration_client)
e2e_budget.add_cost("gpt-4o-mini", prompt_tokens=60, completion_tokens=40)
assert isinstance(result, dict), f"Pipeline no retornó dict para: '{text}'"
assert "sentiment" in result, f"Falta key 'sentiment' para: '{text}'"
@pytest.mark.timeout(15)
def test_short_text_handled(self, integration_client, e2e_budget):
"""Textos muy cortos se manejan sin crash."""
result = analyze_sentiment("Bien", client=integration_client)
e2e_budget.add_cost("gpt-4o-mini", prompt_tokens=50, completion_tokens=40)
assert result["sentiment"] in ["positivo", "negativo", "neutral"]
Paso 5: tests/integration/test_semantic.py
# tests/integration/test_semantic.py
import pytest
from app.sentiment import analyze_sentiment
from tests.semantic import assert_semantically_similar
@pytest.mark.integration
class TestSentimentSemanticQuality:
"""
Tests de calidad semántica usando embeddings para comparar significado.
Usan sentence-transformers local (sin costo adicional de API).
"""
def test_positive_explanation_is_positive(self, integration_client, e2e_budget):
"""
La explicación para texto positivo debe ser semánticamente positiva.
"""
result = analyze_sentiment(
"Excelente calidad, superó todas mis expectativas",
client=integration_client
)
e2e_budget.add_cost("gpt-4o-mini", 100, 60)
# Verificar que la explicación es semánticamente positiva
assert_semantically_similar(
actual=result.get("explanation", ""),
expected="El texto expresa satisfacción y valoración positiva",
threshold=0.65, # Flexible — muchas formas de expresar positividad
client=None # Embeddings locales para ahorrar costo
)
def test_negative_explanation_is_negative(self, integration_client, e2e_budget):
"""
La explicación para texto negativo debe ser semánticamente negativa.
"""
result = analyze_sentiment(
"Terrible, decepcionante, no lo recomendaría a nadie",
client=integration_client
)
e2e_budget.add_cost("gpt-4o-mini", 100, 60)
assert_semantically_similar(
actual=result.get("explanation", ""),
expected="El texto usa lenguaje negativo y expresa insatisfacción",
threshold=0.60,
client=None
)
def test_explanations_for_opposite_sentiments_are_different(
self, integration_client, e2e_budget
):
"""
Las explicaciones para sentimientos opuestos deben ser semánticamente diferentes.
"""
from tests.semantic import get_embedding_local, cosine_similarity
result_pos = analyze_sentiment("Increíble, maravilloso, perfecto", client=integration_client)
result_neg = analyze_sentiment("Terrible, horrible, decepcionante", client=integration_client)
e2e_budget.add_cost("gpt-4o-mini", 200, 120)
if result_pos.get("explanation") and result_neg.get("explanation"):
emb_pos = get_embedding_local(result_pos["explanation"])
emb_neg = get_embedding_local(result_neg["explanation"])
similarity = cosine_similarity(emb_pos, emb_neg)
assert similarity < 0.70, (
f"Explicaciones de sentimientos opuestos son demasiado similares: {similarity:.3f}\n"
f"Positivo: {result_pos['explanation']}\n"
f"Negativo: {result_neg['explanation']}"
)
Paso 6: tests/integration/test_property.py
# tests/integration/test_property.py
import pytest
import json
from hypothesis import given, settings
import hypothesis.strategies as st
from unittest.mock import MagicMock
from tests.helpers import create_openai_chat_response
# Estrategia para outputs válidos del LLM
valid_sentiment_output = st.fixed_dictionaries({
"sentiment": st.sampled_from(["positivo", "negativo", "neutral"]),
"score": st.floats(min_value=0.0, max_value=1.0, allow_nan=False, allow_infinity=False),
"explanation": st.text(max_size=200),
"keywords": st.lists(
st.text(min_size=1, max_size=30, alphabet=st.characters(
whitelist_categories=("L", "N", "Zs")
)),
max_size=5
)
})
@given(llm_output=valid_sentiment_output)
@settings(max_examples=50)
def test_pipeline_contract_for_any_valid_llm_output(llm_output):
"""
PROPIEDAD: Para cualquier output válido del LLM (mockeado),
el pipeline produce un resultado que cumple el contrato.
"""
from app.sentiment import analyze_sentiment
mock_client = MagicMock()
mock_client.chat.completions.create.return_value = create_openai_chat_response(
json.dumps(llm_output, ensure_ascii=False)
)
result = analyze_sentiment("texto de prueba", client=mock_client)
assert result["sentiment"] in ["positivo", "negativo", "neutral"]
assert 0.0 <= result["score"] <= 1.0
assert isinstance(result["keywords"], list)
assert isinstance(result["explanation"], str)
@given(raw_score=st.floats(allow_nan=False, allow_infinity=False))
def test_processor_score_always_clamped(raw_score):
"""
PROPIEDAD: Para cualquier score input, el processor siempre retorna [0, 1].
"""
from app.processors import process_sentiment_output
result = process_sentiment_output({"sentiment": "neutral", "score": raw_score})
assert 0.0 <= result["score"] <= 1.0, \
f"Score {raw_score} → {result['score']} — no está en [0, 1]"
@given(sentiment=st.text(max_size=100))
def test_processor_sentiment_always_valid(sentiment):
"""
PROPIEDAD: Para cualquier sentiment input, el resultado siempre es uno de los tres válidos.
"""
from app.processors import process_sentiment_output
result = process_sentiment_output({"sentiment": sentiment, "score": 0.5})
assert result["sentiment"] in ["positivo", "negativo", "neutral"]
@given(raw_input=st.one_of(
st.just(""),
st.just(" "),
st.just("\n\n"),
))
def test_parser_empty_always_raises(raw_input):
"""
PROPIEDAD: Cualquier input vacío/whitespace siempre lanza ValueError.
"""
from app.parsers import parse_json_response
with pytest.raises(ValueError):
parse_json_response(raw_input)
@given(
content=st.fixed_dictionaries({
"x": st.integers(min_value=-1000, max_value=1000),
"y": st.text(min_size=0, max_size=100)
})
)
def test_parser_roundtrip(content):
"""
PROPIEDAD: json.dumps → parse_json_response → mismo resultado.
"""
from app.parsers import parse_json_response
raw = json.dumps(content)
result = parse_json_response(raw)
assert result["x"] == content["x"]
assert result["y"] == content["y"]
Paso 7: Actualizar pytest.ini
[pytest]
testpaths = tests
addopts = -v --tb=short
markers =
unit: Tests unitarios (determinísticos, sin API calls)
integration: Tests con LLM real (requieren OPENAI_API_KEY y RUN_INTEGRATION=true)
e2e: Tests end-to-end del flujo completo
contract: Tests de contrato de prompts
property: Tests property-based con Hypothesis
semantic: Tests con semantic similarity assertions
flaky: Tests con retry habilitado por varianza del LLM
quarantine: Tests inestables — no corren en CI regular
smoke: Tests de humo básicos
# Por defecto: solo unit tests (sin integration, sin quarantine)
# Para integration: pytest -m integration
# Para todo menos quarantine: pytest -m "not quarantine"
Paso 8: CI/CD con GitHub Actions
# .github/workflows/tests.yml
name: Tests
on:
push:
branches: [main, develop]
pull_request:
branches: [main]
schedule:
- cron: "0 2 * * *" # Nightly a las 2am UTC
jobs:
unit-tests:
name: Unit Tests (always)
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- uses: actions/setup-python@v4
with:
python-version: "3.11"
- run: pip install -r requirements.txt
- run: pytest -m "not integration and not quarantine" -v --tb=short
# Sin API key, sin costo, determinístico
integration-tests:
name: Integration Tests (main only)
runs-on: ubuntu-latest
if: github.ref == 'refs/heads/main'
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
RUN_INTEGRATION: "true"
E2E_BUDGET_USD: "0.25"
steps:
- uses: actions/checkout@v3
- uses: actions/setup-python@v4
with:
python-version: "3.11"
- run: pip install -r requirements.txt
- run: pytest -m "integration and not quarantine" --timeout=60 -v --tb=short
nightly-full:
name: Full Test Suite (nightly)
runs-on: ubuntu-latest
if: github.event_name == 'schedule'
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
RUN_INTEGRATION: "true"
FULL_VALIDATION: "true"
steps:
- uses: actions/checkout@v3
- uses: actions/setup-python@v4
with:
python-version: "3.11"
- run: pip install -r requirements.txt
- run: pytest -v --timeout=120 # Incluye quarantine para monitoreo
Paso 9: Verificación final
# 1. Verificar que los unit tests pasan (sin API key)
pytest -m "not integration" -v --tb=short
# Esperado: 50+ passed, 0 failed, 0 errors
# 2. Con API key: verificar integration tests
export OPENAI_API_KEY=sk-...
export RUN_INTEGRATION=true
pytest -m integration -v --timeout=60 --tb=short
# Esperado: los tests se ejecutan (no se skipean), algunos pueden tomar 5-10s
# 3. Property-based tests
pytest -m property -v --tb=short
# Esperado: tests corren con múltiples ejemplos de Hypothesis
# 4. Cobertura completa
pytest -m "not integration" --cov=app --cov-report=term-missing
# Esperado: >80% de cobertura en parsers y processors
# 5. Timing: unit tests deben ser rápidos
time pytest -m "not integration" -q
# Esperado: <15 segundos
Checklist de entrega
Tests escritos
- Al menos 4 tests E2E con LLM real en
test_e2e.py - Al menos 2 tests con semantic similarity en
test_semantic.py - Al menos 4 property-based tests en
test_property.py(con mock) - Tests de flaky management: al menos 1 test con
@pytest.mark.flaky - Tests de propiedades de parsers y processors
Calidad
- Budget tracker implementado y funcionando
- Skip automático cuando no hay API key
- Assertions flexibles en integration tests (no igualdad exacta)
- Property-based tests con propiedades claras y específicas
- Al menos 1 test en quarantine con razón documentada
Infraestructura
-
pytest.inicon todos los markers registrados - CI/CD configurado: unit en cada push, integration solo en main
- Documentación en README de cómo correr cada tipo
Ejercicios adicionales
Ejercicio 1: Agregar un segundo prompt
Añade un prompt classify_document(text) que retorna {category, confidence, tags}. Escribe:
- Contract test (unit, con mock)
- E2E test (integration, con LLM real)
- Semantic test para la categoría
Ver guía
# Contract test (M2 style):
def test_classify_contract(make_classification_client):
client = make_classification_client(category="tecnología", confidence=0.9)
result = classify_document("Artículo sobre Python", client=client)
assert result["category"] in VALID_CATEGORIES
assert 0 <= result["confidence"] <= 1
# E2E test (M3 style):
@pytest.mark.integration
def test_classify_e2e(integration_client, e2e_budget):
result = classify_document("Python es un lenguaje de programación", client=integration_client)
e2e_budget.add_cost("gpt-4o-mini", 100, 50)
assert result["category"] in VALID_CATEGORIES
# Semantic test:
@pytest.mark.integration
def test_classify_semantic_category(integration_client, e2e_budget):
result = classify_document("Python es un lenguaje de programación", client=integration_client)
e2e_budget.add_cost("gpt-4o-mini", 100, 60)
assert_semantically_similar(
actual=result["category"],
expected="tecnología o programación",
threshold=0.65
)
Ejercicio 2: Test de model drift
Diseña un test que detecte si el modelo produce resultados muy diferentes a un conjunto de referencia guardado:
Ver guía
# tests/integration/test_drift.py
REFERENCE_CASES = [
{"input": "Me encanta", "expected_sentiment": "positivo"},
{"input": "Lo odio", "expected_sentiment": "negativo"},
{"input": "El miércoles llegó", "expected_sentiment": "neutral"},
]
@pytest.mark.integration
def test_model_drift(integration_client, e2e_budget):
"""Detecta si el modelo produce sentimientos diferentes a los de referencia."""
mismatches = []
for case in REFERENCE_CASES:
result = analyze_sentiment(case["input"], client=integration_client)
e2e_budget.add_cost("gpt-4o-mini", 60, 40)
if result["sentiment"] != case["expected_sentiment"]:
mismatches.append({
"input": case["input"],
"expected": case["expected_sentiment"],
"got": result["sentiment"]
})
assert len(mismatches) == 0, (
f"Model drift detectado: {len(mismatches)}/{len(REFERENCE_CASES)} casos cambiaron.\n"
+ "\n".join(f" {m['input']}: {m['expected']} → {m['got']}" for m in mismatches)
)
Resumen del proyecto
Al completar este proyecto tienes:
| Tipo | Tests | Estrategia | Ejecución |
|---|---|---|---|
| Unit (M2) | 50+ | Mock, determinístico | Siempre |
| E2E | 5+ | LLM real, assertions flexibles | main + nightly |
| Semantic | 3+ | sentence-transformers, threshold calibrado | main + nightly |
| Property | 5+ | Hypothesis, con mock | Siempre |
| Total | 65+ | Mix de estrategias | Dependiendo del tipo |
Costo por run de integration: <$0.02 con gpt-4o-mini Tiempo total de unit tests: <15 segundos Tiempo total con integration: 1-3 minutos
Recursos adicionales
- pytest-rerunfailures — Retry para flaky tests
- sentence-transformers — Embeddings locales para semantic assertions
- Hypothesis — Property-based testing
- GitHub Actions — CI/CD configuration
- OpenAI Usage — Monitorear costos reales de la API
- Módulo 4: Guardrails — Próximo paso