Тестирование ETL-пайплайнов с pytest в 2026: unit, интеграционные и e2e

Обновлено: 2 сентября 2026

Тестирование ETL-пайплайнов на pytest в 2026 году сводится к трём слоям: unit-тесты для чистых функций-трансформаций, интеграционные тесты с реальными базами через testcontainers и e2e-прогон одного маленького батча end-to-end. Такой набор ловит около 90% багов до продакшена и делает бэкфилы предсказуемыми. Ниже расскажу, как я собираю этот стек с pytest 8.x, Pandera 0.29 и dbt-tests, чтобы пайплайн не разваливался в 3 часа ночи, когда его никто не смотрит.

• pytest 8.x поддерживает Python 3.9+ и остаётся стандартом для тестирования пайплайнов данных, плюс плагины pytest-xdist, pytest-cov, pytest-mock.
• Пирамида тестов данных: много быстрых unit-тестов, средний слой интеграционных тестов с testcontainers, минимум медленных e2e. Обратное соотношение убивает CI.
• Мокать базу в интеграционных тестах, честно говоря, плохая идея: реальные схемы, констрейнты и типы ловят миграционные баги, которые моки пропускают.
• Pandera 0.29 (январь 2026) даёт единые схемы для pandas, Polars, PySpark и Dask, так что можно валидировать один и тот же контракт на всех этапах пайплайна.
• dbt-tests и pytest не конкурируют: dbt отвечает за уровень SQL-моделей, а pytest покрывает Python-код вокруг них.
• Параметризация и hypothesis заменяют десятки повторяющихся тестов и находят краевые случаи, которых руками не придумать.

Почему тесты ETL это не роскошь, а страховка от 3:00

У меня за плечами два инцидента, которые сформировали моё отношение к тестам данных. Первый – сломанный бэкфил, второй – тихая потеря 4% строк из-за неявного каста типа. Оба чинились по паре часов, оба стоили доверия команды на месяцы. С тех пор правило простое: если пайплайн грузит хоть что-то в прод, у него должен быть pytest-набор, который прогоняется меньше чем за минуту и ловит регрессии.

Так вот, хорошие тесты пайплайнов дают три вещи, которые невозможно получить мониторингом. Во-первых, обратная связь на этапе PR: сломанная трансформация не доедет до Airflow, а упадёт в GitHub Actions за две минуты. Во-вторых, они позволяют делать бэкфилы без ужаса, потому что можно запустить старый DAG на исторический период, зная, что контракты входов и выходов не изменились. В-третьих, тесты становятся исполняемой документацией: новый инженер читает test_transform_orders.py и понимает, какие форматы входа поддерживаются и что происходит с NaN.

Мониторинг ловит проблему уже в продакшене. Тесты видят её до того, как она туда попала. Одно не заменяет другое: их роли ортогональны, и data-команды, у которых нет ни одного из двух слоёв, живут в постоянном режиме тушения пожаров.

Пирамида тестирования данных: unit, интеграционные и e2e

Классическая пирамида Майка Кона отлично ложится на пайплайны данных, если правильно понимать её этажи. Внизу лежат сотни быстрых unit-тестов на функции-трансформации: они работают на in-memory DataFrame, выполняются за миллисекунды, запускаются на каждый git push. В середине сидят десятки интеграционных тестов, которые поднимают реальный Postgres или ClickHouse через testcontainers и проверяют, что SQL, миграции и типы согласованы. Наверху остаются единицы e2e-тестов, которые прогоняют весь пайплайн на крошечном сэмпле от источника до витрины.

Соотношение примерно 70/25/5 по количеству тестов. Обратная пропорция – типичная антипаттерн-ловушка: команда сначала пишет один жирный e2e, потом ещё десять, и CI начинает идти двадцать минут. После этого разработчики просто перестают запускать тесты локально, и вся система деградирует до плацебо.

Что ловит каждый слой

• Unit — логические баги в трансформациях: неверные агрегации, потерянные строки при join, неправильная обработка пустых DataFrame и NaN.
• Integration — расхождения между SQL и ORM, сломанные миграции, race conditions при параллельной записи, ошибки в CREATE INDEX.
• E2E — регрессии в оркестрации: DAG собирается, все таски видят друг друга, зависимости в правильном порядке, идемпотентность повторного запуска.

Настройка pytest 8.x и структура проекта

Стандарт 2026 года это src-layout, конфигурация в pyproject.toml и явная регистрация маркеров через --strict-markers. Звучит как бюрократия, но экономит часы на разборе, почему тест не запустился. См. официальную документацию pytest для полного списка опций.

