Módulo 2: Unit Testing LLM Applications

6. Fixture Factories

Descripción

Las fixture factories son funciones que retornan funciones generadoras de fixtures. En vez de una fixture estática que siempre devuelve lo mismo, tienes una factory que puede crear variaciones parametrizadas. Esta técnica es fundamental para testear apps LLM porque permite generar docenas de variaciones de responses sin duplicar código. El resultado: tests más expresivos, DRY, y que cubren más edge cases.

El problema con las fixtures estáticas

Las fixtures estáticas tienen un problema: solo definen un caso:

# Fixture estática — solo un caso
@pytest.fixture
def mock_client_positivo():
    client = MagicMock()
    client.chat.completions.create.return_value = create_openai_chat_response(
        '{"sentiment": "positivo", "score": 0.9, "keywords": ["bien"]}'
    )
    return client

@pytest.fixture
def mock_client_negativo():
    client = MagicMock()
    client.chat.completions.create.return_value = create_openai_chat_response(
        '{"sentiment": "negativo", "score": 0.1, "keywords": ["mal"]}'
    )
    return client

@pytest.fixture
def mock_client_neutral():
    client = MagicMock()
    client.chat.completions.create.return_value = create_openai_chat_response(
        '{"sentiment": "neutral", "score": 0.5, "keywords": ["normal"]}'
    )
    return client

# Para testear 3 sentimientos, necesitas 3 fixtures y 3 tests separados
def test_positivo(mock_client_positivo): ...
def test_negativo(mock_client_negativo): ...
def test_neutral(mock_client_neutral): ...

# Si añades otro sentimiento, creas otra fixture...
# Si necesitas 10 variaciones, escribes 10 fixtures...
# ❌ Este patrón no escala

La solución: factory fixture

Una factory fixture retorna una función que genera el objeto deseado:

# Factory fixture — un solo lugar para crear variaciones
@pytest.fixture
def mock_client_factory():
    """
    Factory que crea mock clients con diferentes responses.
    
    Uso:
        def test_x(mock_client_factory):
            client = mock_client_factory(sentiment="positivo", score=0.9)
            result = analyze_sentiment("texto", client=client)
            assert result["sentiment"] == "positivo"
    """
    def _create(
        sentiment: str = "neutral",
        score: float = 0.5,
        explanation: str = "Análisis de prueba",
        keywords: list = None
    ) -> MagicMock:
        if keywords is None:
            keywords = []
        
        content = json.dumps({
            "sentiment": sentiment,
            "score": score,
            "explanation": explanation,
            "keywords": keywords
        })
        
        client = MagicMock()
        client.chat.completions.create.return_value = create_openai_chat_response(content)
        return client
    
    return _create

# Ahora un solo fixture crea CUALQUIER variación:
def test_positive_sentiment(mock_client_factory):
    client = mock_client_factory(sentiment="positivo", score=0.9)
    result = analyze_sentiment("texto positivo", client=client)
    assert result["sentiment"] == "positivo"

def test_negative_sentiment(mock_client_factory):
    client = mock_client_factory(sentiment="negativo", score=0.1)
    result = analyze_sentiment("texto negativo", client=client)
    assert result["sentiment"] == "negativo"

def test_boundary_score(mock_client_factory):
    client = mock_client_factory(sentiment="positivo", score=1.0)
    result = analyze_sentiment("texto extremo", client=client)
    assert result["score"] == 1.0

Factory con defaults inteligentes

Los defaults bien elegidos hacen la factory más expresiva:

# tests/conftest.py

from tests.helpers import create_openai_chat_response
import json
from unittest.mock import MagicMock
import pytest

# Valores por default realistas
DEFAULT_SENTIMENT_RESPONSE = {
    "sentiment": "neutral",
    "score": 0.5,
    "explanation": "El texto no expresa sentimiento claro.",
    "keywords": []
}

