2. Mocking de Respuestas LLM

Descripción

Mockear correctamente las respuestas del LLM es la base del unit testing para AI. Esta cápsula cubre unittest.mock.patch, MagicMock, AsyncMock, y cómo estructurar mocks que simulen la API real de OpenAI/Anthropic. Mocks realistas evitan sorpresas cuando el código interactúa con la API real y garantizan que los tests validan el comportamiento correcto.

Por qué los mocks deben ser realistas

Un mock mal construido es peor que no tener mock: te da falsa seguridad. Considera este ejemplo:

# Mock trivial (NO hagas esto)
@pytest.fixture
def bad_mock():
    client = MagicMock()
    client.chat.completions.create.return_value = "Hello"  # ❌ String simple
    return client

def test_with_bad_mock(bad_mock):
    result = analyze_sentiment("texto", client=bad_mock)
    # ⚠️ Si tu código hace: response.choices[0].message.content
    # va a fallar con AttributeError: 'str' object has no attribute 'choices'
    # — PERO en este test puede pasar si el test tampoco usa choices

# Mock realista (SÍ hagas esto)
@pytest.fixture
def good_mock():
    client = MagicMock()
    # Replica exactamente la estructura de openai.ChatCompletion
    mock_choice = MagicMock()
    mock_choice.message.content = '{"sentiment": "positivo", "score": 0.85}'
    mock_choice.message.role = "assistant"
    mock_choice.finish_reason = "stop"
    mock_choice.index = 0
    
    mock_response = MagicMock()
    mock_response.choices = [mock_choice]
    mock_response.usage.prompt_tokens = 45
    mock_response.usage.completion_tokens = 25
    mock_response.usage.total_tokens = 70
    mock_response.model = "gpt-4o-mini"
    mock_response.id = "chatcmpl-mock-test-123"
    mock_response.created = 1700000000
    
    client.chat.completions.create.return_value = mock_response
    return client

Regla de oro: Tu mock debe tener exactamente la misma estructura que el objeto real. Así, si tu código accede a response.usage.total_tokens, el mock no falla.

Estructura de respuesta OpenAI: referencia completa

Para construir mocks realistas, necesitas conocer la estructura real:

# Lo que realmente retorna client.chat.completions.create()
# (Simplificado pero fiel a la estructura real)
{
    "id": "chatcmpl-abc123",
    "object": "chat.completion",
    "created": 1700000000,
    "model": "gpt-4o-mini",
    "choices": [
        {
            "index": 0,
            "message": {
                "role": "assistant",
                "content": '{"sentiment": "positivo", "score": 0.85}',
                "tool_calls": None,
                "function_call": None
            },
            "finish_reason": "stop",   # o "length", "content_filter", "tool_calls"
            "logprobs": None
        }
    ],
    "usage": {
        "prompt_tokens": 45,
        "completion_tokens": 25,
        "total_tokens": 70,
        "completion_tokens_details": {
            "reasoning_tokens": 0,
            "audio_tokens": 0
        }
    },
    "system_fingerprint": "fp_abc123"
}

Helper function: create_openai_chat_response

Esta función es el estándar que usarás en todos los tests del módulo:

# tests/helpers.py
from unittest.mock import MagicMock

def create_openai_chat_response(
    content: str,
    model: str = "gpt-4o-mini",
    prompt_tokens: int = 45,
    completion_tokens: int = 25,
    finish_reason: str = "stop"
) -> MagicMock:
    """
    Crea un mock que replica la estructura real de openai.ChatCompletion.
    
    Uso:
        mock_response = create_openai_chat_response('{"sentiment": "positivo"}')
        client.chat.completions.create.return_value = mock_response
    """
    # Construir choice
    mock_message = MagicMock()
    mock_message.role = "assistant"
    mock_message.content = content
    mock_message.tool_calls = None
    mock_message.function_call = None
    
    mock_choice = MagicMock()
    mock_choice.index = 0
    mock_choice.message = mock_message
    mock_choice.finish_reason = finish_reason
    mock_choice.logprobs = None
    
    # Construir usage
    mock_usage = MagicMock()
    mock_usage.prompt_tokens = prompt_tokens
    mock_usage.completion_tokens = completion_tokens
    mock_usage.total_tokens = prompt_tokens + completion_tokens
    
    # Construir respuesta completa
    mock_response = MagicMock()
    mock_response.id = "chatcmpl-mock-test-abc123"
    mock_response.object = "chat.completion"
    mock_response.created = 1700000000
    mock_response.model = model
    mock_response.choices = [mock_choice]
    mock_response.usage = mock_usage
    mock_response.system_fingerprint = "fp_test"
    
    return mock_response

