Testing y validation

Tu cliente tiene cinco features grandes: adapters, factory, fallback, routing, métricas. Sin tests, una refactorización rompe algo y no te enteras hasta que un usuario reporta el bug. Esta cápsula te enseña a escribir la suite que te deja modificar con confianza.

Al terminar vas a poder:

• Estructurar tests separando unit (con mocks) de integración (contra API real)
• Mockear adapters para testear UnifiedClient sin pagar tokens
• Verificar fallback simulando fallos de providers
• Validar contratos (Pydantic) de inputs/outputs
• Correr la suite con pytest y entender la salida

Modelo mental: tres niveles de test

La mayoría de tu suite son unit tests con mocks. Integration tests son pocos y selectivos (no pueden correr en CI sin API keys).

Setup de pytest

Crea tests/conftest.py:

# tests/conftest.py
"""Fixtures compartidos para tests."""
import pytest
from unittest.mock import MagicMock

from unified_ai_client.models import ProviderConfig, ChatResponse, Message
from unified_ai_client.adapters.base import BaseAdapter


@pytest.fixture
def provider_config_basico():
    return ProviderConfig(
        name="test-provider",
        type="openai",
        model="test-model",
        api_key_env="TEST_KEY",
        price_input_per_1m=0.10,
        price_output_per_1m=0.20,
    )


class FakeAdapter(BaseAdapter):
    """Adapter falso que devuelve una respuesta predecible. Útil en tests."""

    def __init__(self, config: ProviderConfig, respuesta_texto: str = "OK",
                 lanza: Exception | None = None):
        super().__init__(config)
        self.respuesta_texto = respuesta_texto
        self.lanza = lanza
        self.llamadas = 0  # contador para verificar uso

    def chat(self, messages, max_tokens=256, temperature=0.7) -> ChatResponse:
        self.llamadas += 1
        if self.lanza:
            raise self.lanza
        return ChatResponse(
            text=self.respuesta_texto,
            model=self.config.model or "test",
            provider=self.name,
            tokens_input=10,
            tokens_output=20,
            duration_ms=100,
            cost_usd=0.000005,
        )


@pytest.fixture
def fake_adapter_factory():
    """Factory de FakeAdapter para usar en tests."""
    def crear(name: str, respuesta: str = "OK", lanza: Exception | None = None):
        config = ProviderConfig(name=name, type="openai", model="test", api_key_env="X")
        return FakeAdapter(config, respuesta_texto=respuesta, lanza=lanza)
    return crear

Test 1 — Models (validación Pydantic)

tests/test_models.py:

# tests/test_models.py
import pytest
from pydantic import ValidationError

from unified_ai_client.models import Message, ChatResponse, ProviderConfig


def test_message_acepta_roles_validos():
    Message(role="system", content="hola")
    Message(role="user", content="hola")
    Message(role="assistant", content="hola")


def test_message_rechaza_rol_invalido():
    with pytest.raises(ValidationError):
        Message(role="invalid", content="hola")


def test_chat_response_defaults():
    r = ChatResponse(text="hola", model="m", provider="p")
    assert r.tokens_input == 0
    assert r.cost_usd is None


def test_provider_config_acepta_tipos_validos():
    ProviderConfig(name="x", type="openai")
    ProviderConfig(name="x", type="openai_compatible")
    ProviderConfig(name="x", type="modal_custom")


def test_provider_config_rechaza_tipo_invalido():
    with pytest.raises(ValidationError):
        ProviderConfig(name="x", type="nonexistent")

Test 2 — Factory (mapeo de tipo → adapter)

tests/test_factory.py:

# tests/test_factory.py
import pytest
from unified_ai_client.factory import ProviderFactory
from unified_ai_client.models import ProviderConfig
from unified_ai_client.adapters.openai_adapter import OpenAIAdapter
from unified_ai_client.exceptions import ConfigError


def test_factory_crea_openai_adapter(monkeypatch):
    monkeypatch.setenv("FAKE_KEY", "x")
    config = ProviderConfig(name="x", type="openai", model="gpt-4o-mini", api_key_env="FAKE_KEY")
    adapter = ProviderFactory.crear(config)
    assert isinstance(adapter, OpenAIAdapter)
    assert adapter.name == "x"


def test_factory_lanza_para_tipo_desconocido():
    config = ProviderConfig(name="x", type="modal_custom")
    with pytest.raises(ConfigError):
        ProviderFactory.crear(config)


def test_factory_permite_registrar_custom(monkeypatch):
    from unified_ai_client.adapters.base import BaseAdapter

    class MiAdapter(BaseAdapter):
        def chat(self, messages, max_tokens=256, temperature=0.7):
            ...

    ProviderFactory.registrar("mi_tipo", MiAdapter)
    config = ProviderConfig(name="x", type="openai")  # ojo: el Literal acepta tipos, así que para tests del registro real ajusta
    # Acá el test demostraría que el registro funciona si el tipo está permitido