[tool.pytest.ini_options]
minversion = "8.0"
addopts = [
    "-ra",
    "--strict-markers",
    "--strict-config",
    "--cov=src",
    "--cov-report=term-missing",
]
testpaths = ["tests"]
markers = [
    "unit: быстрые тесты без внешних зависимостей",
    "integration: требуют Docker и testcontainers",
    "e2e: полный пайплайн, медленно",
    "slow: занимает больше 5 секунд",
]

Дальше идёт conftest.py в корне tests/ с общими фикстурами. Тут же настраиваю seed для воспроизводимости и убиваю недетерминированность numpy-случайности.

import numpy as np
import pandas as pd
import pytest

@pytest.fixture(autouse=True)
def _fix_random_seed():
    np.random.seed(42)

@pytest.fixture(scope="session")
def sample_orders() -> pd.DataFrame:
    return pd.DataFrame({
        "order_id": [1, 2, 3, 4],
        "user_id": [10, 10, 11, 12],
        "amount": [100.0, 250.5, 0.0, None],
        "created_at": pd.to_datetime(
            ["2026-01-01", "2026-01-01", "2026-01-02", "2026-01-03"]
        ),
    })

Скоуп фикстуры это не косметика. session нужен для тяжёлых объектов (Docker-контейнер), function для мутируемых DataFrame, иначе один тест испортит данные другому, и вы будете гоняться за флаки-тестами неделями.

Как писать unit-тесты для трансформаций pandas и Polars

Unit-тест трансформации это функция, которая принимает маленький DataFrame с известным содержимым и проверяет, что на выходе ровно то, что ожидается. Ключевое слово тут «маленький»: 3–10 строк, ноль внешних зависимостей, никакого чтения из S3 или Postgres. Если тест требует настоящий файл, это уже интеграционный тест.

import pandas as pd
import pandas.testing as pdt
import pytest
from src.transforms.orders import aggregate_user_revenue

@pytest.mark.unit
def test_aggregate_user_revenue_ignores_nan(sample_orders):
    result = aggregate_user_revenue(sample_orders)

    expected = pd.DataFrame({
        "user_id": [10, 11, 12],
        "revenue": [350.5, 0.0, 0.0],
        "orders_count": [2, 1, 1],
    })
    pdt.assert_frame_equal(
        result.sort_values("user_id").reset_index(drop=True),
        expected,
        check_dtype=False,
    )

@pytest.mark.unit
def test_aggregate_user_revenue_empty_input():
    empty = pd.DataFrame(columns=["order_id", "user_id", "amount", "created_at"])
    result = aggregate_user_revenue(empty)
    assert result.empty
    assert list(result.columns) == ["user_id", "revenue", "orders_count"]

Обратите внимание на второй тест, тот, что про пустой DataFrame. Это как раз тот край, который отваливается в проде через месяц, когда кто-то перестал отправлять события в Kafka на выходные. Я бы советовал всегда писать отдельный тест для пустого входа, тест для дубликатов и тест для NaN. Это три сценария, где падает больше всего пайплайнов на моей памяти.

Для Polars-трансформаций подход тот же, только через polars.testing.assert_frame_equal и pl.DataFrame. Polars даёт бонус: LazyFrame можно тестировать декларативно, проверяя план оптимизации без реального выполнения.

Интеграционные тесты с testcontainers и Docker

Интеграционные тесты нужны, когда логика зависит от конкретной СУБД: специфический SQL, миграции Alembic, констрейнты внешних ключей, транзакционная семантика. Раньше это делалось через SQLite в памяти, что приводило к классике жанра: тесты зелёные, прод падает, потому что PostgreSQL иначе обрабатывает пустые строки в UNIQUE-констрейнтах. Решение простое: testcontainers-python, который поднимает настоящий Postgres в Docker на время сессии тестов.

import pytest
from sqlalchemy import create_engine, text
from testcontainers.postgres import PostgresContainer

@pytest.fixture(scope="session")
def pg_engine():
    with PostgresContainer("postgres:16-alpine") as pg:
        engine = create_engine(pg.get_connection_url())
        with engine.begin() as conn:
            conn.execute(text(open("migrations/001_init.sql").read()))
        yield engine

@pytest.fixture
def clean_pg(pg_engine):
    with pg_engine.begin() as conn:
        conn.execute(text("TRUNCATE orders, users RESTART IDENTITY CASCADE"))
    yield pg_engine

@pytest.mark.integration
def test_load_orders_idempotent(clean_pg, sample_orders):
    from src.loaders.postgres import load_orders

    load_orders(clean_pg, sample_orders)
    load_orders(clean_pg, sample_orders)  # повторный запуск, тот же результат

    with clean_pg.connect() as conn:
        count = conn.execute(text("SELECT COUNT(*) FROM orders")).scalar()
    assert count == len(sample_orders)

