Когда тестов становится больше десяти

В прошлом уроке мы написали десяток тестов на парсер. Каждый тест жил сам по себе: parse_order("1,2,3.0,USD") — и проверяем результат. Никаких настроек, никакой подготовки.

В реальной жизни такие чистые случаи редки. Обычно перед каждым тестом надо:

• создать временный файл с тестовыми данными;
• открыть соединение к in-memory SQLite;
• замокать переменную окружения;
• инициализировать клиент к API с фейковым токеном.

И всё это много раз в разных тестах. Если копипастить setup в каждый тест — у вас получится 200 строк дублирующегося кода. Если впихнуть в один большой тест — он будет проверять десять вещей, и при падении вы не поймёте, что именно сломалось.

Pytest решает это

Первая фикстура

Фикстура — это функция с декоратором @pytest.fixture. Pytest вызывает её и подставляет результат в любой тест, который попросил аргумент с таким же именем:

import pytest
from my_etl.parser import parse_order


@pytest.fixture
def sample_line() -> str:
    return "1001,42,199.99,USD"


def test_parses_id(sample_line: str) -> None:
    order = parse_order(sample_line)
    assert order.order_id == 1001


def test_parses_currency(sample_line: str) -> None:
    order = parse_order(sample_line)
    assert order.currency == "USD"

Что произошло:

1. pytest увидел, что test_parses_id хочет аргумент sample_line.
2. Нашёл фикстуру sample_line в этом же файле.
3. Вызвал её, получил "1001,42,199.99,USD".
4. Передал в тест.

Это называется

Полезно тем, что:

• Одну фикстуру переиспользует много тестов.
• Тест не знает, как она создаётся — это можно поменять в одном месте.
• Фикстуры могут зависеть друг от друга (фикстура «БД» зависит от фикстуры «config», и pytest сам разрулит порядок).

Фикстуры с setup/teardown через yield

Часто после теста надо прибраться: закрыть соединение, удалить временный файл, остановить mock-сервер. Для этого фикстура использует yield вместо return:

import sqlite3
import pytest
from collections.abc import Iterator


@pytest.fixture
def in_memory_db() -> Iterator[sqlite3.Connection]:
    conn = sqlite3.connect(":memory:")
    conn.execute("CREATE TABLE orders (id INTEGER PRIMARY KEY, amount REAL)")
    yield conn       # сюда уходит тест
    conn.close()     # выполнится после теста, даже если он упал


def test_insert_order(in_memory_db: sqlite3.Connection) -> None:
    in_memory_db.execute("INSERT INTO orders (id, amount) VALUES (1, 99.99)")
    cursor = in_memory_db.execute("SELECT COUNT(*) FROM orders")
    assert cursor.fetchone()[0] == 1

yield conn приостанавливает фикстуру, отдаёт conn тесту, ждёт окончания теста, потом продолжает — закрывает коннект. Это тот же паттерн, что и контекст-менеджеры из урока 02 модуля 4.

Главное — conn.close() выполнится даже если тест упадёт с исключением. Это гарантирует чистоту: незакрытых соединений не остаётся.

Scope: как часто создавать фикстуру

По умолчанию фикстура пересоздаётся на каждый тест. Это безопасно (тесты изолированы), но дорого, если setup тяжёлый — поднять docker-контейнер с PostgreSQL за 5 секунд десять раз — это уже минута на пустом месте.

Параметр scope управляет временем жизни:

@pytest.fixture(scope="function")  # default — пересоздаётся для каждого теста
def per_test_data(): ...

@pytest.fixture(scope="module")    # одна на весь файл
def shared_client(): ...

@pytest.fixture(scope="session")   # одна на весь run pytest
def docker_postgres(): ...

conftest.py: общие фикстуры

Фикстуры, объявленные в test_*.py, видны только в этом файле. Чтобы фикстура была доступна в нескольких файлах — кладём её в conftest.py:

# tests/conftest.py
import sqlite3
import pytest
from collections.abc import Iterator


@pytest.fixture
def in_memory_db() -> Iterator[sqlite3.Connection]:
    conn = sqlite3.connect(":memory:")
    conn.executescript("""
        CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT);
        CREATE TABLE orders (id INTEGER PRIMARY KEY, user_id INTEGER, amount REAL);
    """)
    yield conn
    conn.close()

