4. Post-Deploy Validation — Verificar que Funciona de Verdad

Descripción

En esta cápsula vas a implementar validación post-deploy que va más allá de "la app responde 200 OK." Implementarás health checks que verifican dependencias, smoke tests que envían prompts reales y verifican que la inferencia funciona, y scripts de validación automatizados. Un endpoint que retorna 200 pero cuya inferencia está rota no es un deploy exitoso.

Contexto: La cápsula anterior automatizó el deployment. Pero un pipeline que despliega sin verificar es peligroso — puedes tener un sistema "en producción" que no funciona. Esta cápsula cierra ese gap. Al terminar, sabrás con certeza si tu deploy es exitoso o necesita rollback.

La Diferencia entre "Online" y "Funcionando"

El problema del health check básico

# Health check INSUFICIENTE
@app.get("/health")
async def health():
    return {"status": "ok"}

# Este endpoint retorna 200 incluso si:
# - La API key de OpenAI es inválida
# - El modelo configurado no existe
# - La base de datos vectorial no está accesible
# - El cache Redis está caído
# - El prompt template tiene un error de sintaxis

Un health check que solo verifica que el proceso FastAPI está corriendo es como un médico que solo verifica que el paciente respira. Necesitas tests más profundos.

Los tres niveles de validación

Nivel 1: LIVENESS — ¿El proceso está corriendo?
├── HTTP 200 en /health
├── Verifica: el container no crasheó
└── No verifica: nada funcional

Nivel 2: READINESS — ¿Puede recibir tráfico?
├── Dependencias conectadas (API keys válidas, DB accesible)
├── Verifica: el sistema está listo para procesar
└── No verifica: que la lógica funcione correctamente

Nivel 3: SMOKE TEST — ¿La inferencia funciona end-to-end?
├── Envía prompt real, recibe respuesta válida
├── Verifica: el flujo completo funciona
└── Es la validación definitiva

Nivel 1: Health Check con Dependencias

Health check que verifica el sistema completo

# src/health.py
import time
from fastapi import APIRouter
from src.config import get_settings

router = APIRouter()

async def check_openai_connection() -> dict:
    """Verifica que la API key de OpenAI es válida."""
    settings = get_settings()
    try:
        from openai import AsyncOpenAI
        client = AsyncOpenAI(api_key=settings.openai_api_key)
        models = await client.models.list()
        return {"status": "connected", "models_available": True}
    except Exception as e:
        return {"status": "error", "detail": str(e)[:100]}

async def check_vector_db() -> dict:
    """Verifica que la base de datos vectorial está accesible."""
    try:
        # Adapta a tu vector DB (ChromaDB, Pinecone, etc.)
        import chromadb
        client = chromadb.Client()
        collections = client.list_collections()
        return {"status": "connected", "collections": len(collections)}
    except Exception as e:
        return {"status": "error", "detail": str(e)[:100]}

@router.get("/health")
async def health_check():
    """Liveness check — ¿el proceso está corriendo?"""
    return {
        "status": "healthy",
        "timestamp": time.time(),
        "version": get_settings().version,
    }

@router.get("/health/ready")
async def readiness_check():
    """Readiness check — ¿las dependencias están listas?"""
    settings = get_settings()
    checks = {}

    checks["openai"] = await check_openai_connection()
    checks["config"] = {
        "status": "ok",
        "environment": settings.environment,
        "model": settings.openai_model,
    }

    all_healthy = all(
        c.get("status") in ("connected", "ok")
        for c in checks.values()
    )

    return {
        "status": "ready" if all_healthy else "degraded",
        "checks": checks,
        "timestamp": time.time(),
    }

Registrar en la app

# src/main.py
from fastapi import FastAPI
from src.health import router as health_router

app = FastAPI(title="AI System")
app.include_router(health_router)

Verificación local

# Liveness (debe ser rápido, <100ms)
$ curl http://localhost:8000/health
{"status":"healthy","timestamp":1741456800.0,"version":"1.0.0"}