@pytest.fixture
def make_sentiment_client():
    """
    Factory para crear mock clients de sentimiento.
    
    Defaults: neutral, score 0.5 — el caso más "vacío".
    Sobrescribe solo lo que necesitas testear.
    """
    def _create(**kwargs) -> MagicMock:
        # Mezclar defaults con los kwargs proporcionados
        response_data = {**DEFAULT_SENTIMENT_RESPONSE, **kwargs}
        
        client = MagicMock()
        client.chat.completions.create.return_value = create_openai_chat_response(
            json.dumps(response_data)
        )
        return client
    
    return _create

# Uso expresivo — solo especificas lo que cambia:
def test_sentiment(make_sentiment_client):
    # Solo necesito cambiar el sentiment
    client = make_sentiment_client(sentiment="positivo", score=0.95)
    result = analyze_sentiment("Me encanta", client=client)
    assert result["sentiment"] == "positivo"

def test_low_confidence(make_sentiment_client):
    # Solo me importa el score bajo
    client = make_sentiment_client(score=0.1)
    result = analyze_sentiment("Texto ambiguo", client=client)
    assert result["score"] == 0.1

Factory para edge cases

Las factories brillan cuando necesitas cubrir muchos edge cases:

@pytest.fixture
def make_error_client():
    """
    Factory para crear clients que simulan errores del LLM.
    
    Uso:
        client = make_error_client("rate_limit")
        client = make_error_client("timeout")
        client = make_error_client("empty_response")
    """
    def _create(error_type: str = "generic") -> MagicMock:
        client = MagicMock()
        
        error_map = {
            "rate_limit": openai.RateLimitError(
                message="Rate limit exceeded",
                response=MagicMock(status_code=429),
                body={}
            ),
            "timeout": openai.APITimeoutError(request=MagicMock()),
            "connection": openai.APIConnectionError(request=MagicMock()),
            "auth": openai.AuthenticationError(
                message="Invalid API key",
                response=MagicMock(status_code=401),
                body={}
            ),
            "service_unavailable": openai.APIStatusError(
                message="Service unavailable",
                response=MagicMock(status_code=503),
                body={}
            ),
            "empty_response": None,  # Caso especial: retorna None
            "generic": Exception("Error inesperado del LLM")
        }
        
        if error_type == "empty_response":
            client.chat.completions.create.return_value = create_openai_chat_response("")
        else:
            error = error_map.get(error_type, error_map["generic"])
            client.chat.completions.create.side_effect = error
        
        return client
    
    return _create

# Tests de error handling:
@pytest.mark.parametrize("error_type,expected_result_key", [
    ("rate_limit", "error"),
    ("timeout", "error"),
    ("empty_response", "sentiment"),  # Debe usar fallback, no crashear
])
def test_error_handling(make_error_client, error_type, expected_result_key):
    """La app maneja todos los tipos de error gracefully."""
    client = make_error_client(error_type)
    result = analyze_sentiment("texto", client=client)
    
    assert result is not None, "La app no debe retornar None"
    assert expected_result_key in result, \
        f"Se esperaba '{expected_result_key}' en el resultado, se obtuvo: {result}"

Factory + parametrize: la combinación más poderosa

Cuando combinas factories con parametrize, obtienes una cobertura masiva con poco código:

# Define los casos de test como datos
SENTIMENT_TEST_CASES = [
    pytest.param(
        {"sentiment": "positivo", "score": 0.95, "keywords": ["excelente", "increíble"]},
        "positivo",
        id="muy_positivo"
    ),
    pytest.param(
        {"sentiment": "positivo", "score": 0.6},
        "positivo",
        id="levemente_positivo"
    ),
    pytest.param(
        {"sentiment": "neutral", "score": 0.5},
        "neutral",
        id="neutral_exacto"
    ),
    pytest.param(
        {"sentiment": "negativo", "score": 0.3},
        "negativo",
        id="levemente_negativo"
    ),
    pytest.param(
        {"sentiment": "negativo", "score": 0.05, "keywords": ["terrible", "horrible"]},
        "negativo",
        id="muy_negativo"
    ),
]