Test 3 — UnifiedClient con FakeAdapter

tests/test_client.py:

# tests/test_client.py
import pytest
from unittest.mock import patch

from unified_ai_client import UnifiedClient
from unified_ai_client.models import ClientConfig, ProviderConfig
from unified_ai_client.exceptions import (
    AllProvidersFailedError,
    RateLimitError,
    AuthError,
    TimeoutError,
    ConfigError,
)
from tests.conftest import FakeAdapter


def crear_client_con_fakes(primary_fake, fallback_fakes=None):
    """Helper: crea un UnifiedClient real pero con FakeAdapters inyectados."""
    fallback_fakes = fallback_fakes or []
    config = ClientConfig(
        primary=primary_fake.name,
        fallback=[f.name for f in fallback_fakes],
        providers={
            primary_fake.name: primary_fake.config,
            **{f.name: f.config for f in fallback_fakes},
        },
    )
    # Crea cliente y reemplaza los adapters reales por los fakes
    with patch("unified_ai_client.factory.ProviderFactory.crear",
               side_effect=lambda cfg: next(
                   a for a in [primary_fake, *fallback_fakes] if a.name == cfg.name
               )):
        client = UnifiedClient(config)
    return client


def test_chat_devuelve_respuesta_del_primary(fake_adapter_factory):
    primary = fake_adapter_factory("primary", respuesta="del primary")
    client = crear_client_con_fakes(primary)
    r = client.chat("hola")
    assert r.text == "del primary"
    assert r.provider == "primary"
    assert primary.llamadas == 1


def test_fallback_cuando_primary_falla(fake_adapter_factory):
    primary = fake_adapter_factory(
        "primary",
        lanza=TimeoutError("primary", "timeout"),
    )
    fb = fake_adapter_factory("fallback", respuesta="rescate")
    client = crear_client_con_fakes(primary, [fb])
    r = client.chat("hola")
    assert r.text == "rescate"
    assert r.provider == "fallback"
    assert primary.llamadas == 1
    assert fb.llamadas == 1


def test_levanta_error_si_todos_fallan(fake_adapter_factory):
    p = fake_adapter_factory("p", lanza=TimeoutError("p", "t"))
    fb = fake_adapter_factory("fb", lanza=AuthError("fb", "bad key"))
    client = crear_client_con_fakes(p, [fb])

    with pytest.raises(AllProvidersFailedError) as exc:
        client.chat("hola")
    # Debe contener detalle de ambos errores
    assert "p" in str(exc.value)
    assert "fb" in str(exc.value)


def test_circuit_breaker_se_abre_tras_n_fallos(fake_adapter_factory):
    primary = fake_adapter_factory("primary", lanza=TimeoutError("primary", "t"))
    fb = fake_adapter_factory("fb", respuesta="ok")
    client = crear_client_con_fakes(primary, [fb])

    # Hacer 3 requests; cada una hace fallar primary y va al fallback
    for _ in range(3):
        client.chat("hola")
    assert primary.llamadas == 3

    # Después de 3 fallos consecutivos, circuit abre. Próxima request no toca primary.
    primary.llamadas = 0
    client.chat("hola")
    assert primary.llamadas == 0  # circuit abierto, primary saltado
    assert fb.llamadas == 4


def test_use_fallback_false_no_intenta_otros(fake_adapter_factory):
    p = fake_adapter_factory("p", lanza=TimeoutError("p", "t"))
    fb = fake_adapter_factory("fb", respuesta="ok")
    client = crear_client_con_fakes(p, [fb])

    with pytest.raises(AllProvidersFailedError):
        client.chat("hola", use_fallback=False)

    assert p.llamadas == 1
    assert fb.llamadas == 0


def test_metrics_registran_exitos_y_errores(fake_adapter_factory):
    p = fake_adapter_factory("primary", lanza=TimeoutError("primary", "t"))
    fb = fake_adapter_factory("fb", respuesta="ok")
    client = crear_client_con_fakes(p, [fb])

    client.chat("uno")
    client.chat("dos")

    metrics = client.get_metrics()
    assert metrics.total_requests == 4  # 2 errores primary + 2 exitos fb
    assert metrics.total_exitos == 2
    assert metrics.total_errores == 2
    assert metrics.por_provider["primary"]["errores"] == 2
    assert metrics.por_provider["fb"]["exitos"] == 2