# Readiness (puede tardar 1-2s por las verificaciones)
$ curl http://localhost:8000/health/ready
{
  "status": "ready",
  "checks": {
    "openai": {"status": "connected", "models_available": true},
    "config": {"status": "ok", "environment": "development", "model": "gpt-4o-mini"}
  },
  "timestamp": 1741456801.0
}

Nivel 2: Smoke Tests de Inferencia

Qué es un smoke test

Un smoke test envía un request real al sistema desplegado y verifica que la respuesta es correcta. No es un test unitario — es una verificación end-to-end contra producción.

Smoke test flow:
1. Enviar prompt conocido al endpoint de inferencia
2. Verificar que la respuesta tiene la estructura esperada
3. Verificar que la respuesta contiene contenido coherente
4. Medir latencia y verificar que está dentro del target
5. PASS o FAIL

Script de smoke test

# scripts/smoke_test.py
"""
Smoke tests para validar deployment.
Ejecutar después de cada deploy:
    python scripts/smoke_test.py https://tu-app.platform.app
"""
import sys
import time
import json
import urllib.request
import urllib.error

class SmokeTestRunner:
    def __init__(self, base_url: str):
        self.base_url = base_url.rstrip("/")
        self.results = []

    def run_test(self, name: str, method: str, path: str,
                 body: dict = None, expected_status: int = 200,
                 validate_body: callable = None,
                 max_latency_ms: int = 5000) -> bool:
        """Ejecuta un smoke test individual."""
        url = f"{self.base_url}{path}"
        start = time.time()

        try:
            data = json.dumps(body).encode() if body else None
            headers = {"Content-Type": "application/json"} if body else {}
            req = urllib.request.Request(url, data=data, headers=headers, method=method)
            response = urllib.request.urlopen(req, timeout=30)
            status = response.status
            response_body = json.loads(response.read().decode())

        except urllib.error.HTTPError as e:
            status = e.code
            response_body = {"error": str(e)}
        except Exception as e:
            status = 0
            response_body = {"error": str(e)}

        latency_ms = (time.time() - start) * 1000

        passed = True
        errors = []

        if status != expected_status:
            passed = False
            errors.append(f"Expected status {expected_status}, got {status}")

        if latency_ms > max_latency_ms:
            passed = False
            errors.append(f"Latency {latency_ms:.0f}ms > {max_latency_ms}ms target")

        if validate_body and status == expected_status:
            try:
                body_valid = validate_body(response_body)
                if not body_valid:
                    passed = False
                    errors.append("Body validation failed")
            except Exception as e:
                passed = False
                errors.append(f"Body validation error: {e}")

        result = {
            "name": name,
            "passed": passed,
            "status": status,
            "latency_ms": round(latency_ms),
            "errors": errors,
        }
        self.results.append(result)

        icon = "PASS" if passed else "FAIL"
        print(f"  [{icon}] {name} — {status} — {latency_ms:.0f}ms")
        if errors:
            for e in errors:
                print(f"         {e}")

        return passed

    def report(self) -> bool:
        """Imprime resumen y retorna True si todos pasaron."""
        total = len(self.results)
        passed = sum(1 for r in self.results if r["passed"])
        failed = total - passed

        print(f"\nResults: {passed}/{total} passed, {failed} failed")

        if failed > 0:
            print("\nFailed tests:")
            for r in self.results:
                if not r["passed"]:
                    print(f"  - {r['name']}: {', '.join(r['errors'])}")

        return failed == 0