@pytest.mark.parametrize("response_data,expected_sentiment", SENTIMENT_TEST_CASES)
def test_sentiment_variations(make_sentiment_client, response_data, expected_sentiment):
    """
    Testea múltiples variaciones de sentimiento con una sola función.
    La factory crea el cliente con los datos apropiados para cada caso.
    """
    client = make_sentiment_client(**response_data)
    result = analyze_sentiment("texto de prueba", client=client)
    
    assert result["sentiment"] == expected_sentiment
    assert 0 <= result["score"] <= 1
    assert isinstance(result.get("keywords", []), list)

Factory para mock de múltiples llamadas (chains)

Las factories son especialmente útiles para chains que hacen múltiples llamadas al LLM:

@pytest.fixture
def make_chain_client():
    """
    Factory para crear clients que responden a múltiples llamadas LLM.
    
    Uso para chains:
        client = make_chain_client([
            '{"summary": "Resumen del texto"}',           # Primera llamada
            '{"category": "tecnología", "tags": ["AI"]}'   # Segunda llamada
        ])
    """
    def _create(responses: list[str]) -> MagicMock:
        """
        responses: lista de JSON strings, una por cada llamada al LLM.
        """
        client = MagicMock()
        client.chat.completions.create.side_effect = [
            create_openai_chat_response(resp)
            for resp in responses
        ]
        return client
    
    return _create

# Test de chain con múltiples llamadas:
def test_summarize_and_classify_chain(make_chain_client):
    """Chain que resume y luego clasifica."""
    client = make_chain_client([
        '{"summary": "Python es popular para AI"}',              # Primera llamada: resumen
        '{"category": "tecnología", "confidence": 0.95}'         # Segunda llamada: clasificación
    ])
    
    result = summarize_and_classify("Artículo largo sobre Python y AI...", client=client)
    
    assert result["summary"] == "Python es popular para AI"
    assert result["category"] == "tecnología"
    assert client.chat.completions.create.call_count == 2

# Test que verifica el comportamiento cuando la segunda llamada falla:
def test_chain_second_call_fails(make_chain_client, make_error_client):
    """Chain maneja fallo en la segunda llamada."""
    client = MagicMock()
    client.chat.completions.create.side_effect = [
        create_openai_chat_response('{"summary": "Resumen"}'),  # Primera: éxito
        openai.APITimeoutError(request=MagicMock())               # Segunda: timeout
    ]
    
    result = summarize_and_classify("texto", client=client)
    
    # Debe retornar el resumen aunque la clasificación falle
    assert result["summary"] == "Resumen"
    assert result.get("category") is None or "error" in result

Factory con async support

Para apps async, la factory debe retornar AsyncMock:

@pytest.fixture
def make_async_client():
    """Factory para mock clients async."""
    def _create(
        sentiment: str = "neutral",
        score: float = 0.5,
        **kwargs
    ) -> MagicMock:
        content = json.dumps({"sentiment": sentiment, "score": score, **kwargs})
        
        client = MagicMock()
        # AsyncMock para la función async
        client.chat.completions.create = AsyncMock(
            return_value=create_openai_chat_response(content)
        )
        return client
    
    return _create

@pytest.mark.asyncio
async def test_async_analyze(make_async_client):
    client = make_async_client(sentiment="positivo", score=0.9)
    result = await analyze_sentiment_async("texto", client=client)
    assert result["sentiment"] == "positivo"

Factory que combina client + patch

Para casos donde necesitas parchear el cliente global de la app:

@pytest.fixture
def patched_openai_factory(mocker):
    """
    Factory que parchea el cliente OpenAI global y permite configurar la respuesta.
    
    Uso:
        def test_x(patched_openai_factory):
            mock_create = patched_openai_factory(sentiment="positivo")
            result = analyze_sentiment("texto")  # Sin pasar client
            assert result["sentiment"] == "positivo"
            mock_create.assert_called_once()  # Verifica que se llamó al LLM
    """
    def _create(**response_kwargs) -> MagicMock:
        mock_create = mocker.patch("app.sentiment.client.chat.completions.create")
        
        content = json.dumps({
            "sentiment": response_kwargs.get("sentiment", "neutral"),
            "score": response_kwargs.get("score", 0.5),
            "explanation": response_kwargs.get("explanation", "Análisis de prueba"),
            "keywords": response_kwargs.get("keywords", [])
        })
        
        mock_create.return_value = create_openai_chat_response(content)
        return mock_create  # Retorna el mock para que puedas hacer assertions
    
    return _create

