6. Flaky Test Management

Descripción

Un test flaky pasa a veces y falla otras sin cambiar el código. En apps AI, la varianza del LLM es la causa más común. Esta cápsula cubre las cuatro estrategias para manejar flakiness: retry logic, tolerance thresholds, quarantine, y relax de assertions. Más importante: cómo diferenciar entre flakiness por varianza legítima del LLM (que se maneja) y flakiness por un test mal escrito o un bug real (que se arregla). Normalizar la flakiness destruye la confianza en la test suite.

El problema con el flaky testing normalizado

Semana 1: Test falla → "Es el LLM, es normal" → ignorado
Semana 2: Test falla → "Otra vez el LLM" → ignorado
Semana 3: Bug real → el test falla → "Seguro es el LLM" → ignorado
Semana 4: El bug llega a producción

La normalización de la flakiness mata el valor de la test suite. Cuando los developers aprenden a ignorar tests que fallan, todos los fallos se ignoran — incluyendo los que indican bugs reales.

Las cuatro causas de flakiness en AI apps

Causa 1: Varianza del LLM (legítima)

# El LLM puede responder de maneras ligeramente diferentes:
result_1 = analyze_sentiment("Texto positivo")
# → "positivo" con score 0.92

result_2 = analyze_sentiment("Texto positivo")
# → "positivo" con score 0.88

# Si el test hace: assert result["score"] == 0.92 → falla 50% del tiempo
# Solución: assertion sobre rango, no valor exacto

Causa 2: Test mal escrito (arreglar, no tolerar)

# Assertion demasiado estricta para output no-determinístico:
def test_bad():
    result = analyze_sentiment("Me encanta")
    assert result["explanation"] == "El texto usa lenguaje positivo."
    # Falla porque el LLM puede decir "El texto expresa positividad" o
    # "El texto tiene carga emocional positiva" — todos son correctos
    # → ARREGLAR: usar assertion semántica o de propiedades

Causa 3: Infraestructura (arreglar)

# Timeout de API, rate limit, connectivity:
def test_api_timeout():
    result = analyze_sentiment("texto")  # Puede fallar por timeout
    assert result["sentiment"] == "positivo"
    # → ARREGLAR: añadir timeout explícito, retry con backoff, skip condicional

Causa 4: Estado compartido entre tests (arreglar)

# Caché global que afecta resultados:
_cache = {}

def analyze_with_cache(text):
    if text not in _cache:
        _cache[text] = analyze_sentiment(text)
    return _cache[text]

def test_a():
    result = analyze_with_cache("texto x")
    # Modifica _cache

def test_b():
    # _cache["texto x"] puede tener el resultado del test_a
    result = analyze_with_cache("texto x")
    # → ARREGLAR: limpiar estado compartido entre tests, o usar fresh state

Estrategia 1: Retry logic

Para flakiness por varianza del LLM o problemas transitorios de API:

pip install pytest-rerunfailures

# Opción A: por test individual
import pytest

@pytest.mark.integration
@pytest.mark.flaky(reruns=3, reruns_delay=2)
def test_sentiment_quality_with_retry():
    """
    Test de calidad semántica — puede fallar por varianza del LLM.
    Reintenta hasta 3 veces con 2s de delay.
    """
    result = analyze_sentiment("Me encanta este producto")
    
    # Esta assertion puede variar — por eso tiene retry
    assert result["score"] >= 0.7, \
        f"Para texto positivo claro, score debe ser >= 0.7. Obtenido: {result['score']}"

# Opción B: configuración global en pytest.ini
# [pytest]
# addopts = --reruns 2 --reruns-delay 1
# Solo afecta tests marcados con @pytest.mark.flaky

# Opción C: solo para integration tests
# pytest.ini:
# [pytest]
# addopts = -m integration --reruns 2 --reruns-delay 1

Retry con backoff exponencial

Para APIs con rate limits, el delay fijo puede no ser suficiente:

import time
import openai
from functools import wraps