def run_smoke_tests(base_url: str) -> bool:
    """Ejecuta todos los smoke tests."""
    runner = SmokeTestRunner(base_url)

    print(f"Running smoke tests against: {base_url}\n")

    # Test 1: Liveness
    runner.run_test(
        name="Liveness check",
        method="GET",
        path="/health",
        max_latency_ms=1000,
        validate_body=lambda b: b.get("status") == "healthy",
    )

    # Test 2: Readiness
    runner.run_test(
        name="Readiness check",
        method="GET",
        path="/health/ready",
        max_latency_ms=3000,
        validate_body=lambda b: b.get("status") in ("ready", "degraded"),
    )

    # Test 3: Inference - respuesta básica
    runner.run_test(
        name="Inference - basic prompt",
        method="POST",
        path="/api/inference",
        body={"prompt": "Respond with exactly one word: hello"},
        max_latency_ms=10000,
        validate_body=lambda b: "response" in b and len(b["response"]) > 0,
    )

    # Test 4: Inference - estructura de respuesta
    runner.run_test(
        name="Inference - response structure",
        method="POST",
        path="/api/inference",
        body={"prompt": "What is 2+2?"},
        max_latency_ms=10000,
        validate_body=lambda b: all(k in b for k in ["response", "model"]),
    )

    # Test 5: Error handling - prompt vacío
    runner.run_test(
        name="Error handling - empty prompt",
        method="POST",
        path="/api/inference",
        body={"prompt": ""},
        expected_status=422,
        max_latency_ms=1000,
    )

    return runner.report()


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python scripts/smoke_test.py <base_url>")
        print("Example: python scripts/smoke_test.py https://mi-app.railway.app")
        sys.exit(1)

    url = sys.argv[1]
    success = run_smoke_tests(url)
    sys.exit(0 if success else 1)

Ejecución

# Local
$ python scripts/smoke_test.py http://localhost:8000

Running smoke tests against: http://localhost:8000

  [PASS] Liveness check — 200 — 12ms
  [PASS] Readiness check — 200 — 1245ms
  [PASS] Inference - basic prompt — 200 — 2340ms
  [PASS] Inference - response structure — 200 — 1890ms
  [PASS] Error handling - empty prompt — 422 — 8ms

Results: 5/5 passed, 0 failed

# Producción
$ python scripts/smoke_test.py https://mi-app.railway.app

Running smoke tests against: https://mi-app.railway.app

  [PASS] Liveness check — 200 — 89ms
  [PASS] Readiness check — 200 — 1567ms
  [PASS] Inference - basic prompt — 200 — 3210ms
  [PASS] Inference - response structure — 200 — 2890ms
  [PASS] Error handling - empty prompt — 422 — 45ms

Results: 5/5 passed, 0 failed

Integración en el Pipeline de CI/CD

Smoke tests como job de validación

# En .github/workflows/deploy.yml
validate:
  needs: deploy
  runs-on: ubuntu-latest
  timeout-minutes: 5
  steps:
    - uses: actions/checkout@v4

    - name: Set up Python
      uses: actions/setup-python@v5
      with:
        python-version: "3.11"

    - name: Wait for deployment
      run: sleep 60

    - name: Run smoke tests
      run: python scripts/smoke_test.py "${{ vars.PRODUCTION_URL }}"

    - name: Upload results
      if: always()
      run: echo "Smoke test completed at $(date -u)"

Script de validación bash (alternativa ligera)

#!/bin/bash
# scripts/validate-deploy.sh
# Uso: ./scripts/validate-deploy.sh https://mi-app.railway.app

set -e

BASE_URL="${1:?Usage: $0 <base_url>}"
PASSED=0
FAILED=0

check() {
    local name="$1"
    local expected_status="$2"
    local method="$3"
    local path="$4"
    local body="$5"

    if [ -n "$body" ]; then
        response=$(curl -s -w "\n%{http_code}\n%{time_total}" \
            -X "$method" "$BASE_URL$path" \
            -H "Content-Type: application/json" \
            -d "$body")
    else
        response=$(curl -s -w "\n%{http_code}\n%{time_total}" \
            -X "$method" "$BASE_URL$path")
    fi

    http_code=$(echo "$response" | tail -2 | head -1)
    latency=$(echo "$response" | tail -1)
    body_resp=$(echo "$response" | head -n -2)

    latency_ms=$(echo "$latency * 1000" | bc | cut -d. -f1)

    if [ "$http_code" = "$expected_status" ]; then
        echo "  [PASS] $name — $http_code — ${latency_ms}ms"
        PASSED=$((PASSED + 1))
    else
        echo "  [FAIL] $name — expected $expected_status, got $http_code — ${latency_ms}ms"
        echo "         Response: $body_resp"
        FAILED=$((FAILED + 1))
    fi
}