# Uso:
def test_with_patched_factory(patched_openai_factory):
    mock_create = patched_openai_factory(sentiment="positivo", score=0.92)
    
    # No necesitas pasar client — el patch actúa sobre el cliente global
    result = analyze_sentiment("Me encanta")
    
    assert result["sentiment"] == "positivo"
    mock_create.assert_called_once()
    
    # Verificar parámetros de la llamada
    call_kwargs = mock_create.call_args.kwargs
    assert "Me encanta" in str(call_kwargs.get("messages", []))

Organización de factories en conftest.py

Para un proyecto con múltiples módulos, organiza las factories:

# tests/conftest.py — factories compartidas por todos los tests

@pytest.fixture
def make_sentiment_client():
    """Factory para tests de análisis de sentimiento."""
    def _create(sentiment="neutral", score=0.5, **extra):
        content = json.dumps({"sentiment": sentiment, "score": score, **extra})
        client = MagicMock()
        client.chat.completions.create.return_value = create_openai_chat_response(content)
        return client
    return _create

@pytest.fixture
def make_summary_client():
    """Factory para tests de resumen de texto."""
    def _create(summary="Resumen de prueba", confidence=0.8, sources=None):
        if sources is None:
            sources = []
        content = json.dumps({"summary": summary, "confidence": confidence, "sources": sources})
        client = MagicMock()
        client.chat.completions.create.return_value = create_openai_chat_response(content)
        return client
    return _create

@pytest.fixture
def make_classification_client():
    """Factory para tests de clasificación."""
    def _create(category="general", confidence=0.8, tags=None):
        if tags is None:
            tags = []
        content = json.dumps({"category": category, "confidence": confidence, "tags": tags})
        client = MagicMock()
        client.chat.completions.create.return_value = create_openai_chat_response(content)
        return client
    return _create

# tests/unit/conftest.py — factories específicas para unit tests
@pytest.fixture
def make_chain_client():
    """Factory para tests de chains con múltiples llamadas."""
    def _create(responses: list[str]) -> MagicMock:
        client = MagicMock()
        client.chat.completions.create.side_effect = [
            create_openai_chat_response(r) for r in responses
        ]
        return client
    return _create

Anti-patrones en fixture factories

# ❌ Anti-patrón 1: Factory que hace demasiado
@pytest.fixture
def uber_factory():
    def _create(sentiment=None, summary=None, classification=None, error=None, ...):
        # 50 líneas de lógica compleja
        # Difícil de entender qué crea exactamente
        ...
    return _create
# ✅ Mejor: factories pequeñas y específicas por tipo de prompt

# ❌ Anti-patrón 2: Factory con estado mutable compartido
SHARED_STATE = {"call_count": 0}

@pytest.fixture
def factory_with_shared_state():
    def _create():
        SHARED_STATE["call_count"] += 1  # Estado mutable global
        ...
    return _create
# ✅ Mejor: cada llamada a la factory crea un objeto fresco sin estado compartido

# ❌ Anti-patrón 3: Factory que mezcla sync y async sin declararlo
@pytest.fixture
def ambiguous_factory():
    def _create(async_mode=False):
        if async_mode:
            return AsyncMock(...)
        else:
            return MagicMock(...)
    return _create
# ✅ Mejor: factories separadas para sync y async

# ❌ Anti-patrón 4: No usar factory cuando hay 5+ fixtures iguales con variaciones
@pytest.fixture
def client_v1():
    client = MagicMock()
    client.chat.completions.create.return_value = create_openai_chat_response('{"x": 1}')
    return client