Las tres formas de hacer patch

Forma 1: Decorator @patch

from unittest.mock import patch, MagicMock
from tests.helpers import create_openai_chat_response

@patch("app.sentiment.client.chat.completions.create")
def test_analyze_sentiment_decorator(mock_create):
    """Usando @patch como decorator."""
    # Configurar el mock
    mock_create.return_value = create_openai_chat_response(
        '{"sentiment": "positivo", "score": 0.9, "explanation": "Texto positivo"}'
    )
    
    # Ejecutar
    result = analyze_sentiment("Me encanta este producto")
    
    # Assertions sobre el resultado
    assert result["sentiment"] == "positivo"
    assert result["score"] == 0.9
    
    # Assertions sobre el mock (verificar que se llamó correctamente)
    mock_create.assert_called_once()
    call_args = mock_create.call_args
    assert call_args.kwargs["model"] == "gpt-4o-mini"
    assert "Me encanta este producto" in str(call_args.kwargs["messages"])

Forma 2: Context manager with patch

def test_analyze_sentiment_context():
    """Usando with patch como context manager."""
    with patch("app.sentiment.client.chat.completions.create") as mock_create:
        mock_create.return_value = create_openai_chat_response(
            '{"sentiment": "negativo", "score": 0.2, "explanation": "Texto negativo"}'
        )
        
        result = analyze_sentiment("Este producto es terrible")
        
        assert result["sentiment"] == "negativo"
        assert result["score"] == 0.2

Forma 3: Fixture en conftest.py (la más usada)

# tests/conftest.py
import pytest
from unittest.mock import MagicMock
from tests.helpers import create_openai_chat_response

@pytest.fixture
def mock_openai_client():
    """Fixture estándar: mock del cliente OpenAI para unit tests."""
    client = MagicMock()
    client.chat.completions.create.return_value = create_openai_chat_response(
        '{"sentiment": "neutral", "score": 0.5, "explanation": "Texto neutral"}'
    )
    return client

# En el test:
def test_analyze_sentiment_fixture(mock_openai_client):
    result = analyze_sentiment("Texto de prueba", client=mock_openai_client)
    assert result["sentiment"] == "neutral"

Comparación de los tres enfoques

El patrón "donde se usa" (crítico)

El error más común en mocking: hacer patch en el lugar equivocado.

# ❌ INCORRECTO: patch donde se DEFINE
@patch("openai.OpenAI")
def test_wrong(mock_openai):
    result = analyze_sentiment("texto")  # No funciona — el objeto ya fue importado
    
# ✅ CORRECTO: patch donde se USA (donde se importa en tu módulo)
@patch("app.sentiment.openai.OpenAI")    # si app.sentiment hace: import openai
# o
@patch("app.sentiment.OpenAI")           # si app.sentiment hace: from openai import OpenAI
# o
@patch("app.sentiment.client")           # si app.sentiment tiene: client = OpenAI(...)

La regla: patch("modulo.donde.se.usa.el.nombre")

# app/sentiment.py
from openai import OpenAI          # → patch("app.sentiment.OpenAI")
import openai                       # → patch("app.sentiment.openai.OpenAI")
client = OpenAI()                   # → patch("app.sentiment.client")

Mocking de errores y edge cases

Los errores son tan importantes como los casos de éxito. Tu app debe manejar errores del LLM gracefully.

Simular errores de API

import openai
from unittest.mock import patch

def test_handles_rate_limit():
    """La app debe manejar rate limit sin crashear."""
    with patch("app.sentiment.client.chat.completions.create") as mock_create:
        # Simular error 429
        mock_create.side_effect = openai.RateLimitError(
            message="Rate limit exceeded",
            response=MagicMock(status_code=429),
            body={"error": {"message": "Rate limit exceeded"}}
        )
        
        result = analyze_sentiment("texto")
        
        # La app debe retornar error controlado, no crashear
        assert result["error"] == "rate_limit"
        # o: assert result is None
        # o: verificar que loguea el error

def test_handles_timeout():
    """La app debe manejar timeouts."""
    with patch("app.sentiment.client.chat.completions.create") as mock_create:
        mock_create.side_effect = openai.APITimeoutError(
            request=MagicMock()
        )
        
        with pytest.raises(TimeoutError):
            analyze_sentiment("texto")
        # o verificar fallback behavior

def test_handles_api_error():
    """Error genérico de API."""
    with patch("app.sentiment.client.chat.completions.create") as mock_create:
        mock_create.side_effect = openai.APIError(
            message="Service unavailable",
            request=MagicMock(),
            body=None
        )
        
        result = analyze_sentiment("texto")
        assert "error" in result