echo "Validating deployment: $BASE_URL"
echo ""

check "Health check" "200" "GET" "/health"
check "Readiness check" "200" "GET" "/health/ready"
check "Inference" "200" "POST" "/api/inference" \
    '{"prompt": "Say OK"}'
check "Error handling" "422" "POST" "/api/inference" \
    '{"prompt": ""}'

echo ""
echo "Results: $PASSED passed, $FAILED failed"

if [ "$FAILED" -gt 0 ]; then
    echo "DEPLOYMENT VALIDATION FAILED"
    exit 1
fi

echo "DEPLOYMENT VALIDATED SUCCESSFULLY"

chmod +x scripts/validate-deploy.sh
./scripts/validate-deploy.sh https://mi-app.railway.app

Validación Continua (Scheduled)

Health checks periódicos con GitHub Actions

# .github/workflows/health-monitor.yml
name: Health Monitor

on:
  schedule:
    - cron: '*/30 * * * *'  # Cada 30 minutos
  workflow_dispatch:

jobs:
  health-check:
    runs-on: ubuntu-latest
    timeout-minutes: 2
    steps:
      - name: Check production health
        run: |
          status=$(curl -s -o /dev/null -w "%{http_code}" \
            "${{ vars.PRODUCTION_URL }}/health")
          if [ "$status" != "200" ]; then
            echo "ALERT: Production health check failed: $status"
            exit 1
          fi
          echo "Production healthy at $(date -u)"

Monitoreo externo gratuito

Configuración recomendada en UptimeRobot:
├── Monitor 1: GET /health (cada 5 min)
├── Monitor 2: GET /health/ready (cada 15 min)
├── Alert: Email + Slack cuando falle 2 veces consecutivas
└── Status page: público para stakeholders (opcional)

Troubleshooting

Problema 1: "El smoke test de inferencia siempre falla por timeout"

Causa: El primer request después de un deploy tiene cold start largo (la plataforma está iniciando el container).

Solución:

# Warm-up request antes de los smoke tests
def warmup(base_url: str, retries: int = 5, delay: int = 10):
    """Envía requests de warm-up hasta que el servicio responda."""
    for i in range(retries):
        try:
            req = urllib.request.Request(f"{base_url}/health")
            response = urllib.request.urlopen(req, timeout=15)
            if response.status == 200:
                print(f"  Service ready after {i+1} attempts")
                return True
        except Exception:
            pass
        print(f"  Warming up... attempt {i+1}/{retries}")
        time.sleep(delay)
    return False

# En run_smoke_tests:
if not warmup(base_url):
    print("Service not available after warm-up")
    return False

Problema 2: "Health check pasa pero readiness muestra OpenAI como error"

Causa: La API key no está configurada en la plataforma de producción, o está vencida.

Solución:

# Verificar la key en producción
curl -s https://api.openai.com/v1/models \
    -H "Authorization: Bearer $OPENAI_API_KEY" | python -m json.tool | head -5

# Si da error 401: la key es inválida
# Si da error de conexión: la plataforma bloquea outbound requests

# Verificar en la plataforma
# Railway: railway variables | grep OPENAI
# Fly.io: flyctl secrets list

Problema 3: "Los smoke tests pasan la primera vez pero fallan intermitentemente"

Causa: Rate limiting de la API de OpenAI, cold starts en free tier, o memoria insuficiente.

Solución:

# Agregar retry logic al smoke test runner
def run_test_with_retry(self, max_retries=2, **kwargs):
    for attempt in range(max_retries + 1):
        if self.run_test(**kwargs):
            return True
        if attempt < max_retries:
            print(f"         Retrying in 5s...")
            time.sleep(5)
    return False

Problema 4: "El script de validación funciona local pero falla en CI"

Causa: Network differences entre GitHub Actions runners y tu máquina local.

Solución:

# Dar más tiempo en CI
- name: Wait for deployment
  run: sleep 90  # 90 segundos en vez de 45

# Aumentar timeouts
- name: Run smoke tests
  run: python scripts/smoke_test.py "${{ vars.PRODUCTION_URL }}"
  timeout-minutes: 5

Ejercicios Prácticos

Ejercicio 1: Health check con dependencias

Implementa el health check de tres niveles (liveness, readiness) en tu app FastAPI.

# src/health.py
import time
from fastapi import APIRouter

router = APIRouter(tags=["health"])

@router.get("/health")
async def liveness():
    return {"status": "healthy", "timestamp": time.time()}

@router.get("/health/ready")
async def readiness():
    from src.config import get_settings
    settings = get_settings()

    checks = {}

    # Check OpenAI
    try:
        from openai import AsyncOpenAI
        client = AsyncOpenAI(api_key=settings.openai_api_key)
        await client.models.list()
        checks["openai"] = {"status": "connected"}
    except Exception as e:
        checks["openai"] = {"status": "error", "detail": str(e)[:80]}

    # Check config
    errors = settings.validate_for_production()
    checks["config"] = {
        "status": "ok" if not errors else "error",
        "environment": settings.environment,
        "errors": errors,
    }

    all_ok = all(c["status"] in ("connected", "ok") for c in checks.values())
    return {
        "status": "ready" if all_ok else "degraded",
        "checks": checks,
    }

curl -s http://localhost:8000/health | python -m json.tool
curl -s http://localhost:8000/health/ready | python -m json.tool

Ejercicio 2: Smoke test script completo

Crea el archivo scripts/smoke_test.py con al menos 4 tests (liveness, readiness, inference, error handling).

# Test local
python scripts/smoke_test.py http://localhost:8000

# Test producción
python scripts/smoke_test.py https://tu-app.railway.app

Ejercicio 3: Script de validación bash

Crea scripts/validate-deploy.sh como alternativa ligera al smoke test Python.

# Crear el script
touch scripts/validate-deploy.sh
chmod +x scripts/validate-deploy.sh

# Copiar el contenido del script bash de la cápsula

# Ejecutar local
./scripts/validate-deploy.sh http://localhost:8000

# Ejecutar contra producción
./scripts/validate-deploy.sh https://tu-app.railway.app

Ejercicio 4: Configura monitoreo externo

Registra tu URL de producción en UptimeRobot (o alternativa) y configura alertas.

Resumen

• Un health check que solo retorna 200 no es suficiente — necesita verificar dependencias
• Tres niveles: liveness (¿vive?), readiness (¿listo?), smoke test (¿funciona la inferencia?)
• Los smoke tests envían prompts reales y verifican respuestas — son la validación definitiva
• El script de smoke test debe ser ejecutable tanto local como contra producción
• Integra en CI/CD: el pipeline debe fallar si los smoke tests fallan
• Monitoreo continuo: usa UptimeRobot o similar para verificación periódica gratuita
• Un deploy sin validación post-deploy es un deploy a ciegas — no sabes si funciona hasta que un usuario se queja

Recursos Adicionales

1. Kubernetes Health Checks — Liveness vs Readiness — Conceptos de health checks (aplica fuera de K8s)
2. UptimeRobot — Monitoreo gratuito de uptime
3. Better Stack Uptime — Monitoreo con status pages
4. Testing in Production — Charity Majors — Por qué y cómo testear en producción
5. Smoke Testing — Martin Fowler — Definición y mejores prácticas
6. Health Check Patterns — Microsoft — Patrones de health check monitoring