@pytest.fixture
def client_v2():
    client = MagicMock()
    client.chat.completions.create.return_value = create_openai_chat_response('{"x": 2}')
    return client

# ... client_v3, client_v4, client_v5 ...
# ✅ Mejor: make_client(x=1), make_client(x=2)

Cuándo usar cada patrón

Situación	Patrón recomendado
Un solo caso estándar	Fixture estática
2-3 casos predefinidos conocidos	`@pytest.fixture(params=[...])`
Variaciones dinámicas en cada test	Factory fixture
Muchas combinaciones	Factory + `@pytest.mark.parametrize`
Tests de error handling	Factory especializada para errores
Chains con múltiples LLM calls	Factory con `side_effect` lista

Ejercicios

Ejercicio 1: Crear una factory básica

Crea una factory fixture make_classification_client que genere clients con respuestas de clasificación. La respuesta tiene: category (string), confidence (float), tags (lista).

Ver solución

@pytest.fixture
def make_classification_client():
    """Factory para tests de clasificación de documentos."""
    def _create(
        category: str = "general",
        confidence: float = 0.8,
        tags: list = None
    ) -> MagicMock:
        if tags is None:
            tags = []
        
        content = json.dumps({
            "category": category,
            "confidence": confidence,
            "tags": tags
        })
        
        client = MagicMock()
        client.chat.completions.create.return_value = create_openai_chat_response(content)
        return client
    
    return _create

# Tests que usan la factory:
def test_tech_classification(make_classification_client):
    client = make_classification_client(category="tecnología", confidence=0.95, tags=["AI", "Python"])
    result = classify_document("Artículo sobre Python y AI", client=client)
    assert result["category"] == "tecnología"
    assert result["confidence"] == 0.95

def test_default_classification(make_classification_client):
    client = make_classification_client()  # Sin argumentos, usa defaults
    result = classify_document("Texto genérico", client=client)
    assert result["category"] == "general"

Ejercicio 2: Factory con parametrize

Usa la factory del Ejercicio 1 con @pytest.mark.parametrize para testear 5 categorías diferentes:

Ver solución

CLASSIFICATION_CASES = [
    pytest.param("tecnología", 0.95, ["AI", "Python"], id="tecnología"),
    pytest.param("ciencia", 0.88, ["física", "investigación"], id="ciencia"),
    pytest.param("deportes", 0.92, ["fútbol", "competencia"], id="deportes"),
    pytest.param("política", 0.75, ["elecciones"], id="política"),
    pytest.param("general", 0.60, [], id="sin_categoría_clara"),
]

@pytest.mark.parametrize("category,confidence,tags", CLASSIFICATION_CASES)
def test_all_categories(make_classification_client, category, confidence, tags):
    """Verifica que todas las categorías se procesan correctamente."""
    client = make_classification_client(
        category=category,
        confidence=confidence,
        tags=tags
    )
    
    result = classify_document("Texto de prueba", client=client)
    
    assert result["category"] == category
    assert result["confidence"] == confidence
    assert result["tags"] == tags

Ejercicio 3: Factory para error handling

Crea una factory make_error_client que permita simular diferentes tipos de errores. Escribe tests para:

Rate limit → la app retorna {"error": "rate_limit"}
Timeout → la app retorna {"error": "timeout"}
Respuesta vacía → la app usa category="unknown"

Ver solución

@pytest.fixture
def make_error_client():
    def _create(error_type: str) -> MagicMock:
        client = MagicMock()
        
        if error_type == "rate_limit":
            client.chat.completions.create.side_effect = openai.RateLimitError(
                message="Rate limit exceeded",
                response=MagicMock(status_code=429),
                body={}
            )
        elif error_type == "timeout":
            client.chat.completions.create.side_effect = openai.APITimeoutError(
                request=MagicMock()
            )
        elif error_type == "empty_response":
            client.chat.completions.create.return_value = create_openai_chat_response("")
        
        return client
    
    return _create