def test_routing_cost_first(fake_adapter_factory):
    caro = fake_adapter_factory("caro", respuesta="caro")
    caro.config.cost_tier = "high"

    barato = fake_adapter_factory("barato", respuesta="barato")
    barato.config.cost_tier = "low"

    client = crear_client_con_fakes(caro, [barato])
    r = client.chat("hola", priority="cost-first")
    # cost-first debería ir primero por barato (low) antes que caro (high)
    assert r.provider == "barato"

Test 4 — Contract testing del OpenAIAdapter

Cuando golpeas el API real (integration test), verifica que la respuesta cumple tu contrato:

tests/test_integration_openai.py:

# tests/test_integration_openai.py
"""Tests que golpean el API real. Solo correr con OPENAI_API_KEY seteada."""
import os
import pytest

from unified_ai_client import UnifiedClient
from unified_ai_client.models import ChatResponse

skip_si_no_hay_key = pytest.mark.skipif(
    not os.environ.get("OPENAI_API_KEY"),
    reason="OPENAI_API_KEY no seteada"
)


@skip_si_no_hay_key
def test_openai_real_devuelve_chat_response_valido():
    CONFIG = {
        "primary": "openai-mini",
        "providers": {
            "openai-mini": {
                "name": "openai-mini",
                "type": "openai",
                "model": "gpt-4o-mini",
                "api_key_env": "OPENAI_API_KEY",
                "price_input_per_1m": 0.15,
                "price_output_per_1m": 0.60,
            }
        },
    }
    client = UnifiedClient.from_dict(CONFIG)
    r = client.chat("Di 'hola' y nada más", max_tokens=10)

    # Contract assertions
    assert isinstance(r, ChatResponse)
    assert len(r.text) > 0
    assert r.provider == "openai-mini"
    assert r.model == "gpt-4o-mini"
    assert r.tokens_input > 0
    assert r.tokens_output > 0
    assert r.duration_ms > 0
    assert r.cost_usd is not None and r.cost_usd > 0
    assert r.raw_response is not None

Correr los tests

# Todos los tests
pytest tests/ -v

# Solo unit tests (sin tocar APIs reales)
pytest tests/ -v -k "not integration"

# Solo integration tests
pytest tests/test_integration_openai.py -v

# Con cobertura
pip install pytest-cov
pytest tests/ --cov=unified_ai_client --cov-report=term-missing

Output esperado de unit tests:

tests/test_models.py::test_message_acepta_roles_validos PASSED
tests/test_models.py::test_message_rechaza_rol_invalido PASSED
tests/test_models.py::test_chat_response_defaults PASSED
tests/test_models.py::test_provider_config_acepta_tipos_validos PASSED
tests/test_models.py::test_provider_config_rechaza_tipo_invalido PASSED
tests/test_factory.py::test_factory_crea_openai_adapter PASSED
tests/test_factory.py::test_factory_lanza_para_tipo_desconocido PASSED
tests/test_client.py::test_chat_devuelve_respuesta_del_primary PASSED
tests/test_client.py::test_fallback_cuando_primary_falla PASSED
tests/test_client.py::test_levanta_error_si_todos_fallan PASSED
tests/test_client.py::test_circuit_breaker_se_abre_tras_n_fallos PASSED
tests/test_client.py::test_use_fallback_false_no_intenta_otros PASSED
tests/test_client.py::test_metrics_registran_exitos_y_errores PASSED
tests/test_client.py::test_routing_cost_first PASSED

14 passed in 0.42s

Bajo medio segundo, perfecto para CI.

CI: GitHub Actions

.github/workflows/test.yml:

name: Tests

on: [push, pull_request]

jobs:
  unit-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install -e ".[dev]"
      - run: pytest tests/ -v -k "not integration"

  integration-tests:
    runs-on: ubuntu-latest
    needs: unit-tests
    if: github.event_name == 'push' && github.ref == 'refs/heads/main'
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install -e ".[dev]"
      - env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: pytest tests/test_integration_openai.py -v

Unit tests corren en cada PR (rápidos, sin API keys). Integration tests solo en main (con secrets de GitHub).

Patrones avanzados

Pattern 1 — Property-based testing con Hypothesis

Para escapar de "probé los casos que se me ocurrieron":

from hypothesis import given, strategies as st

@given(st.text(min_size=1, max_size=1000))
def test_chat_acepta_cualquier_prompt(prompt):
    client = crear_client_con_fakes(fake_adapter("p", respuesta="ok"))
    r = client.chat(prompt)
    assert r.text == "ok"

Pattern 2 — Snapshot testing

Si te preocupa que cambios accidentales modifiquen el formato de output:

import json

def test_metrics_format_estable(snapshot, ...):
    metrics_json = client.metrics.to_json()
    snapshot.assert_match(metrics_json, "metrics.json")