Тест идемпотентности это единственный способ убедиться, что бэкфил не создаст дубликатов. Я запускаю пайплайн дважды подряд и проверяю, что состояние базы одинаковое. Если этот тест падает, не мержим PR, точка. Идемпотентность важнее производительности, потому что производительность можно улучшить, а поломанные данные восстановить получится только из бэкапов (если они у вас вообще есть).

Как правильно замокать внешний API

Внешние API это единственное место, где моки безусловно уместны. Хиты по чужому rate-limit в CI это плохая идея, тесты становятся флаки, а вы платите за квоту. Стандарт де-факто это pytest-mock или responses для requests и respx для httpx.

import httpx
import pytest
import respx
from src.extractors.stripe import fetch_charges

@pytest.mark.unit
@respx.mock
def test_fetch_charges_paginates():
    respx.get("https://api.stripe.com/v1/charges").mock(
        side_effect=[
            httpx.Response(200, json={"data": [{"id": "ch_1"}], "has_more": True}),
            httpx.Response(200, json={"data": [{"id": "ch_2"}], "has_more": False}),
        ]
    )
    charges = list(fetch_charges(api_key="sk_test_x"))
    assert [c["id"] for c in charges] == ["ch_1", "ch_2"]

Одно важное «но»: мок должен точно повторять форму ответа реального API. Иначе получите ту же болезнь, что и с моками БД, когда тесты зелёные, а прод падает. Я держу в репозитории фикстуры с реальными (обезличенными) ответами API, снятыми через vcrpy или руками, и перегенерирую их раз в квартал.

Тестирование dbt-моделей вместе с pytest

dbt и pytest не конкурируют, они дополняют друг друга. dbt-tests живут внутри dbt-проекта и проверяют инварианты SQL-моделей: unique, not_null, relationships, custom singular tests. Pytest покрывает всё вокруг: Python-хелперы, макросы через dbt-unit-testing, оркестрацию и связку с Airflow. Актуальный референс это официальная документация dbt по data tests.

# tests/dbt/test_dbt_project.py
import subprocess
import pytest

@pytest.mark.integration
def test_dbt_build_passes(clean_pg, monkeypatch):
    monkeypatch.setenv("DBT_PROFILES_DIR", "tests/dbt/profiles")
    monkeypatch.setenv("DBT_TARGET_URL", str(clean_pg.url))

    result = subprocess.run(
        ["dbt", "build", "--select", "orders_daily+"],
        capture_output=True,
        text=True,
    )
    assert result.returncode == 0, result.stdout + result.stderr

Ключевой приём это прогнать dbt build на testcontainer-Postgres с крошечным сидированным сэмплом. Так вы ловите миграции моделей, ломаные ссылки в ref() и упавшие тесты dbt как единый шаг CI. Отдельные dbt-тесты в изолированной Snowflake-среде тоже полезны, но локально они не запускаются, поэтому у Python-инженеров нет короткого цикла обратной связи.

Валидация схем данных с Pandera 0.29 в тестах

Pandera 0.29, вышедшая в январе 2026, окончательно превратилась из «либы для валидации DataFrame» в универсальный слой контрактов данных. Одна схема работает поверх pandas, Polars, PySpark, Dask и Ibis, и её же можно использовать в pytest-фикстурах для генерации мок-данных через hypothesis.

import pandera.pandas as pa
from pandera.typing import DataFrame, Series

class OrderSchema(pa.DataFrameModel):
    order_id: Series[int] = pa.Field(ge=1, unique=True)
    user_id: Series[int] = pa.Field(ge=1)
    amount: Series[float] = pa.Field(ge=0, nullable=True)
    created_at: Series[pa.DateTime]

    class Config:
        strict = True

@pa.check_types
def aggregate_user_revenue(df: DataFrame[OrderSchema]) -> pd.DataFrame:
    return (
        df.assign(amount=df["amount"].fillna(0))
          .groupby("user_id", as_index=False)
          .agg(revenue=("amount", "sum"), orders_count=("order_id", "count"))
    )

Декоратор @pa.check_types валидирует входы и выходы в рантайме. В тестах я использую тот же класс как источник правды: если фикстура собрана через OrderSchema.example(), невозможно случайно проверить трансформацию на данных, которых в реальности не бывает. Если хотите разобрать похожий паттерн подробнее, посмотрите разбор scikit-learn Pipeline и ColumnTransformer, где тот же принцип «схема как контракт» применяется к ML-препроцессингу.

Параметризация и hypothesis: находим краевые случаи

Параметризация это способ превратить 12 копипаст-тестов в один читаемый тест с таблицей входов. Это база pytest и первый шаг перед переходом на property-based testing.

import pytest
from src.transforms.currency import normalize_amount

@pytest.mark.parametrize("raw,currency,expected_usd", [
    ("100.00", "USD", 100.00),
    ("100,00", "EUR", 108.20),
    ("10 000", "JPY", 67.30),
    ("0", "USD", 0.0),
    ("-50", "USD", -50.0),
])
def test_normalize_amount(raw, currency, expected_usd):
    assert normalize_amount(raw, currency) == pytest.approx(expected_usd, rel=1e-2)

