← назад к разделу

Модульные тесты проверяют логику в изоляции, но не говорят, правильно ли сервис работает с базой, брокером или соседним сервисом. Для этого нужны интеграционные тесты — с настоящими зависимостями. В Python для них достаточно pytest, фикстуры на всю сессию и библиотеки testcontainers, которая поднимает PostgreSQL в Docker прямо из теста.

Обязательно

Проблема моков и SQLite

Самый быстрый способ протестировать репозиторий — подменить базу. В Python для этого есть два соблазна, и оба обманывают одинаково.

Мок репозитория через MagicMock или monkeypatch: тест обработчика проходит, но сам SQL ни разу не выполнился. Запрос с опечаткой в имени колонки, неверный JOIN, забытый WHERE deleted_at IS NULL — всё это тест не видит, потому что репозиторий в нём ненастоящий.

SQLite вместо PostgreSQL выглядит честнее: SQLAlchemy один и тот же, sqlite:// поднимается без Docker за миллисекунду. Но это другая база. В ней нет jsonb, типов timestamptz и массивов, иначе работают ON CONFLICT и RETURNING, не проверяются внешние ключи без отдельной настройки, а транзакционная изоляция устроена по-своему. Запрос, который SQLAlchemy собрал для SQLite, проходит, а на PostgreSQL падает на первом же DISTINCT ON. Тест зелёный ровно до прода.

Интеграционный тест поднимает тот же PostgreSQL, что работает в проде, и убирает целый класс ошибок: запрос действительно выполняется, строки действительно раскладываются по полям модели, UNIQUE действительно не даёт вставить дубль.

testcontainers: PostgreSQL один раз на сессию

testcontainers запускает Docker-контейнер из кода теста. Для PostgreSQL есть готовый класс, и он надёжнее общего DockerContainer: ждёт не открытия порта, а ответа базы на select version() через psql внутри контейнера — образ при первом старте поднимается дважды, и ожидание по порту иногда пускает тест раньше времени.

сквозные: TestClient до базы и обратно единицы, критичные сценарии интеграционные: Testcontainers PostgreSQL десятки, секунды, Docker модульные: домен без инфраструктуры сотни, миллисекунды

Контейнер поднимают один раз на сессию и держат для десятков интеграционных тестов; быстрых модульных тестов в разы больше, сквозных единицы.

Контейнер стоит секунды, поэтому его поднимают один раз на сессию pytest фикстурой в conftest.py, а не в каждом тесте:

import pytest
import sqlalchemy as sa
from alembic import command
from alembic.config import Config
from testcontainers.community.postgres import PostgresContainer


@pytest.fixture(scope="session")
def database_url() -> str:
    with PostgresContainer("postgres:16-alpine", driver="psycopg") as postgres:
        yield postgres.get_connection_url()


@pytest.fixture(scope="session")
def engine(database_url: str):
    config = Config("alembic.ini")
    config.set_main_option("sqlalchemy.url", database_url)
    command.upgrade(config, "head")
    engine = sa.create_engine(database_url)
    yield engine
    engine.dispose()

Фикстура с scope="session" выполняется один раз на прогон: до первого теста, который её попросил, поднимает контейнер, после последнего — гасит, потому что with закрывается при выходе из генератора. get_connection_url() возвращает строку вида postgresql+psycopg://test:test@localhost:49843/test: порт случайный, его всегда спрашивают у контейнера, а не пишут в настройки.

Две оговорки. Первая: контейнер один на процесс pytest, и при параллельном прогоне через pytest-xdist у восьми рабочих процессов будет восемь контейнеров — это восемь стартов и восемь баз. Выход — держать интеграционные тесты в немногих модулях или поднимать базу снаружи и отдавать адрес через переменную среды: фикстура сначала смотрит в TEST_DATABASE_URL, и только если её нет, стартует контейнер. Вторая: контейнер не удаляется, если процесс убили сигналом до выхода из with. За уборкой следит сторожевой контейнер Ryuk, который testcontainers поднимает сам: он гасит всё, что осталось после завершения процесса.

Откуда в пустом контейнере таблицы

Контейнер поднимается с пустой базой: в образе postgres:16-alpine нет ни одной вашей таблицы. Кто их создаёт?

Те же миграции, что и в проде. В Python миграции не запускаются «при старте приложения» сами по себе: их запускает alembic upgrade head в конвейере выката или команда сервиса. В тестах то же самое — накатить миграции в фикстуре сразу после подъёма контейнера, как в примере выше: command.upgrade(config, "head") делает ровно то, что команда в терминале, только с адресом тестовой базы.

