Тесты для трансформаций: pytest

Код загрузки меняют часто: добавилось поле, сменился формат даты, источник переехал. Без тестов каждое изменение проверяется запуском на проде, и ошибка в разборе строки обнаруживается по неверному отчёту через неделю. Тесты в пайплайне это не про покрытие, а про то, чтобы трансформацию можно было переписать за час и быть уверенным, что она считает то же самое.

Что тестировать

Тестировать стоит то, что ломается при правках и что нельзя быстро проверить глазами: разбор строк и приведение типов, правила очистки, расчёт полей, дедупликацию, границы окон инкремента. То есть чистые функции из середины пайплайна.

Не стоит тестировать библиотеки (csv умеет читать CSV без вас), настройки и соединения с реальной базой: это проверяется интеграционным прогоном на тестовом окружении, а не юнит-тестом.

Структура кода из прошлого урока делает это возможным: чистые функции принимают данные и возвращают данные, и в тесте им можно подать три строки вместо файла на миллион.

pytest

Тест это функция с именем test_* в файле test_*.py, внутри обычный assert:

from load_orders import parse_amount
 
def test_parses_plain_number():
    assert parse_amount("1200") == 1200
 
def test_strips_spaces():
    assert parse_amount(" 1200 ") == 1200
 
def test_empty_is_none():
    assert parse_amount("") is None

pytest находит такие файлы сам, запускает каждый тест отдельно и при падении показывает значения по обе стороны assert, без assertEqual и прочих обёрток. Один тест проверяет одну вещь, и его имя говорит, какую: по списку упавших тестов понятно, что сломалось, без чтения кода.

Запуск: pytest в корне проекта, pytest -x остановиться на первом падении, pytest -k amount только тесты с amount в имени.

Таблица граничных случаев

Для функций разбора ценность тестов в границах: пустая строка, пробелы, ноль, отрицательное, ведущие нули, не число, None. Перед тем как писать тесты, стоит выписать таблицу «вход → ожидание», и она же становится тестом:

import pytest
 
@pytest.mark.parametrize("raw, expected", [
    ("1200", 1200),
    (" 1200 ", 1200),
    ("0", 0),
    ("007", 7),
    ("", None),
    ("   ", None),
])
def test_parse_amount(raw, expected):
    assert parse_amount(raw) == expected

parametrize запускает один тест на каждую строку таблицы, и в отчёте каждая строка отдельно. Добавить случай это добавить строку. Такая таблица ещё и документация: по ней видно, что функция обещает.

Случаи, которые должны падать, проверяются отдельно:

@pytest.mark.parametrize("raw", ["-5", "1.5", "12 000", "abc"])
def test_rejects_bad_input(raw):
    with pytest.raises(ValueError):
        parse_amount(raw)

pytest.raises проходит, только если внутри блока возникло исключение указанного типа. Тест на «функция падает на плохом входе» так же важен, как на хорошем: молчаливое принятие "-5" это ошибка данных в витрине.

Файлы, база и API в тестах

Функция, которая читает файл, тестируется на временном файле. pytest даёт фикстуру tmp_path: пустая папка, уникальная для теста, удаляется после:

def test_sum_file_skips_blank_lines(tmp_path):
    path = tmp_path / "numbers.txt"
    path.write_text("10\n\n20\n", encoding="utf-8")
    assert sum_file(path) == 30

Базу и API в юнит-тестах заменяют подделкой: объект с тем же интерфейсом, который записывает вызовы или отдаёт заранее заданные ответы. В задачах прошлых уроков это были FakeCursor и FakeSession, и это не упрощение для песочницы, а обычная практика. Код загрузки для этого принимает соединение или сессию параметром, а не создаёт внутри: тогда в тест передаётся подделка, а в прод настоящий объект. Для готовых подделок с проверкой вызовов есть unittest.mock, для requests библиотека responses.

Тест, который ходит в настоящую базу, это интеграционный тест: он нужен, но живёт отдельно, помечается @pytest.mark.integration и запускается не на каждый коммит.

Фикстуры

Фикстура это подготовка данных для теста, вынесенная в функцию с декоратором @pytest.fixture. Тест получает её по имени параметра:

@pytest.fixture
def orders():
    return [
        {"order_id": 1, "status": "paid", "amount": 100},
        {"order_id": 2, "status": "new", "amount": None},
    ]
 
def test_only_paid(orders):
    assert [o["order_id"] for o in only_paid(orders)] == [1]

Маленький набор данных, собранный руками, лучше среза с прода: в нём видно, какие случаи покрыты, и он не содержит персональных данных.

Как понять, что тест полезен

Полезный тест падает, когда функцию сломали. Проверить это просто: внести в функцию ошибку (убрать strip, заменить is None на not value) и посмотреть, какой тест покраснел. Если никакой, тестов на этот случай нет. Задачи ниже устроены ровно так: песочница подменяет функцию сломанной версией и проверяет, что ваши тесты это замечают.

Тест, который проверяет реализацию, а не поведение (например, что внутри вызван strip), ломается при любом рефакторинге и ничего не защищает. Тестируются вход и выход.

Как отвечать на собеседовании

Как тестируете код загрузок? Чистые функции трансформации юнит-тестами на pytest с таблицей граничных случаев, файлы через временную папку, база и API через подделки с тем же интерфейсом. Интеграционные тесты с настоящей базой отдельно, не на каждый коммит.

Как проверить, что функция падает на плохом входе? with pytest.raises(ValueError) вокруг вызова: тест проходит, только если исключение возникло.

Что такое фикстура? Функция подготовки данных или окружения с @pytest.fixture, которую тест получает по имени параметра. tmp_path встроенная фикстура с временной папкой.

Как понять, что тестов достаточно? Сломать функцию и посмотреть, упал ли тест. Если правка проходит незамеченной, этого случая в тестах нет.

Практика

В задачах ниже вы пишете тесты, а не функции. Проверяемая функция уже есть над редактором. Песочница запускает ваши тесты дважды: с правильной функцией (все должны пройти) и с несколькими сломанными версиями (хотя бы один тест должен упасть на каждой). pytest в песочнике загружается как пакет, его raises работает, запуск тестов делает обвязка.

import pytest


def parse_amount(raw):
    value = raw.strip()
    if value == "":
        return None
    if not value.isdigit():
        raise ValueError(f"не число: {raw!r}")
    return int(value)


def _no_strip(raw):
    if raw == "":
        return None
    if not raw.isdigit():
        raise ValueError(raw)
    return int(raw)


def _zero_for_empty(raw):
    value = raw.strip()
    if value == "":
        return 0
    if not value.isdigit():
        raise ValueError(raw)
    return int(value)


def _accepts_negative(raw):
    value = raw.strip()
    if value == "":
        return None
    return int(value)

def _run_tests(impl=None):
    g = globals()
    original = g["parse_amount"]
    if impl is not None:
        g["parse_amount"] = impl
    tests = [v for k, v in sorted(g.items()) if k.startswith("test_") and callable(v)]
    failed = 0
    try:
        for test in tests:
            try:
                test()
            except BaseException:
                failed += 1
    finally:
        g["parse_amount"] = original
    return len(tests), failed