Si el formato cambia, el test falla y debes confirmar el cambio.

Pattern 3 — Test de regresión por bug fix

Cada vez que arreglas un bug, escribe un test que falle sin el fix. Eso garantiza que ese bug nunca vuelve.

Trampas comunes

Trampa 1 — "Tests que dependen de orden." Si un test resetea estado compartido y el siguiente lo asume reseteado, tu suite es frágil. Usa fixtures con scope="function" (default) para aislar.

Trampa 2 — "Mockeo cosas internas y mi test pasa pero el real falla." Mockea bordes (HTTP calls, system clock), no internos de tu propia librería. Si mockeas OpenAIAdapter.chat, tu test prueba "que UnifiedClient llama a chat", no "que la integración con OpenAI funciona".

Trampa 3 — "Sin tests de integración." Unit tests con mocks no detectan: cambios de breaking change en el SDK de OpenAI, errores de auth, problemas de formato de response real. Necesitas algunos integration tests corriendo periódicamente.

Trampa 4 — "Tests de integración en cada CI run." $0.10 por test × 50 tests × 30 PRs/mes = $150/mes solo en CI. Filtra integration tests a main o a un cron nightly.

Trampa 5 — "Suite tarda 5 minutos." Sospecha de mocks faltantes. Una suite unit decente para una librería de este tamaño debería correr en <2 segundos.

Ejercicio

Escribe tests para:

1. Que routing quality-first ranquee correctamente cuando hay 3 providers con cost_tier distinto
2. Que reset_metrics efectivamente limpia el collector y el siguiente summary muestra 0 requests
3. Que un AuthError en primary no detiene fallback (el cliente debería seguir intentando OpenRouter)
4. Que un prompt vacío falle con ValidationError antes de llegar a un adapter

def test_routing_quality_first_va_por_caro_primero(fake_adapter_factory):
    barato = fake_adapter_factory("barato", lanza=TimeoutError("b", "t"))
    barato.config.cost_tier = "low"

    medio = fake_adapter_factory("medio", lanza=TimeoutError("m", "t"))
    medio.config.cost_tier = "medium"

    caro = fake_adapter_factory("caro", respuesta="caro_ok")
    caro.config.cost_tier = "high"

    # Primary cualquiera; el orden lo cambia priority
    client = crear_client_con_fakes(medio, [barato, caro])
    r = client.chat("hola", priority="quality-first")
    assert r.provider == "caro"


def test_reset_metrics(fake_adapter_factory):
    p = fake_adapter_factory("p", respuesta="x")
    client = crear_client_con_fakes(p)
    client.chat("uno")
    assert client.get_metrics().total_requests == 1

    client.reset_metrics()
    assert client.get_metrics().total_requests == 0


def test_auth_error_en_primary_sigue_al_fallback(fake_adapter_factory):
    p = fake_adapter_factory("p", lanza=AuthError("p", "bad key"))
    fb = fake_adapter_factory("fb", respuesta="rescate")
    client = crear_client_con_fakes(p, [fb])
    r = client.chat("hola")
    assert r.provider == "fb"


def test_prompt_vacio_falla_validation():
    # Test que UnifiedClient.chat rechaza prompt vacío antes de llamar al adapter
    # Requiere agregar validación en chat(): if not prompt: raise ValueError
    p = fake_adapter("p", respuesta="x")
    client = crear_client_con_fakes(p)
    with pytest.raises((ValueError, ValidationError)):
        client.chat("")

Resumen

Aprendiste:

• ✅ Tres niveles de test: unit (mocks), integration (API real), contract (validación de shape)
• ✅ FakeAdapter reusable para testear UnifiedClient sin pagar tokens
• ✅ Cómo verificar fallback, circuit breaker, routing y métricas con mocks deterministicos
• ✅ Skip de integration tests cuando no hay API key
• ✅ CI con GitHub Actions: unit en cada PR, integration solo en main
• ✅ Patterns: property-based, snapshot, test de regresión

Checkpoint: si pytest tests/ -v pasa todos los tests verdes y pytest --cov muestra >80% de cobertura, estás listo.

Siguiente cápsula

08 — Proyecto final consolida todo. Versión completa del paquete con todos los providers (incluyendo Modal custom), documentación final, ejemplos de uso real, y publicación opcional a PyPI. Es el deliverable que llevas a tu portfolio.

Recursos

1. pytest documentation — la fuente de verdad.
2. pytest-mock — fixture mocker más limpio que unittest.mock.
3. Hypothesis — property-based testing.
4. VCR.py — graba HTTP responses para integration tests deterministicos.
5. Coverage.py — medir qué % de tu código cubren los tests.