Самая дорогая ловушка в этом месте — Base.metadata.create_all(engine) «для скорости». Он создаёт таблицы из моделей SQLAlchemy, а не из миграций: индекс, добавленный только в миграции, ограничение CHECK, написанное руками, значение по умолчанию на стороне базы — всего этого в тестовой схеме не будет. Через месяц модели и миграции расходятся, тесты зелёные, а прод падает на колонке, которой в миграции нет. Схема у тестов одна — та, что накатывают миграции, а расхождение моделей с миграциями ловит alembic check в сборке.

Справочные данные (статусы, типы, тарифы), которые в проде вставляет миграция, появятся в тестовой базе сами и должны там остаться. Это важно, когда дойдёте до очистки данных между тестами.

Изоляция между тестами при одной базе

Контейнер один на сессию — значит, база одна на все тесты. Первый тест создал заказ, второй посчитал заказы и получил не ту цифру; третий проходит один и падает в наборе. Способов изолировать три, и границы у них разные.

Откат транзакции. Тест открывает соединение и внешнюю транзакцию, отдаёт коду сессию, привязанную к этому соединению, и откатывает всё в конце:

from sqlalchemy.orm import Session


@pytest.fixture
def session(engine):
    with engine.connect() as connection:
        transaction = connection.begin()
        session = Session(bind=connection, join_transaction_mode="create_savepoint")
        yield session
        session.close()
        transaction.rollback()

def test_order_repo_save(session):
    repo = OrderRepository(session)

    order_id = repo.save(Order.new(customer_id="cust-1", total=500))
    session.commit()

    assert repo.by_id(order_id).customer_id == "cust-1"

join_transaction_mode="create_savepoint" — то, что делает приём рабочим: session.commit() внутри кода превращается в освобождение точки сохранения, а не в настоящую фиксацию, и внешний rollback в конце теста убирает всё. Быстро и ничего не надо убирать. Ограничений три. Не проверяется поведение при фиксации: отложенные ограничения и всё, что происходит в момент настоящего COMMIT, не случается вовсе. Не работает для кода, который берёт соединение сам — фоновая задача со своим engine, второй сервис, — его данные остаются в базе. И такие тесты нельзя пускать параллельно в одном процессе на одном соединении.

Очистка таблиц. Честный способ для всего, что проверяет фиксацию или проходит мимо тестовой сессии: тест пишет по-настоящему, а помощник возвращает базу в исходное состояние одним запросом:

REFERENCE_TABLES = ("alembic_version", "order_status_ref")


@pytest.fixture
def clean_database(engine):
    with engine.begin() as connection:
        tables = connection.execute(sa.text(
            "SELECT quote_ident(tablename) FROM pg_tables "
            "WHERE schemaname = 'public' AND tablename <> ALL(:skip)"
        ), {"skip": list(REFERENCE_TABLES)}).scalars().all()
        connection.execute(sa.text(f"TRUNCATE TABLE {', '.join(tables)} RESTART IDENTITY CASCADE"))

Все таблицы в одном TRUNCATE, тогда внешние ключи не мешают; RESTART IDENTITY сбрасывает счётчики; таблица версий Alembic и справочники исключены, иначе следующий тест не найдёт того, что в проде есть всегда. Список таблиц берётся из базы, и новая таблица попадает в уборку сама.

Уникальные данные. Ничего не удалять, а делать каждый тест независимым по данным: свой uuid4() на покупателя, свой номер заказа. Порядок не важен, соседи не мешают, и это единственный способ, который выживает при pytest-xdist на одной базе. Цена: любая проверка «а сколько всего записей» становится неверной, считать надо всегда с условием по своему ключу.

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

Сквозной тест: HTTP до базы и обратно

В Python для сквозного теста не нужен ни сервер, ни порт: приложение FastAPI вызывается через ASGITransport, а сессию базы подставляют через dependency_overrides — ту же зависимость, что в проде даёт сессию из пула:

import httpx
import pytest

from app.main import app, get_session


@pytest.fixture
async def client(session):
    app.dependency_overrides[get_session] = lambda: session
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        yield client
    app.dependency_overrides.clear()


async def test_create_order(client, session):
    response = await client.post(
        "/api/v1/orders",
        json={"customerId": "cust-1", "lines": [{"sku": "A1", "qty": 2}]},
    )

    assert response.status_code == 201, response.text
    assert session.execute(sa.text("SELECT count(*) FROM orders")).scalar() == 1

Такой тест проверяет ровно то, чего не видит ни мок, ни тест репозитория по отдельности: разбор тела, валидацию, обработчик, транзакцию в сценарии и то, что в базе действительно появилась строка. Проверять результат нужно в базе, а не только по коду ответа: 201 с пустой таблицей — классическая ошибка сценария, который открыл транзакцию и не зафиксировал. Благодаря dependency_overrides даже сквозной тест работает с откатом: обработчик получает ту же сессию на точке сохранения, что и тест. Это отличие от стеков, где обработчик берёт соединение из пула сам, и там для HTTP-тестов остаётся только очистка.