Edge cases en el contenido

@pytest.fixture
def mock_client_empty_response():
    """Mock que retorna respuesta vacía."""
    client = MagicMock()
    client.chat.completions.create.return_value = create_openai_chat_response("")
    return client

@pytest.fixture
def mock_client_malformed_json():
    """Mock que retorna JSON malformado."""
    client = MagicMock()
    client.chat.completions.create.return_value = create_openai_chat_response(
        '{"sentiment": "positivo", "score": 0.9'  # JSON incompleto
    )
    return client

@pytest.fixture
def mock_client_markdown_json():
    """Mock que retorna JSON dentro de markdown code block."""
    client = MagicMock()
    client.chat.completions.create.return_value = create_openai_chat_response(
        '```json\n{"sentiment": "positivo", "score": 0.9}\n```'
    )
    return client

@pytest.fixture
def mock_client_truncated():
    """Mock que simula respuesta truncada (finish_reason=length)."""
    client = MagicMock()
    client.chat.completions.create.return_value = create_openai_chat_response(
        '{"sentiment": "positivo", "score": 0.9, "explanation": "Texto muy la',  # Truncado
        finish_reason="length"
    )
    return client

# Tests usando estos mocks:
def test_handles_empty_response(mock_client_empty_response):
    """El parser debe manejar respuesta vacía sin crashear."""
    result = analyze_sentiment("texto", client=mock_client_empty_response)
    assert result["sentiment"] == "unknown"  # o el fallback que definas
    
def test_handles_malformed_json(mock_client_malformed_json):
    """El parser debe manejar JSON malformado."""
    result = analyze_sentiment("texto", client=mock_client_malformed_json)
    assert "error" in result or result.get("sentiment") == "unknown"
    
def test_handles_markdown_json(mock_client_markdown_json):
    """El parser debe extraer JSON de markdown code blocks."""
    result = analyze_sentiment("texto", client=mock_client_markdown_json)
    assert result["sentiment"] == "positivo"  # Debe parsear correctamente

Mock con side_effect para múltiples llamadas

def test_retry_logic():
    """La app reintenta en caso de error transitorio."""
    with patch("app.sentiment.client.chat.completions.create") as mock_create:
        # Primera llamada falla, segunda éxito
        mock_create.side_effect = [
            openai.APITimeoutError(request=MagicMock()),  # Primera: timeout
            create_openai_chat_response(                    # Segunda: éxito
                '{"sentiment": "positivo", "score": 0.9}'
            )
        ]
        
        result = analyze_sentiment("texto")
        
        assert result["sentiment"] == "positivo"
        assert mock_create.call_count == 2  # Se llamó dos veces

def test_chain_multiple_calls():
    """Una chain hace dos llamadas LLM: extract + classify."""
    with patch("app.chain.client.chat.completions.create") as mock_create:
        mock_create.side_effect = [
            create_openai_chat_response('["feature1", "feature2"]'),  # Extracción
            create_openai_chat_response('{"category": "tech", "confidence": 0.95}')  # Clasificación
        ]
        
        result = extract_and_classify("Texto de entrada")
        
        assert result["category"] == "tech"
        assert mock_create.call_count == 2

Async mocking con AsyncMock

Si tu app usa async LLM calls, necesitas AsyncMock:

# app/async_sentiment.py
import asyncio
import openai

async def analyze_sentiment_async(text: str, client) -> dict:
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": text}]
    )
    return parse_sentiment(response.choices[0].message.content)

# test_async.py
import pytest
from unittest.mock import AsyncMock, MagicMock
from tests.helpers import create_openai_chat_response

@pytest.mark.asyncio
async def test_analyze_sentiment_async():
    """Test de función async con AsyncMock."""
    # AsyncMock para funciones async
    mock_create = AsyncMock(
        return_value=create_openai_chat_response(
            '{"sentiment": "positivo", "score": 0.9}'
        )
    )
    
    mock_client = MagicMock()
    mock_client.chat.completions.create = mock_create
    
    result = await analyze_sentiment_async("texto", client=mock_client)
    
    assert result["sentiment"] == "positivo"
    mock_create.assert_called_once()

# Fixture async:
@pytest.fixture
def async_mock_client():
    """Mock async del cliente OpenAI."""
    client = MagicMock()
    client.chat.completions.create = AsyncMock(
        return_value=create_openai_chat_response(
            '{"sentiment": "neutral", "score": 0.5}'
        )
    )
    return client

Verificar que el mock fue llamado correctamente

