Тестирование
Учимся тестировать ML-приложения
Вопрос 1
### Вопрос
Как написать минимальный тест, который убеждается, что модель вообще способна обучаться: градиенты текут и параметры обновляются до запуска полноценной тренировки на больших данных?
Ответ
### Ответ
Напишем smoke test для обучения. Проверяем три вещи: у всех обучаемых параметров есть ненулевые градиенты, параметры изменились после шага оптимизатора, loss убывает при переобучении на одном батче.
```python
def test_model_parameters_update(small_model):
torch.manual_seed(42)
x = torch.randn(4, 3, 32, 32)
y = torch.randint(0, 10, (4,))
params_before = {name: p.clone() for name, p in small_model.named_parameters()}
small_model.train()
optimizer = torch.optim.AdamW(small_model.parameters(), lr=1e-3)
loss = nn.CrossEntropyLoss()(small_model(x), y)
loss.backward()
optimizer.step()
for name, param in small_model.named_parameters():
assert param.grad is not None, f'нет градиента у {name}'
assert param.grad.abs().sum() > 0, f'нулевой градиент у {name}'
assert not torch.equal(param, params_before[name]), f'параметр {name}
не изменился'
def test_loss_decreases_on_overfit(small_model):
torch.manual_seed(42)
x = torch.randn(8, 3, 32, 32)
y = torch.randint(0, 10, (8,))
optimizer = torch.optim.AdamW(small_model.parameters(), lr=1e-3)
criterion = nn.CrossEntropyLoss()
small_model.train()
initial_loss = criterion(small_model(x), y).item()
for _ in range(50):
optimizer.zero_grad()
loss = criterion(small_model(x), y)
loss.backward()
optimizer.step()
assert loss.item() < initial_loss * 0.5, \
f'loss не убывает: {initial_loss:.4f} → {loss.item():.4f}'
```
Если используете PyTorch Lightning, то там есть встроенный `overfit_batches=1` в `Trainer`, который делает то же самое. Но понимать, что происходит под капотом всё равно полезно.
Вопрос 2
### Вопрос
Как реализовать регрессионный тест, который проверяет, что точность модели не деградировала? Сохранять сырые предсказания и сравнивать через == не работает — одни и те же веса дают разные float-значения при смене железа.
Ответ
### Ответ
Регрессионный тест не должен требовать совпадения тензоров, достаточно
проверить, что, например, F1 не деградировал относительно бейзлайна.
```python
# tests/conftest.py
def pytest_addoption(parser):
parser.addoption('--update-golden', action='store_true', default=False)
# tests/test_regression.py
GOLDEN_FILE = Path('tests/golden/model_metrics.json')
REGRESSION_THRESHOLD = 0.01
def compute_metrics(model, test_loader):
model.eval()
all_preds, all_labels = [], []
with torch.no_grad():
for x, y in test_loader:
preds = model(x).argmax(dim=-1)
all_preds.extend(preds.tolist())
all_labels.extend(y.tolist())
return {'f1': f1_score(all_labels, all_preds, average='macro')}
def test_model_regression(model, test_loader, request):
current_metrics = compute_metrics(model, test_loader)
if request.config.getoption('--update-golden'):
GOLDEN_FILE.write_text(json.dumps(current_metrics, indent=2))
pytest.skip('Golden file updated.')
golden_data = json.loads(GOLDEN_FILE.read_text())
for metric_name, golden_value in golden_data.items():
current_value = current_metrics[metric_name]
assert current_value >= golden_value - REGRESSION_THRESHOLD, (
f'Регрессия {metric_name}: было {golden_value:.4f}, стало
{current_value:.4f}'
)
```
Вопрос 3
### Вопрос
В pytest есть фикстуры с разными скоупами:
- `function`: выполняется до/после каждой функции-теста, если в неё передали эту фикстуру
- `class`: до/после каждого класса
- `module`: каждого модуля с тестами
- `package`: каждого пакета с тестами
- `session`: фикстура выполняется один раз на всю тестовую сессию
Зачем сделали столько разных видов скоупов? Почему для всех случаев нельзя обойтись только `function` или только `session`?
Ответ
### Ответ
Дело в балансе изоляция vs скорость
В идеале нам хотелось бы, чтобы тесты выполнялись очень быстро. Но что, если наши фикстуры поднимают базу данных или zookeeper/kafka? Если мы используем function-фикстуру, то перед каждой функцией-тестом будем ждать, пока поднимется kafka, а это очень долго. Так что обойтись только function-фикстурами нам не получится. Поэтому зачастую тяжёлые инфраструктурные объекты инициализируются как session-фикстуры или вообще используются не через механизм фикстур в pytest, а просто поднимаются «сбоку» перед запуском тестов.
Почему тогда не использовать везде session? Проблема в том, что session-фикстура — по сути глобальная мутабельная переменная. Из-за этого одни тесты могут влиять на результат других. Приведём пример
```python
@pytest.fixture(scope='session')
def user():
return {'name': 'Regular user', 'role': 'user'}
def test_promote_to_admin(user):
promote(user)
assert user['role'] == 'admin' # проходит
def test_regular_user_cannot_delete(user):
result = delete_resource(user, resource_id=1)
assert result.status_code == 403 # падает — user уже admin, получаем 200
```
Здесь второй тест доходит до `delete_resource`, но падает на ассерте — из-за того, что первый тест повысил пользователя до admin.
Если бы user был function-фикстурой, каждый тест получал бы свежий словарь и все тесты прошли бы независимо от порядка запуска.
Отсюда правило, на которое можно часто ориентироваться:
- `session/module` — для дорогих или немутабельных ресурсов. В идеале нам бы хотелось поставить здесь «и» вместо «или», но на деле почти всегда оказывается, что что-то дорогое обязательно будет мутабельным
- `function` — для всего, что тесты могут изменить и что не очень долго инициализировать
- `class/module` — промежуточный вариант, когда группа связанных тестов разделяет состояние намеренно, но изолирована от остальных
Вопрос 4
### Вопрос
Пусть нужно протестировать функцию `predict` с 10 комбинациями входных тензоров и ожидаемых результатов. Ты используешь `@pytest.mark.parametrize` и в выводе видишь тест-кейсы вида `test_predict[tensor0-1]`, `test_predict[tensor1-2]`. Бесполезные имена и при падении непонятно, какой именно сценарий сломался. Как сделать читаемые ID?
Ответ
**Ответ:**
У `@pytest.mark.parametrize` есть три способа задать читаемые ID: через `ids=` список строк, через `ids=` функцию, и через `pytest.param(..., id="...")` прямо в наборе данных.
```python
import pytest
import torch
# ids как список строк
@pytest.mark.parametrize(
'input_tensor, expected_class',
[
(torch.zeros(1, 3, 224, 224), 0),
(torch.ones(1, 3, 224, 224), 1),
],
ids=['black_image', 'white_image'],
)
def test_predict_v1(model, input_tensor, expected_class): ...
# pytest.param с id (данные и имя рядом)
@pytest.mark.parametrize(
'input_tensor, expected_class',
[
pytest.param(torch.zeros(1, 3, 224, 224), 0, id='black_image'),
pytest.param(torch.ones(1, 3, 224, 224), 1, id='white_image'),
],
)
def test_predict_v2(model, input_tensor, expected_class): ...
# ids как функция (автоматически для всех параметров)
def tensor_id(val):
if isinstance(val, torch.Tensor):
return f'tensor_{val.shape}_{val.dtype}'
return None # None = использовать дефолтный id
@pytest.mark.parametrize('x, y', [...], ids=tensor_id)
def test_predict_v3(model, x, y): ...
```
Результат: `test_predict_v2[black_image], test_predict_v2[white_image]`. Эти ID также можно также использовать с `pytest -k 'black_image'` для запуска нужного сценария.
Вопрос 5
### Вопрос
Что не так с этим тестом?
```python
def normalize(x: torch.Tensor) -> torch.Tensor:
return (x - x.mean()) / x.std()
def test_normalize():
x = torch.tensor([1.0, 2.0, 3.0])
expected = (x - x.mean()) / x.std()
assert torch.allclose(normalize(x), expected)
```
Ответ
### Ответ
`expected` вычисляется по той же формуле, что и сама функция. Тест не проверяет
правильность реализации. Он проверяет, что функция возвращает то же, что мы
сами только что посчитали тем же способом. Если в `normalize` закралась ошибка
(например, делим на дисперсию вместо стандартного отклонения), тест всё равно
пройдёт.
`expected` должен выражать ожидаемое поведение: что мы как пользователи ожидаем получить. В этом случае — результат ручного вычисления:
```python
def test_normalize():
x = torch.tensor([1.0, 2.0, 3.0])
expected = torch.tensor([-1.0, 0.0, 1.0])
assert torch.allclose(normalize(x), expected)
```
Здесь `expected` вычислен вручную: среднее равно 2, std равно 1, поэтому каждый элемент смещается на −1, 0, +1.
Вопрос 6
### Вопрос
В тестах нужно проверить, что ML-сервис читает `API_KEY` из переменных окружения. В CI переменная не установлена, тест падает с `KeyError`. Как безопасно мокать `os.environ` так, чтобы изменения не «утекали» между тестами?
Ответ
### Ответ
Можно использовать `patch.dict`. Он модифицирует словарь на время теста и восстанавливает исходное состояние. `clear=True` полезен для полной изоляции.
```python
import os
from unittest.mock import patch
# 1: patch.dict как контекстный менеджер
def test_reads_api_key():
with patch.dict(os.environ, {'OPENAI_API_KEY': 'test-key-123'}):
assert get_api_key() == 'test-key-123'
# 2: как декоратор
@patch.dict(os.environ, {'OPENAI_API_KEY': 'test-key-123'})
def test_reads_api_key_decorator():
assert get_api_key() == 'test-key-123'
# 3: полная изоляция, убрать все реальные переменные из словаря
def test_missing_key_raises():
with patch.dict(os.environ, {}, clear=True):
with pytest.raises(KeyError):
get_api_key()
```
В pytest можно использовать `monkeypatch`. Он автоматически откатывается после каждого теста:
```python
@pytest.fixture(autouse=True)
def mock_env(monkeypatch):
monkeypatch.setenv('OPENAI_API_KEY', 'test-key')
monkeypatch.setenv('MODEL_PATH', '/tmp/model.pt')
```
Вопрос 7
### Вопрос:
У тебя есть FastAPI-сервис с зависимостью `get_model()`, которая возвращает тяжёлую PyTorch-модель. В CI нет GPU, и `torch.load()` занимает 30 секунд. Как переписать тесты, чтобы они не загружали реальную модель?
Ответ
### Ответ:
Используй `app.dependency_overrides`. Это словарь, где ключ — оригинальная dependency-функция, а значение — это заглушка, которой хотим подменить оригинальную функцию.
**Ловушка:** `dependency_overrides` — это глобальное состояние. Если забыть очистить его после теста, эффект пролезет в следующие тесты.
```python
# main.py
app = FastAPI()
def get_model():
model = torch.load('convnext100500.pt')
model.eval()
return model
@app.post('/classify')
async def classify(data: dict, model=Depends(get_model)):
...
# conftest.py
class FakeModel:
def __call__(self, x):
return torch.zeros(1, 1000)
def eval(self):
return self
def fake_get_model():
return FakeModel()
@pytest.fixture
def client():
app.dependency_overrides[get_model] = fake_get_model
with TestClient(app) as c:
yield c
app.dependency_overrides.clear() # не забываем очистить
```
Вопрос 8
### Вопрос
Мы пишем сервис, который скачивает изображения с внешнего API и прогоняет их
через модель. Тест написан так:
```python
def test_pipeline_e2e():
url = 'https://api.example.com/images/cats'
result = run_pipeline(url)
assert result['label'] == 'cat'
assert result['confidence'] > 0.9
```
Тест иногда падает без изменений в коде. Назовите проблемы и покажите, как их
исправить.
Ответ
### Ответ
Внешний API может быть недоступен или отвечать медленно: тест будет то проходить, то падать без изменений в коде. Также API сегодня может вернуть одно изображение, а завтра другое. Поэтому тест проверяет не только логику пайплайна, а ещё и текущее состояние внешнего сервиса.
Решение: поднять тестовый сервер с фиксированным ответом.
```python
@pytest.fixture()
def image_server(httpserver):
with open('tests/fixtures/cat.jpg', 'rb') as f:
image_bytes = f.read()
httpserver.expect_request('/images/cats').respond_with_data(
image_bytes, content_type='image/jpeg'
)
yield httpserver
def test_pipeline_e2e(image_server):
url = image_server.url_for('/images/cats')
result = run_pipeline(url)
assert result['label'] == 'cat'
assert result['confidence'] > 0.9
```
Теперь тест полностью детерминирован: он всегда получает один и тот же файл из `tests/fixtures/cat.jpg` и проверяет именно логику пайплайна, а не доступность внешнего сервиса.
Вопрос 9
### Вопрос
Что не так с этим тестом?
```python
def test_inference_pipeline():
image = load_image('tests/fixtures/cat.jpg')
prediction = predict(image)
assert prediction['label'] == 'cat'
assert prediction['confidence'] > 0.9
visualization = draw_bboxes(image, prediction['boxes'])
assert visualization.shape == image.shape
assert visualization.dtype == np.uint8
save_result(prediction, path='tests/output/result.json')
with open('tests/output/result.json') as f:
saved = json.load(f)
assert saved['label'] == prediction['label']
```
Ответ
### Ответ
Тест проверяет три независимые вещи: качество предсказания, отрисовку боксов и сохранение результата. Если тест падает, непонятно что именно сломалось. Нужно разбить на три отдельных теста:
```python
def test_predict():
image = load_image('tests/fixtures/cat.jpg')
prediction = predict(image)
assert prediction['label'] == 'cat'
assert prediction['confidence'] > 0.9
def test_draw_bboxes():
image = load_image('tests/fixtures/cat.jpg')
boxes = [{'x': 10, 'y': 10, 'w': 50, 'h': 50}]
visualization = draw_bboxes(image, boxes)
assert visualization.shape == image.shape
assert visualization.dtype == np.uint8
def test_save_result():
prediction = {'label': 'cat', 'confidence': 0.95}
save_result(prediction, path='tests/output/result.json')
with open('tests/output/result.json') as f:
saved = json.load(f)
assert saved['label'] == prediction['label']
```
Теперь при падении сразу видно какой именно шаг сломался.
Вопрос 10
### Вопрос
Что не так с этим тестом?
```python
def test_classify_image():
image = load_image('tests/fixtures/sample.jpg')
result = classify(image)
if result['confidence'] > 0.9:
assert result['label'] in ['cat', 'dog']
elif result['confidence'] > 0.5:
assert result['label'] == 'unknown'
else:
assert result['label'] == 'unrecognized'
```
Ответ
### Ответ
Внутри этого теста три разные ветки и какая из них выполнится, зависит от
того, что вернёт classify. Тест недетерминирован: при разных прогонах может
тестироваться разное поведение. При падении непонятно, какой сценарий
сломался.
Каждую ветку нужно вынести в отдельный тест с явно заданными входными данными:
```python
def test_classify_high_confidence():
image = load_image('tests/fixtures/clear_cat.jpg')
result = classify(image)
assert result['label'] in ['cat', 'dog']
def test_classify_medium_confidence():
image = load_image('tests/fixtures/blurry.jpg')
result = classify(image)
assert result['label'] == 'unknown'
def test_classify_low_confidence():
image = load_image('tests/fixtures/noise.jpg')
result = classify(image)
assert result['label'] == 'unrecognized'
```
Теперь каждый тест проверяет ровно один сценарий и всегда проверяет именно
его.
Вопрос 11
### Вопрос:
Первый прогон тестов в CI занимает 5 минут, из которых 4.5 минуты — `pip install -r requirements.txt`. Как правильно настроить кеш в GitLab CI? Есть ли ловушки в конфигурации кеша pip?
Ответ
### Ответ:
Главная ловушка: кеш без ключа по файлу зависимостей устаревает и можно получить кешированные старые пакеты после обновления `requirements.txt`. Кешировать `.venv/` эффективнее чем только `.cache/pip`.
```yaml
# .gitlab-ci.yml
variables:
PIP_CACHE_DIR: "$CI_PROJECT_DIR/.cache/pip"
test:
image: python:3.12-slim
cache:
- key:
files:
- requirements.txt
- requirements-dev.txt
paths:
- .cache/pip
- .venv/
policy: pull-push
before_script:
- python -m venv .venv
- source .venv/bin/activate
- pip install -r requirements.txt -r requirements-dev.txt
script:
- source .venv/bin/activate
- pytest tests
```
Почему `.venv/` в кеше лучше чем только `.cache/pip`: pip-кеш хранит wheel-файлы, но не установленные пакеты. При кешировании `.venv/` мы пропускаем `pip install` полностью при попадании в кеш.
Вопрос 12
### Вопрос
Что не так с этим тестом?
```python
def test_normalize():
x = torch.tensor([1, 2, 3])
result = normalize(x)
assert result.tolist() == [-1.0, 0.0, 1.0]
```
Ответ
### Ответ
Сравнение `float` через `==` ненадёжно из-за погрешностей вычислений с плавающей
запятой. `normalize` может вернуть [-1.0000001, 2e-8, 0.9999999] — верный результат, но тест упадёт.
Нужно сравнивать с погрешностью:
```python
def test_normalize():
x = torch.tensor([1, 2, 3])
result = normalize(x)
expected = torch.tensor([-1.0, 0.0, 1.0])
assert torch.allclose(result, expected, atol=1e-6)
```
Для numpy — `np.testing.assert_allclose`, для скалярных значений — `pytest.approx`:
`assert accuracy(preds, labels) == pytest.approx(0.857, abs=1e-3)`
Вопрос 13
### Вопрос
Коллега предлагает покрыть ML-сервис только e2e-тестами: «Зачем юнит-тесты,
если e2e проверяют реальное поведение?» Что не так с этой логикой?
Ответ
### Ответ
E2e-тесты медленные (нужно поднять модель, сервис, все зависимости), хрупкие
(ломаются от изменений в любом слое), и при падении не говорят, где именно
сломалось. Если всё покрыто только e2e — разработка тормозит (на каждый чих жалко запускать тесты), а дебаг превращается в угадайку.
Можно делить тесты по уровням:
```python
# unit — быстрые, без GPU, без I/O
def test_normalize_output_range():
raw = torch.tensor([[255.0, 0.0, 128.0]])
result = normalize_input(raw)
assert result.min() >= -1.0 and result.max() <= 1.0
# инварианты модели — модель как чёрный ящик
def test_model_output_shape(dummy_model, sample_batch):
output = dummy_model(sample_batch)
assert output.shape == (sample_batch.shape[0], NUM_CLASSES)
# интеграционные — поднимаем только сервис, модель — mock'аем
def test_predict_endpoint_returns_200(mock_model):
with patch('app.model.loaded_model', mock_model):
response = TestClient(app).post('/predict', json={'data': [1.0]})
assert response.status_code == 200
# e2e — тестируем весь сервис целиком вместе с моделью
# pytest -m e2e
```