Не только PostgreSQL: Kafka, Redis, свои образы

testcontainers — библиотека не про базу, а про «подними мне что угодно из образа и дай адрес». Три сценария, которые нужны почти всем.

Брокер. Готовый KafkaContainer().with_kraft() поднимает брокер в режиме KRaft, без ZooKeeper, и отдаёт адрес через get_bootstrap_server(). Старт занимает секунды, поэтому его тоже держат на сессию. Проверка обязательно с ожиданием, а не следующей строкой: сообщение доходит до потребителя асинхронно, и assert len(received) == 1 сразу после записи будет мигать. Готового помощника вроде Eventually в pytest нет, пишут свой: опрос условия в цикле с asyncio.sleep(0.05) и пределом в несколько секунд.

Что угодно из образа. Нет готового класса — берут DockerContainer: заглушка внешнего партнёра, хранилище, совместимое с S3, почтовый ловец:

from testcontainers.core.container import DockerContainer
from testcontainers.core.waiting_utils import wait_for_logs

minio = (
    DockerContainer("minio/minio:RELEASE.2024-10-13T13-34-11Z")
    .with_command("server /data")
    .with_exposed_ports(9000)
)
minio.start()
wait_for_logs(minio, "API:")
endpoint = f"http://{minio.get_container_host_ip()}:{minio.get_exposed_port(9000)}"

Адрес и порт всегда спрашивают у контейнера, а не пишут напрямую: изнутри контейнера порт свой, а снаружи случайный.

Контейнер между прогонами. Переиспользования контейнера между запусками, как в версиях для других языков, у testcontainers для Python нет: каждый pytest поднимает свой. Чтобы не платить стартом локально, базу держат снаружи — контейнер из docker compose для разработки — и отдают её адрес через ту самую переменную TEST_DATABASE_URL. Из этого следует неочевидное требование: внешняя база сохраняет данные между прогонами, поэтому тесты обязаны быть чистыми сами по себе. Тест, который проходит на свежей базе и падает на второй запуск, — это найденная зависимость от пустой базы.

Цена входа: тестам нужен Docker

Про testcontainers говорят «достаточно установленного Docker», и за этим стоит несколько решений, принимаемых один раз на команду.

Агент сборки. Тестам нужен доступ к демону Docker, а агент обычно сам работает в контейнере. Варианты: смонтировать сокет демона хоста, поднять отдельный демон рядом или указать DOCKER_HOST на удалённый — тогда контейнеры поднимаются на выделенной машине. Для облачных агентов у всех крупных систем сборки Docker есть из коробки.

Машина разработчика. Docker Desktop в больших компаниях платный. Заменители работают: Colima и Rancher Desktop на macOS, Podman с сокетом, совместимым с Docker. testcontainers ищет демон по стандартным путям; если не нашёл — путь задаётся переменной DOCKER_HOST. Тем, у кого Docker нет вовсе, интеграционные тесты выключают локально, но не в сборке:

import pytest

pytestmark = pytest.mark.integration

[tool.pytest.ini_options]
markers = ["integration: тесты с контейнерами"]

pytest -m "not integration" такие модули пропустит, pytest -m integration прогонит только их. Второй способ — проверить доступность демона в фикстуре и вызвать pytest.skip("нет Docker"); это удобнее, когда интеграционные и обычные тесты живут в одном модуле.

Закрытый контур. Без выхода в интернет образы не скачаются. Решение — внутреннее зеркало реестра и переменная TESTCONTAINERS_HUB_IMAGE_NAME_PREFIX, которая дописывает его ко всем именам образов; в зеркало кладут и образ Ryuk, иначе «всё скачалось, а тесты не стартуют».

Время и ресурсы. Каждый контейнер — память и секунды старта. Поэтому контейнеры поднимают один раз на сессию, а не на тест, и не поднимают то, что в этом прогоне не нужно.

Когда полная связка, а когда срез

СитуацияЧто использовать
Запросы репозитория, миграции, ограничения базыконтейнер + сессия на точке сохранения с откатом
Сквозной путь HTTP → сценарий → база → ответASGITransport с подменённой сессией или очистка
Разбор тела, валидация, коды ответа обработчикаTestClient + заглушка сценария через dependency_overrides, без базы
Бизнес-правила доменаобычный тест, без всего

Правило то же, что и везде: поднимать ровно столько, сколько нужно. В Python цена лишнего — ещё один контейнер на сессию и лишние секунды в каждом прогоне.

Дополнительно: при первом чтении можно пропустить

Глубже: мигающие тесты: общее состояние, паузы, порты и параллельный прогонрасширенное