def retry_with_backoff(max_retries=3, base_delay=1.0):
    """Decorator para retry con backoff exponencial."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except openai.RateLimitError:
                    if attempt == max_retries - 1:
                        raise
                    delay = base_delay * (2 ** attempt)
                    time.sleep(delay)
                except openai.APITimeoutError:
                    if attempt == max_retries - 1:
                        raise
                    time.sleep(base_delay)
        return wrapper
    return decorator

@retry_with_backoff(max_retries=3, base_delay=1.0)
def analyze_sentiment_with_retry(text, client):
    return analyze_sentiment(text, client=client)

# Test que usa la función con retry:
@pytest.mark.integration
def test_with_api_retry(integration_client):
    result = analyze_sentiment_with_retry("texto", integration_client)
    assert result["sentiment"] in ["positivo", "negativo", "neutral"]

Estrategia 2: Tolerance thresholds

Para tests que validan propiedades estadísticas — esperan pasar la mayoría, no necesariamente todas:

def run_with_tolerance(func, n_runs: int = 5, min_pass: int = 4):
    """
    Ejecuta func n_runs veces y verifica que al menos min_pass pasan.
    
    Args:
        func: La función a ejecutar (retorna True si pasa, False si falla)
        n_runs: Total de ejecuciones
        min_pass: Mínimo de ejecuciones que deben pasar
    """
    results = []
    for i in range(n_runs):
        try:
            passed = func()
            results.append(passed if passed is not None else True)
        except AssertionError:
            results.append(False)
    
    n_passed = sum(results)
    assert n_passed >= min_pass, (
        f"Solo {n_passed}/{n_runs} runs pasaron (mínimo requerido: {min_pass})\n"
        f"Resultados: {results}"
    )
    return n_passed

# Ejemplo de uso:
@pytest.mark.integration
def test_sentiment_quality_with_tolerance(integration_client):
    """
    IMPORTANTE: Este test usa tolerance porque el score puede variar.
    Acepta que 4 de 5 runs produzcan score >= 0.7 para texto positivo claro.
    """
    def run_once():
        result = analyze_sentiment(
            "Me encanta absolutamente todo, es increíble",
            client=integration_client
        )
        return result["score"] >= 0.7
    
    # Acepta hasta 1 fallo en 5 runs
    run_with_tolerance(run_once, n_runs=5, min_pass=4)

# Versión con assertion directa (más simple):
@pytest.mark.integration
def test_sentiment_passes_majority():
    """4 de 5 análisis del mismo texto deben dar 'positivo'."""
    text = "Perfecto, me encanta, es maravilloso"
    scores = []
    
    for _ in range(5):
        result = analyze_sentiment(text)
        scores.append(result["sentiment"] == "positivo")
    
    n_positivo = sum(scores)
    assert n_positivo >= 4, f"Solo {n_positivo}/5 clasificaron como positivo"

Estrategia 3: Quarantine

Para tests inestables que no puedes arreglar inmediatamente:

# pytest.ini — registrar el marker
[pytest]
markers =
    unit: Tests unitarios, determinísticos, sin API
    integration: Tests con LLM real
    flaky: Tests que pueden fallar por varianza — con retry habilitado
    quarantine: Tests inestables en cuarentena — no corren en CI regular
    e2e: Tests end-to-end del flujo completo
    smoke: Tests de humo básicos

# En el test:
@pytest.mark.quarantine(reason="Falla 20% del tiempo por varianza del score")
@pytest.mark.integration
def test_exact_score_value():
    """
    CUARENTENA: Este test verifica un score exacto que varía.
    Debe ser arreglado para usar assertion de rango en vez de valor exacto.
    Ver issue: #123
    """
    result = analyze_sentiment("texto positivo")
    assert result["score"] == 0.92  # ← Demasiado estricto

# .github/workflows/ci.yml
# CI regular: excluir quarantine
- name: Unit tests
  run: pytest -m "not integration and not quarantine" -v

# Nightly: incluir quarantine para monitorear
- name: Full test suite including quarantine
  run: pytest -v  # Sin filtro
  if: github.event_name == 'schedule'  # Solo en nightly

Workflow de quarantine

Test falla repetidamente
       ↓
Análisis: ¿Por qué falla?
  ├── Assertion demasiado estricta → Relajar assertion (arreglar)
  ├── Timing/infraestructura → Añadir retry o timeout (arreglar)
  ├── Estado compartido → Aislar tests (arreglar)
  └── Varianza legítima del LLM que no se puede evitar
       ↓
Añadir @pytest.mark.quarantine(reason="...", issue="URL")
       ↓
Crear issue para arreglar o eliminar el test
       ↓
En nightly: monitorear si mejora
       ↓
Cuando se arregla: remover quarantine

Estrategia 4: Relajar assertions

La solución más efectiva y más ignorada: arreglar la assertion, no el test:

# ─── ANTES (frágil) ───────────────────────────────────────────────
def test_sentiment_fragile():
    result = analyze_sentiment("Este producto es excelente")
    # ❌ Demasiado específico — el LLM puede dar 0.88, 0.91, 0.95, etc.
    assert result["score"] == 0.92
    # ❌ La explicación puede variar en cada run
    assert result["explanation"] == "El texto usa adjetivos positivos fuertes."
    # ❌ Keywords pueden variar en orden o selección
    assert result["keywords"] == ["excelente", "producto"]

# ─── DESPUÉS (robusto) ────────────────────────────────────────────
def test_sentiment_robust(make_sentiment_client):
    # Para test de contrato → usar mock (M2)
    client = make_sentiment_client(
        sentiment="positivo",
        score=0.92,
        keywords=["excelente"]
    )
    result = analyze_sentiment("Este producto es excelente", client=client)
    
    # ✅ Rango en vez de valor exacto
    assert 0.7 <= result["score"] <= 1.0
    # ✅ Propiedad en vez de texto exacto
    assert isinstance(result["explanation"], str) and len(result["explanation"]) > 0
    # ✅ Subset en vez de lista exacta
    assert any(kw in result["keywords"] for kw in ["excelente", "producto", "positivo"])

# ─── PARA TEST DE CALIDAD REAL (con LLM real) ────────────────────
@pytest.mark.integration
def test_sentiment_quality(integration_client):
    result = analyze_sentiment("Este producto es excelente", client=integration_client)
    
    # ✅ Solo assertions robustas para el LLM real
    assert result["sentiment"] == "positivo"  # Esto sí debe ser determinístico
    assert result["score"] >= 0.6  # Rango, no valor exacto
    assert len(result.get("explanation", "")) > 10  # Tiene algo de explicación

Detectar y monitorear flakiness

# Script para detectar tests flaky corriendo múltiples veces:
# run_flaky_detector.sh
#!/bin/bash
FAIL_COUNT=0
TOTAL_RUNS=10

for i in $(seq 1 $TOTAL_RUNS); do
    if ! pytest tests/integration/ -q --tb=no 2>/dev/null; then
        FAIL_COUNT=$((FAIL_COUNT + 1))
    fi
done

echo "Fallos: $FAIL_COUNT / $TOTAL_RUNS runs"
if [ $FAIL_COUNT -gt 2 ]; then
    echo "⚠️  Hay tests flaky (más del 20% de fallo)"
    exit 1
fi

# En pytest: usar pytest-repeat para detectar flakiness
# pip install pytest-repeat

# Ejecutar cada test 5 veces:
# pytest --count=5 tests/integration/ -v

# Si algún test falla en alguna de las 5 ejecuciones, se reporta como fallido

Cuándo aplicar cada estrategia

La regla de oro: nunca ignorar un fallo sin investigar

# ❌ Lo que hace un equipo que normalizó la flakiness:
@pytest.mark.skip(reason="Siempre falla, no sé por qué")
def test_important_behavior():
    ...

# ❌ Lo que hace un equipo resignado:
@pytest.mark.flaky(reruns=10)  # 10 reintentos porque nadie investigó
def test_something():
    ...

# ✅ Lo que hace un equipo que gestiona bien la flakiness:
@pytest.mark.quarantine(
    reason="Falla ~15% del tiempo. Score varía 0.65-0.85 para este input. "
           "Investigando si es varianza del modelo o bug en el prompt. "
           "Issue: github.com/repo/issues/456"
)
@pytest.mark.integration
def test_specific_score():
    """Test en cuarentena mientras se investiga el fallo intermitente."""
    result = analyze_sentiment("texto ambiguo")
    assert result["score"] > 0.75  # Falla ~15% del tiempo

Ejercicios

Ejercicio 1: Diagnosticar el tipo de flakiness

Para cada caso, identifica la causa y la estrategia:

1. Test que verifica result["explanation"] == "El texto es positivo" — falla 40% del tiempo
2. Test que falla cuando se ejecuta después de test_b pero pasa solo
3. Test que falla con APITimeoutError en picos de tráfico
4. Test que verifica result["sentiment"] == "positivo" — falla 5% del tiempo

Ejercicio 2: Implementar tolerance

Escribe un test que verifica que analyze_sentiment clasifica correctamente textos positivos "la mayoría del tiempo" (4 de 5 runs):

@pytest.mark.integration
@pytest.mark.skipif(not os.getenv("OPENAI_API_KEY"), reason="Requiere API key")
def test_positive_classification_with_tolerance(integration_client):
    """
    Para texto claramente positivo, el modelo debe clasificar como 'positivo'
    al menos 4 de 5 veces.
    """
    text = "Me encanta este producto, es increíble, lo recomiendo totalmente"
    classifications = []
    
    for _ in range(5):
        result = analyze_sentiment(text, client=integration_client)
        classifications.append(result["sentiment"])
    
    n_positive = classifications.count("positivo")
    
    assert n_positive >= 4, (
        f"Solo {n_positive}/5 clasificaciones fueron 'positivo'.\n"
        f"Clasificaciones: {classifications}"
    )

Ejercicio 3: Quarantine workflow completo

Un test en tu suite falla aproximadamente el 25% del tiempo. Describe el proceso completo que seguirías para gestionarlo:

Ejercicio 4: Arreglar este test

El siguiente test falla ~30% del tiempo. Arréglalo para que sea estable:

@pytest.mark.integration
def test_negative_sentiment():
    result = analyze_sentiment("Este producto es horrible, me arrepiento de comprarlo")
    assert result["sentiment"] == "negativo"
    assert result["score"] == 0.05
    assert result["explanation"] == "El texto expresa insatisfacción muy fuerte."
    assert result["keywords"] == ["horrible", "arrepiento"]

# Versión arreglada: assertions robustas
@pytest.mark.integration
@pytest.mark.skipif(not os.getenv("OPENAI_API_KEY"), reason="Requiere API key")
def test_negative_sentiment_robust(integration_client):
    """
    Test arreglado: usa assertions robustas para el LLM real.
    El sentimiento 'negativo' es determinístico para texto tan explícito.
    Score y redacción de explicación varían — no los verificamos exactamente.
    """
    result = analyze_sentiment(
        "Este producto es horrible, me arrepiento de comprarlo",
        client=integration_client
    )
    
    # ✅ Esto sí es determinístico para texto tan claro:
    assert result["sentiment"] == "negativo"
    
    # ✅ Rango en vez de valor exacto:
    assert result["score"] <= 0.3, \
        f"Para texto muy negativo, score debe ser bajo (<=0.3), es {result['score']}"
    
    # ✅ Propiedad en vez de texto exacto:
    assert isinstance(result["explanation"], str)
    assert len(result["explanation"]) > 5
    
    # ✅ Subset en vez de lista exacta:
    assert any(kw in result["keywords"] for kw in ["horrible", "arrepiento", "malo", "negativo"])

Ejercicio 5: Política de flakiness del equipo

Escribe una política de 5 puntos para tu equipo sobre cómo manejar tests flaky:

## Política de Flaky Tests

1. **Ningún test flaky en el branch main sin investigación previa.**
   Si un test falla, debe ser investigado antes de mergearlo.

2. **No usaremos `@pytest.mark.skip` sin razón documentada.**
   Un test skippeado sin razón es un test que no funciona. Siempre documentar por qué y cuándo se resolverá.

3. **Quarantine es temporal, no permanente.**
   Un test en quarantine tiene un issue asociado. Si no se arregla en 2 semanas, se elimina.

4. **Retry máximo de 3 intentos, con razón documentada.**
   `@pytest.mark.flaky(reruns=3)` requiere un comentario explicando por qué el test es inherentemente variable.

5. **Monitorear la tasa de fallo de integration tests en nightly.**
   Si >10% de tests fallan en nightly, priorizar arreglo antes de nuevas features.

Resumen

• Cuatro estrategias: retry, tolerance threshold, quarantine, relajar assertions
• Dos tipos de flakiness: por varianza del LLM (manejar) vs por test mal escrito/bug (arreglar)
• Nunca normalizar: "los tests de AI son flaky" es falso — se pueden gestionar
• Quarantine es temporal: siempre con issue asociado y deadline
• La solución más efectiva: relajar la assertion en vez de añadir retry
• Regla de oro: ningún fallo se ignora sin investigación

Recursos adicionales

1. pytest-rerunfailures — Plugin de retry
2. pytest-flaky — Alternativa para marcar tests flaky
3. Non-Determinism — Martin Fowler — Estrategias generales para flakiness
4. Dealing with flaky tests at Google — Cómo lo maneja Google
5. pytest-repeat — Para detectar flakiness ejecutando tests múltiples veces
6. Quarantine pattern — Patrón de cuarentena en detalle