Testing ML Systems
Unit, integration y e2e con pytest: mocking, cobertura y tests deterministas.
El testing en sistemas de ML es diferente al testing en software tradicional porque tiene una capa extra de complejidad: el modelo mismo. Un sistema de ML tiene componentes testeables con técnicas estándar (la lógica de la API, el preprocessing, el postprocessing) y componentes que requieren estrategias distintas (el comportamiento del modelo ante distribuciones específicas). La clave es separar claramente qué estás testeando en cada test.
La pirámide de testing aplica directamente. La base son los unit tests: rápidos, deterministas, sin dependencias externas. Prueban funciones puras de preprocessing, la lógica de los casos de uso con mocks, y las reglas de validación. El nivel medio son los tests de integración: prueban los adaptadores reales (sklearn, postgres, la API HTTP) con datos controlados. En la cúspide están los e2e (end-to-end): el flujo completo desde request HTTP hasta respuesta, típicamente contra un ambiente de staging.
pytest es el framework estándar en Python porque su API es mínima y su sistema de fixtures es extremadamente poderoso. Una fixture es una función que provee un estado o recurso reutilizable: un cliente HTTP de test, una instancia del caso de uso con mocks, un dataset pequeño de referencia. Las fixtures eliminan el duplicado de setup entre tests y hacen explícito qué estado necesita cada test.
El mocking es esencial en la arquitectura hexagonal: permite testear un caso de uso sin instanciar el modelo real, sin base de datos, sin red. unittest.mock.MagicMock genera objetos que imitan la interfaz de cualquier clase. pytest-mock expone el mismo mecanismo como una fixture (mocker) que se limpia automáticamente entre tests. La regla: si un test necesita más de 100 ms para correr, probablemente está haciendo I/O real que debería estar mockeado.
Los tests de modelos de ML requieren pensar en invariantes, no en valores exactos. No puedes aseverar 'la probabilidad es 0.734' porque el modelo es estocástico y puede reentrenarse. Sí puedes aseverar que 'la probabilidad está en [0, 1]', que 'el cliente de mayor tenure tiene menor probabilidad de churn que el de menor tenure' (prueba de dirección), y que 'el modelo procesa 100 requests en menos de 1 segundo' (prueba de performance). Estos tests son estables ante reentrenamientos.
La cobertura (coverage) mide qué porcentaje de las líneas de código son ejecutadas por los tests. Una cobertura del 80-90% en lógica de negocio es alcanzable y valiosa; perseguir el 100% tiene rendimientos decrecientes y puede llevar a tests sin valor (que solo ejecutan código sin verificar nada). Usa pytest-cov para medir y configura coverage.ini para excluir código de infraestructura que es difícil de testear unitariamente.
# Install
pip install pytest pytest-mock pytest-cov httpx
# tests/conftest.py — shared fixtures
import pytest
from unittest.mock import MagicMock
from fastapi.testclient import TestClient
from src.domain.ports import ChurnScore, CustomerFeatures, ModelPort
from src.application.use_cases import PredictChurnUseCase
from src.infrastructure.adapters.fastapi_api import app
@pytest.fixture
def mock_model() -> MagicMock:
model = MagicMock(spec=ModelPort)
model.predict.return_value = ChurnScore(probability=0.82, label='churn')
return model
@pytest.fixture
def use_case(mock_model: MagicMock) -> PredictChurnUseCase:
return PredictChurnUseCase(model=mock_model)
@pytest.fixture
def client(mock_model: MagicMock) -> TestClient:
app.state.model = mock_model
return TestClient(app)
# tests/unit/test_use_case.py
def test_predict_returns_churn_score(use_case, mock_model):
features = CustomerFeatures(age=35, tenure=12.0, monthly_charges=75.0)
score = use_case.execute(features)
assert 0.0 <= score.probability <= 1.0
assert score.label in ('churn', 'no_churn')
mock_model.predict.assert_called_once_with(features)
# tests/integration/test_api.py
def test_predict_endpoint_returns_200(client):
response = client.post('/predict', json={
'age': 35, 'tenure': 12.0, 'monthly_charges': 75.0
})
assert response.status_code == 200
data = response.json()
assert 'churn_probability' in data
assert 0.0 <= data['churn_probability'] <= 1.0
# Run with coverage
pytest --cov=src --cov-report=term-missing tests/
# pytest.ini
[pytest]
testpaths = tests
filterwarnings = ignore::DeprecationWarningDebugging lab
Detecta y corrige el error en el código.
- 5.5.5.1
import random from src.application.use_cases import PredictChurnUseCase from src.infrastructure.adapters.sklearn_model import SklearnModelAdapter def test_predict_churn(): model = SklearnModelAdapter('model.pkl') use_case = PredictChurnUseCase(model=model) features = CustomerFeatures(age=random.randint(18, 80), tenure=random.random() * 24) score = use_case.execute(features) assert score.probability > 0
- 5.5.5.2
def test_predict_endpoint(client): response = client.post('/predict', json={ 'age': 35, 'tenure': 12.0, 'monthly_charges': 75.0 }) assert response.status_code == 200 assert response.json() == {'churn_probability': 0.82, 'label': 'churn'}
- 5.5.5.3
def test_invalid_age_returns_422(client): response = client.post('/predict', json={ 'age': -5, 'tenure': 12.0, 'monthly_charges': 75.0 }) assert response.status_code == 200
- 5.5.5.4
# Test de modelo — valores exactos import joblib import numpy as np def test_model_predicts_correctly(): model = joblib.load('model.pkl') X = np.array([[35, 12.0, 75.0]]) prob = model.predict_proba(X)[0][1] assert prob == 0.7341892
- 5.5.5.5
# Conftest con estado global mutable model_instance = None @pytest.fixture def use_case(): global model_instance if model_instance is None: model_instance = MagicMock(spec=ModelPort) model_instance.predict.return_value = ChurnScore(probability=0.5, label='no_churn') return PredictChurnUseCase(model=model_instance)