No solo el output es importante — también debes verificar que tu código llamó al LLM con los parámetros correctos:

def test_calls_llm_with_correct_params(mock_openai_client):
    """Verifica que el LLM se llama con los parámetros correctos."""
    with patch("app.sentiment.client.chat.completions.create") as mock_create:
        mock_create.return_value = create_openai_chat_response(
            '{"sentiment": "positivo", "score": 0.9}'
        )
        
        analyze_sentiment("Texto de prueba")
        
        # Verificar que se llamó exactamente una vez
        mock_create.assert_called_once()
        
        # Obtener los argumentos con que se llamó
        call_kwargs = mock_create.call_args.kwargs
        
        # Verificar modelo
        assert call_kwargs.get("model") == "gpt-4o-mini"
        
        # Verificar que el texto apareció en los messages
        messages = call_kwargs.get("messages", [])
        user_message = next(m for m in messages if m["role"] == "user")
        assert "Texto de prueba" in user_message["content"]
        
        # Verificar temperatura (si la controlas)
        assert call_kwargs.get("temperature", 0) == 0

def test_does_not_call_llm_for_cached_result(mock_openai_client):
    """Verifica que el caché evita llamadas extra al LLM."""
    with patch("app.sentiment.client.chat.completions.create") as mock_create:
        mock_create.return_value = create_openai_chat_response(
            '{"sentiment": "positivo", "score": 0.9}'
        )
        
        # Primera llamada
        analyze_sentiment("mismo texto")
        # Segunda llamada (debe usar caché)
        analyze_sentiment("mismo texto")
        
        # Solo debe haber llamado al LLM una vez
        mock_create.assert_called_once()

Comparación: unittest.mock vs pytest-mock

# Con unittest.mock (estándar Python):
from unittest.mock import patch, MagicMock

@patch("app.sentiment.client.chat.completions.create")
def test_with_unittest(mock_create):
    mock_create.return_value = create_openai_chat_response("...")
    # ...

# Con pytest-mock (más ergonómico):
def test_with_pytest_mock(mocker):
    mock_create = mocker.patch("app.sentiment.client.chat.completions.create")
    mock_create.return_value = create_openai_chat_response("...")
    # ...

Estructura de Anthropic: diferencias

Si usas Anthropic, la estructura del mock es diferente:

# Respuesta real de Anthropic:
{
    "id": "msg_abc123",
    "type": "message",
    "role": "assistant",
    "content": [
        {
            "type": "text",
            "text": '{"sentiment": "positivo", "score": 0.9}'
        }
    ],
    "model": "claude-3-haiku-20240307",
    "stop_reason": "end_turn",
    "usage": {
        "input_tokens": 45,
        "output_tokens": 25
    }
}

# Mock para Anthropic:
def create_anthropic_response(content: str, model: str = "claude-3-haiku-20240307"):
    mock_content_block = MagicMock()
    mock_content_block.type = "text"
    mock_content_block.text = content
    
    mock_usage = MagicMock()
    mock_usage.input_tokens = 45
    mock_usage.output_tokens = 25
    
    mock_response = MagicMock()
    mock_response.id = "msg_mock_test"
    mock_response.type = "message"
    mock_response.role = "assistant"
    mock_response.content = [mock_content_block]
    mock_response.model = model
    mock_response.stop_reason = "end_turn"
    mock_response.usage = mock_usage
    
    return mock_response

Tip de abstracción: Si tu app puede usar OpenAI o Anthropic, considera una función normalize_response(response, provider) en tu código que unifique la interfaz — eso también hace el testing más simple.

Ejercicios

Ejercicio 1: Construir un mock realista

Tu función generate_title llama a OpenAI y parsea el output así:

raw = response.choices[0].message.content
return raw.strip()  # El título es el texto completo

Construye un mock para testear que:

1. El título retornado es el contenido del mock
2. La función se llamó exactamente una vez
3. La función fue llamada con max_tokens=50

from unittest.mock import patch, MagicMock
from tests.helpers import create_openai_chat_response

def test_generate_title():
    with patch("app.title.client.chat.completions.create") as mock_create:
        mock_create.return_value = create_openai_chat_response(
            "  Introducción a Python  "  # Con espacios para verificar strip()
        )
        
        result = generate_title("Escribe un tutorial de Python")
        
        # Verificar resultado
        assert result == "Introducción a Python"  # strip() aplicado
        
        # Verificar llamada
        mock_create.assert_called_once()
        call_kwargs = mock_create.call_args.kwargs
        assert call_kwargs.get("max_tokens") == 50

Ejercicio 2: Mock de error con fallback