Тест, который то зелёный, то красный, разрушает доверие к сборке быстрее, чем отсутствие тестов. Причин немного, и каждая находится.

Общее состояние. Переменная модуля, кэш lru_cache, строка в базе от предыдущего теста. Находят запуском в случайном порядке: pytest-randomly перемешивает тесты при каждом прогоне и печатает зерно (Using --randomly-seed=…), чтобы повторить порядок. Тест, который проходит один (pytest путь::имя) и падает в наборе, зависит от соседей.

Паузы вместо ожиданий. await asyncio.sleep(0.5) перед проверкой асинхронного результата проходит на ноутбуке и падает на загруженном раннере. Ждут условия, а не времени: цикл опроса с пределом, из раздела про брокер. Код с повторами по таймеру тестируют, подставляя в него нулевые задержки, а не дожидаясь настоящих секунд.

Время и случайность. Код, читающий datetime.now() напрямую, падает в полночь и при смене пояса раннера. Время внедряют зависимостью now: Callable[[], datetime] или замораживают time_machine.travel(...); случайность берут из random.Random(seed) с зерном, которое печатают при падении.

Порты и процессы. testcontainers сам берёт свободный порт — свой uvicorn на 8000 в тесте поднимать не надо, для HTTP есть ASGITransport. При pytest-xdist каждый рабочий процесс поднимает свой контейнер и свою базу: это дороже, но тесты не делят данные.

Цикл событий. Асинхронная фикстура на всю сессию (engine для asyncpg) требует, чтобы цикл событий тоже жил всю сессию: в pytest-asyncio это loop_scope="session", иначе первый же тест после смены цикла упадёт с attached to a different loop.

Процесс. Мигающий тест не перезапускают до зелёного через pytest-rerunfailures, а помечают pytest.mark.skip с номером задачи и чинят; сборка считает долю перезапусков, и её рост — тревога.

Глубже: тест миграций: накатить на непустую базурасширенное

Статья дважды говорит, что схему накатывают миграции, и стоит показать, как это проверить. Проверок три.

С нуля. Любой интеграционный тест на контейнере это проверяет заодно: alembic upgrade head на пустой базе падает на первой же сломанной миграции. Отдельно в сборке запускают alembic check: он сравнивает модели с миграциями и падает, если кто-то поменял модель и забыл миграцию.

На непустой базе. Пустая таблица не покажет, что ALTER TABLE держит блокировку минуту, что SET NOT NULL на колонку с существующими NULL невозможен, что перелив данных не идемпотентен. Отдельный шаг конвейера восстанавливает снимок схемы прода с обезличенным образцом данных (или заполняет таблицы генератором до реалистичного размера) и применяет только новые миграции с lock_timeout, замеряя время. Запускают на PR, где менялся каталог alembic/versions.

Старый образ на новой схеме. Во время выката предыдущая версия сервиса работает с новой схемой. Тесты предыдущей версии (образ с прошлым тегом уже есть) прогоняют против базы с новыми миграциями: зелёный прогон означает, что миграция совместима и откат безопасен. Красный — ломающее изменение, которое раскладывают на шаги.

Что тесты не проверят: таймауты на таблицах в миллиард строк, которые больше любого стенда. Для них остаются CONCURRENTLY, пакетный перелив и репетиция на копии.

Коротко

  • Мок репозитория и SQLite проверяют ваши ожидания о SQL, а не SQL: интеграционный тест поднимает тот же PostgreSQL, что в проде, через testcontainers.
  • Контейнер стартует один раз на сессию фикстурой scope="session" в conftest.py; PostgresContainer ждёт настоящей готовности, адрес с случайным портом даёт get_connection_url().
  • Таблицы создают те же миграции, что в проде: command.upgrade(config, "head") сразу после старта контейнера; create_all расходится с миграциями за месяц, расхождение ловит alembic check.
  • Изоляция: сессия на внешнем соединении с join_transaction_mode="create_savepoint" и откатом для репозиториев, TRUNCATE всех таблиц одним запросом для фиксации, уникальные данные для pytest-xdist.
  • Сквозной тест — ASGITransport с сессией через dependency_overrides; результат проверяют в базе, а не только по коду ответа.
  • Kafka и Redis — готовые классы, остальное — DockerContainer с wait_for_logs; порт всегда спрашивают у контейнера; переиспользования между прогонами нет, локально базу держат снаружи через переменную среды.
  • Docker нужен агенту сборки и машине разработчика; без него тесты выключают маркером integration локально, но не в сборке.
  • Мигание лечат по причинам: pytest-randomly находит общее состояние, опрос условия вместо пауз, time_machine и зерно, loop_scope="session" для асинхронных фикстур.
  • Миграции проверяют трижды: с нуля и alembic check, на снимке прода с данными и временем, старым образом против новой схемы.

Что почитать дальше