Дальше идут property-based тесты через hypothesis. Библиотека сама придумывает входные данные и ищет комбинации, на которых функция падает. Для трансформаций данных особенно полезна hypothesis-jsonschema и интеграция с Pandera: OrderSchema.strategy() генерирует валидные DataFrame, а вы проверяете инвариант «после агрегации сумма выручки не меняется».

from hypothesis import given, settings
import pandera.pandas as pa

@given(OrderSchema.strategy(size=50))
@settings(max_examples=25, deadline=None)
def test_aggregate_preserves_total_revenue(df):
    original_total = df["amount"].fillna(0).sum()
    aggregated = aggregate_user_revenue(df)
    assert aggregated["revenue"].sum() == pytest.approx(original_total)

Честно говоря, за одну неделю такие тесты нашли у нас три бага, которые вручную никто бы не придумал: NaN в group-by ключе, отрицательные суммы после конвертации валюты и переполнение int32 на суммарной выручке большого клиента.

Как встроить всё это в CI/CD и не сойти с ума

Тесты, которые не запускаются автоматически, это не тесты, а комментарии. Минимальный набор для GitHub Actions такой: unit-тесты на каждый push, интеграционные тесты на PR в main, e2e по nightly-расписанию. Разделение по маркерам pytest даёт эту логику бесплатно.

# .github/workflows/tests.yml
jobs:
  unit:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: {python-version: "3.12"}
      - run: pip install -e ".[test]"
      - run: pytest -m unit -n auto --cov-fail-under=85

  integration:
    if: github.event_name == 'pull_request'
    runs-on: ubuntu-latest
    services:
      docker:
        image: docker:24-dind
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: {python-version: "3.12"}
      - run: pip install -e ".[test]"
      - run: pytest -m integration --maxfail=3

Флаг -n auto у pytest-xdist распараллеливает unit-тесты по всем ядрам runner-а. На среднем пайплайне это сокращает время с 90 до 20 секунд. Интеграционные не параллелю, потому что они делят один Docker daemon и мешают друг другу.

Плюс coverage-gate через --cov-fail-under=85. Не гонитесь за 100%: оставшиеся 15% часто уходят на if __name__ == "__main__", edge-cases в CLI-парсерах и защитные raise NotImplementedError. Порог 85% реалистичен, а до 100% люди обычно докручивают моками ради галочки, что деградирует качество тестов.

Отдельно рекомендую подключить pytest-xdist, pytest-timeout (защита от зависших тестов) и pytest-benchmark для регрессионных проверок производительности критичных трансформаций. При работе с SQL-аналитикой посмотрите также разбор DuckDB в Python. Он отлично подходит как in-process движок для интеграционных тестов, где не нужен полноценный Postgres.

Часто задаваемые вопросы

Нужно ли мокать базу данных в тестах пайплайнов?

Нет, для интеграционных тестов лучше поднимать реальную БД через testcontainers. Моки БД пропускают баги миграций, констрейнтов и типов, которые PostgreSQL и SQLite обрабатывают по-разному. Моки уместны только для внешних HTTP-API, где реальные запросы медленны и стоят денег.

В чём разница между unit- и интеграционным тестом ETL-пайплайна?

Unit-тест проверяет одну функцию-трансформацию в изоляции на in-memory DataFrame и выполняется за миллисекунды. Интеграционный тест поднимает реальные внешние сервисы (БД, брокер сообщений) через testcontainers и проверяет связку между слоями. Оба нужны: unit находят логические баги, интеграционные ловят расхождения контрактов.

Как тестировать dbt-модели вместе с Python-кодом?

Запускайте dbt build на testcontainer-Postgres или DuckDB как отдельный pytest-тест с маркером integration. Это даёт единый CI-шаг и короткий цикл обратной связи для Python-инженеров. Отдельные dbt-tests в Snowflake-таргете дополняют, но не заменяют локальный прогон.

Какое покрытие тестами считается достаточным для ETL-пайплайна?

Порог 80–85% по pytest-cov это рабочий минимум. Более высокие цифры обычно достигаются формальными моками, которые не проверяют реальную логику. Важнее не процент, а покрытие критичных краёв: пустой вход, дубликаты, NaN, идемпотентность повторного запуска.

Стоит ли использовать hypothesis для тестов данных?

Да, особенно в связке с Pandera 0.29. Property-based тесты автоматически генерируют DataFrame по схеме и находят краевые случаи, которые невозможно предугадать вручную: NaN в join-ключах, переполнения, некорректные типы. Начните с 20–30 примеров на тест и увеличивайте по мере необходимости.