Теперь любой test_*.py в tests/ и его подпапках может попросить in_memory_db без импорта. Pytest найдёт её сам.

conftest.py можно класть рекурсивно: tests/conftest.py — общий, tests/loader/conftest.py — только для тестов в tests/loader/.

Built-in фикстуры

Pytest даёт несколько фикстур из коробки. Самые полезные для DE:

tmp_path

tmp_path — это

import csv
from pathlib import Path
from my_etl.parser import parse_csv_file


def test_parses_csv(tmp_path: Path) -> None:
    file = tmp_path / "orders.csv"
    file.write_text("id,amount\n1,99.99\n2,42.00\n", encoding="utf-8")

    result = parse_csv_file(file)

    assert len(result) == 2
    assert result[0].amount == 99.99

Никаких tempfile.mkdtemp() + try/finally rmtree — pytest всё делает сам.

capsys

capsys ловит вывод в stdout/stderr — полезно для тестов CLI:

def greet(name: str) -> None:
    print(f"Hello, {name}!")


def test_greet(capsys) -> None:
    greet("Anna")
    captured = capsys.readouterr()
    assert captured.out == "Hello, Anna!\n"
    assert captured.err == ""

monkeypatch

monkeypatch — точечно подменяет атрибуты, методы, переменные окружения только на время теста, потом откатывает:

import os


def test_uses_env_token(monkeypatch) -> None:
    monkeypatch.setenv("API_TOKEN", "fake-token")
    assert os.environ["API_TOKEN"] == "fake-token"


# После теста os.environ["API_TOKEN"] восстановлено к прежнему значению.

Подробнее про monkeypatch — в уроке 03, где разбираем мокирование.

caplog

caplog ловит сообщения logging:

import logging


def test_logs_warning(caplog) -> None:
    with caplog.at_level(logging.WARNING):
        my_function_that_warns()
    assert "skipping bad record" in caplog.text

parametrize: один тест, много входов

Допустим, у нас 8 разных входов в parse_int и для каждого свой ожидаемый результат. Без parametrize это было бы 8 почти одинаковых функций:

def test_parse_int_42() -> None:
    assert parse_int("42") == 42

def test_parse_int_negative() -> None:
    assert parse_int("-7") == -7

# ... ещё 6 таких

С

import pytest
from my_etl.parser import parse_int


@pytest.mark.parametrize(
    ("text", "expected"),
    [
        ("42", 42),
        ("  42  ", 42),
        ("-7", -7),
        ("0", 0),
        ("", None),
        ("   ", None),
        ("abc", None),
        ("3.14", None),
    ],
)
def test_parse_int(text: str, expected: int | None) -> None:
    assert parse_int(text) == expected

При запуске pytest -v увидите 8 отдельных тестов с собственными ID:

test_parser.py::test_parse_int[42-42] PASSED
test_parser.py::test_parse_int[  42  -42] PASSED
test_parser.py::test_parse_int[-7--7] PASSED
test_parser.py::test_parse_int[0-0] PASSED
test_parser.py::test_parse_int[-None] PASSED
test_parser.py::test_parse_int[   -None] PASSED
test_parser.py::test_parse_int[abc-None] PASSED
test_parser.py::test_parse_int[3.14-None] PASSED

ID составляется из значений автоматически. Если упадёт — точно видно, какой именно input сломал тест.

Читаемые IDs через ids=

Когда input — длинная строка или непечатаемый объект, авто-ID становится мусором. Передаём свои:

@pytest.mark.parametrize(
    ("text", "expected"),
    [
        ("42", 42),
        ("  42  ", 42),
        ("", None),
        ("abc", None),
    ],
    ids=["plain", "with-spaces", "empty", "garbage"],
)
def test_parse_int(text: str, expected: int | None) -> None:
    assert parse_int(text) == expected

Теперь имена тестов читаемые: test_parse_int[plain], test_parse_int[empty].

Cartesian product через несколько parametrize

Если у вас два независимых параметра — parametrize можно стакать. Pytest сделает декартово произведение:

@pytest.mark.parametrize("currency", ["USD", "EUR", "RUB"])
@pytest.mark.parametrize("amount", [0.01, 100.0, 1_000_000.0])
def test_format_amount(currency: str, amount: float) -> None:
    result = format_amount(amount, currency)
    assert currency in result
    assert f"{amount:.2f}" in result

Получится 3 × 3 = 9 тестов. Удобно, когда комбинаций много, но логика — одна.

Параметризация фикстур

Сами фикстуры тоже можно параметризовать — каждый их вариант пойдёт во все зависящие тесты:

@pytest.fixture(params=["sqlite", "duckdb"])
def db_backend(request) -> str:
    return request.param


def test_load(db_backend: str) -> None:
    # этот тест выполнится 2 раза: для sqlite и для duckdb
    ...

Это нужно реже, чем parametrize-декоратор, но удобно когда хочется прогнать всю серию тестов по разным «средам».

DE-кейс: фикстура с реалистичной БД

Соберём то, что junior DE напишет в первую неделю — фикстуру с in-memory SQLite, в которой уже есть схема и пара записей. Используем её для тестов аналитических функций.

# tests/conftest.py
import sqlite3
import pytest
from collections.abc import Iterator


@pytest.fixture
def db() -> Iterator[sqlite3.Connection]:
    """In-memory SQLite с типичной DE-схемой и сидом тестовых данных."""
    conn = sqlite3.connect(":memory:")
    conn.row_factory = sqlite3.Row
    conn.executescript(
        """
        CREATE TABLE users (
            id INTEGER PRIMARY KEY,
            name TEXT NOT NULL,
            country TEXT NOT NULL
        );

        CREATE TABLE orders (
            id INTEGER PRIMARY KEY,
            user_id INTEGER NOT NULL REFERENCES users(id),
            amount REAL NOT NULL,
            currency TEXT NOT NULL,
            created_at TEXT NOT NULL
        );

        INSERT INTO users (id, name, country) VALUES
            (1, 'Anna', 'RU'),
            (2, 'Bob', 'US'),
            (3, 'Carol', 'RU');

        INSERT INTO orders (id, user_id, amount, currency, created_at) VALUES
            (101, 1, 100.0, 'USD', '2026-01-15'),
            (102, 1, 50.0,  'USD', '2026-01-20'),
            (103, 2, 200.0, 'EUR', '2026-01-15'),
            (104, 3, 75.0,  'USD', '2026-02-01');
        """
    )
    conn.commit()
    yield conn
    conn.close()

Тесты:

# tests/test_analytics.py
import pytest
import sqlite3
from my_etl.analytics import total_by_country, top_users


class TestTotalByCountry:
    def test_groups_correctly(self, db: sqlite3.Connection) -> None:
        result = total_by_country(db)
        assert result == {"RU": 225.0, "US": 200.0}

    def test_empty_db(self, db: sqlite3.Connection) -> None:
        db.executescript("DELETE FROM orders; DELETE FROM users;")
        assert total_by_country(db) == {}


class TestTopUsers:
    @pytest.mark.parametrize(
        ("limit", "expected_first"),
        [
            (1, "Anna"),
            (3, "Anna"),
            (10, "Anna"),
        ],
    )
    def test_top_user_is_anna(
        self, db: sqlite3.Connection, limit: int, expected_first: str
    ) -> None:
        rows = top_users(db, limit=limit)
        assert rows[0]["name"] == expected_first

Каждый тест получает свежую in-memory БД (scope=function), а сама фикстура живёт в conftest.py — переиспользуется во всех тестовых файлах.

Что мы получили

• Фикстура — функция с @pytest.fixture. Pytest передаёт результат в тесты по имени аргумента.
• yield-фикстура — setup/teardown как контекст-менеджер. Cleanup выполнится даже если тест упал.
• scope: function (default), class, module, session. Шире = быстрее, но больше шанс side-effect’ов между тестами.
• conftest.py — общие фикстуры для соседних файлов, без импорта.
• Встроенные фикстуры: tmp_path, capsys, monkeypatch, caplog.
• @pytest.mark.parametrize — один тест, таблица входов. ids= для читаемых имён.
• Стакающиеся parametrize — cartesian product, но не злоупотреблять.

В следующем уроке — мокирование: как тестировать функцию, которая ходит в API или БД, без реального API и БД.