def test_rate_limit_handling(make_error_client):
    client = make_error_client("rate_limit")
    result = classify_document("texto", client=client)
    assert result == {"error": "rate_limit"}

def test_timeout_handling(make_error_client):
    client = make_error_client("timeout")
    result = classify_document("texto", client=client)
    assert result == {"error": "timeout"}

def test_empty_response_fallback(make_error_client):
    client = make_error_client("empty_response")
    result = classify_document("texto", client=client)
    assert result.get("category") == "unknown"

Ejercicio 4: Factory para chain

Crea una factory make_pipeline_client para un pipeline de tres pasos:

Primera LLM call: extrae entidades
Segunda LLM call: clasifica las entidades
Tercera LLM call: genera un resumen

Ver solución

@pytest.fixture
def make_pipeline_client():
    """Factory para pipelines de 3 pasos."""
    def _create(
        entities: list = None,
        classifications: list = None,
        summary: str = "Resumen generado"
    ) -> MagicMock:
        if entities is None:
            entities = [{"entity": "Python", "type": "TECNOLOGÍA"}]
        if classifications is None:
            classifications = [{"entity": "Python", "category": "lenguaje"}]
        
        client = MagicMock()
        client.chat.completions.create.side_effect = [
            create_openai_chat_response(json.dumps(entities)),
            create_openai_chat_response(json.dumps(classifications)),
            create_openai_chat_response(json.dumps({"summary": summary}))
        ]
        return client
    
    return _create

def test_pipeline_three_steps(make_pipeline_client):
    client = make_pipeline_client(
        entities=[{"entity": "FastAPI", "type": "FRAMEWORK"}],
        classifications=[{"entity": "FastAPI", "category": "web"}],
        summary="FastAPI es un framework web moderno"
    )
    
    result = run_pipeline("Artículo sobre FastAPI", client=client)
    
    assert result["summary"] == "FastAPI es un framework web moderno"
    assert client.chat.completions.create.call_count == 3

Ejercicio 5: Cuándo NO usar factory

Describe 3 situaciones donde una fixture estática es mejor que una factory:

Ver guía

Un solo caso estándar para todos los tests del módulo: Si todos los tests del módulo usan el mismo mock response (el "caso feliz" genérico), una fixture estática es más clara que una factory con defaults que nadie sobrescribe.
```
# Más claro:
@pytest.fixture
def standard_client():
    client = MagicMock()
    client.chat.completions.create.return_value = create_openai_chat_response(STANDARD_RESPONSE)
    return client
```
Fixture costosa que debe reutilizarse (scope="session"): Las factories crean objetos nuevos en cada llamada. Si la creación es costosa (e.g., inicializar una conexión real), una fixture con scope="module" o scope="session" es más eficiente.
```
@pytest.fixture(scope="module")
def db_connection():
    conn = create_real_db_connection()  # Costoso
    yield conn
    conn.close()
```
El caso es muy específico y no hay variaciones: Si solo tienes un test que necesita una respuesta con un error específico de API, es más claro escribirlo directamente en el test con with patch(...) que crear una factory para ese único caso.

Resumen

Factory fixture = fixture que retorna una función generadora de objetos configurables
Elimina la duplicación de fixtures estáticas con variaciones similares
Defaults inteligentes: especifica solo lo que cambia en cada test
Combina con parametrize para cobertura masiva con poco código
Factories especializadas: para errores, chains, async — una por responsabilidad
Cuándo usar estática: un solo caso, scope elevado, sin variaciones necesarias

Recursos adicionales

Factories as Fixtures — pytest docs — La referencia oficial
pytest fixtures — scope — Cuándo usar function vs session scope
pytest.mark.parametrize — Para combinar con factories
Python functools.partial — Alternativa funcional a factories
Dependency Injection en pytest — Cómo las fixtures son DI
conftest.py — pytest docs — Compartir factories entre módulos
pytest-lazy-fixture — Para parametrize con fixtures

← Anterior5. Testing Parsers y Output Processors Siguiente →7. Proyecto: Prompt Contract Tests