Tu app tiene este comportamiento: si el LLM falla, retorna {"sentiment": "unknown", "error": True}.

Escribe el test que verifica este comportamiento:

import openai
from unittest.mock import patch, MagicMock

def test_fallback_on_api_error():
    with patch("app.sentiment.client.chat.completions.create") as mock_create:
        mock_create.side_effect = openai.APIConnectionError(
            message="Connection failed",
            request=MagicMock()
        )
        
        result = analyze_sentiment("Texto de prueba")
        
        # Verificar fallback
        assert result["sentiment"] == "unknown"
        assert result["error"] is True
        # La app no debe relanzar la excepción

Ejercicio 3: Múltiples llamadas secuenciales

Tu función summarize_and_classify hace dos llamadas:

1. Primera: resume el texto
2. Segunda: clasifica el resumen

Escribe el test con side_effect que verifica que ambas llamadas se hacen:

from tests.helpers import create_openai_chat_response

def test_summarize_and_classify():
    with patch("app.chain.client.chat.completions.create") as mock_create:
        # Respuestas para la primera y segunda llamada
        mock_create.side_effect = [
            create_openai_chat_response(
                '{"summary": "Python es popular para AI"}'
            ),
            create_openai_chat_response(
                '{"category": "technology", "confidence": 0.95}'
            )
        ]
        
        result = summarize_and_classify("Texto largo sobre Python...")
        
        # Verificar resultado final
        assert result["summary"] == "Python es popular para AI"
        assert result["category"] == "technology"
        
        # Verificar que se hicieron exactamente 2 llamadas
        assert mock_create.call_count == 2

Ejercicio 4: AsyncMock para función async

Convierte el test del Ejercicio 1 para trabajar con una versión async de generate_title:

# La función async:
async def generate_title_async(prompt: str, client) -> str:
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=50
    )
    return response.choices[0].message.content.strip()

import pytest
from unittest.mock import AsyncMock, MagicMock
from tests.helpers import create_openai_chat_response

@pytest.mark.asyncio
async def test_generate_title_async():
    mock_client = MagicMock()
    mock_client.chat.completions.create = AsyncMock(
        return_value=create_openai_chat_response("  Introducción a Python  ")
    )
    
    result = await generate_title_async("Escribe un tutorial de Python", client=mock_client)
    
    assert result == "Introducción a Python"
    mock_client.chat.completions.create.assert_called_once()
    call_kwargs = mock_client.chat.completions.create.call_args.kwargs
    assert call_kwargs.get("max_tokens") == 50

Ejercicio 5: Diagnosticar el patch

El siguiente test no funciona — el mock no se aplica. ¿Por qué y cómo arreglarlo?

# app/summarizer.py
import openai
client = openai.OpenAI()

def summarize(text: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": text}]
    )
    return response.choices[0].message.content

# test_summarizer.py
@patch("openai.OpenAI")   # ← ¿Está bien?
def test_summarize(mock_openai_class):
    mock_instance = MagicMock()
    mock_openai_class.return_value = mock_instance
    mock_instance.chat.completions.create.return_value = create_openai_chat_response("Test")
    
    result = summarize("texto")  # ← ¿Funciona?

@patch("app.summarizer.client")
def test_summarize(mock_client):
    mock_client.chat.completions.create.return_value = create_openai_chat_response("Test")
    
    result = summarize("texto")
    
    assert result == "Test"
    mock_client.chat.completions.create.assert_called_once()

def summarize(text: str, client=None) -> str:
    if client is None:
        client = openai.OpenAI()
    response = client.chat.completions.create(...)
    return response.choices[0].message.content

Resumen

• Mocks deben replicar la estructura real de la API (choices, message, usage): usar create_openai_chat_response
• Patch donde se usa, no donde se define — patch("modulo.que.lo.usa.nombre")
• Tres formas de patch: decorator, context manager, fixture — cada una tiene su lugar
• Edge cases: empty response, malformed JSON, finish_reason=length, API errors
• AsyncMock para funciones async — no uses MagicMock para corrutinas
• Verificar interacciones: assert_called_once(), call_args.kwargs para validar que el LLM se llama correctamente

Recursos adicionales

1. unittest.mock — Documentación oficial Python — Referencia completa
2. Where to patch — Python docs — Guía crítica para entender el scope del patch
3. pytest-mock — Plugin para sintaxis más limpia
4. OpenAI API Reference — Chat Completions — Estructura real del response object
5. Anthropic API Reference — Estructura de Anthropic messages
6. Testing async code with pytest-asyncio — Para funciones async
7. Mock cookbook — Guía rápida con patrones comunes