> Как тестировать миграции базы данных? (Python)

Короткий ответ

Миграции тестируют на трёх уровнях: проверка синтаксиса и применимости (dry-run на пустой и реальной базе), проверка идемпотентности и обратимого отката, и проверка целостности данных после применения (например, что backfill не потерял строки). В Python для этого используют pytest с фикстурами, создающими временную базу (PostgreSQL в Docker или SQLite для простых случаев), и инструменты вроде alembic upgrade head --sql для генерации SQL без применения. Главное - тесты должны гоняться в CI на чистой схеме и на схеме с данными, приближенными к продакшену.

Подробное объяснение

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

Ключевые аспекты:

• Тест на пустой базе: проверяет, что все миграции применяются последовательно от нуля до текущей версии без ошибок. Это ловит синтаксические ошибки, конфликты имён, неправильные внешние ключи.
• Тест на базе с данными: создаёт схему на предыдущей версии, заполняет её тестовыми данными (включая краевые случаи: NULL, дубликаты, длинные строки), затем применяет целевую миграцию. Проверяет, что backfill, изменение типов, переименование колонок не ломают данные.
• Тест отката: применяет миграцию вверх, затем вниз (downgrade), и проверяет, что схема и данные вернулись к исходному состоянию. Для этого нужны корректные upgrade/downgrade функции в Alembic.
• Тест идемпотентности: для повторяемых миграций (например, создание индексов с IF NOT EXISTS) проверяет, что повторное применение не падает и не создаёт дубликатов.
• Тест производительности: для больших таблиц проверяет, что миграция выполняется за приемлемое время (например, добавление индекса на 10 млн строк). Обычно это отдельный интеграционный тест, не входящий в обычный CI.

В Python с Alembic типичный подход - использовать фикстуру pytest, которая:

1. Поднимает временную базу (через testing.postgresql или Docker-контейнер).
2. Применяет все миграции до нужной ревизии.
3. Заполняет данными.
4. Применяет целевую миграцию.
5. Проверяет инварианты через SQL-запросы или ORM.

Для проверки SQL без применения можно использовать alembic upgrade head --sql - это генерирует SQL, который можно проанализировать статически (например, убедиться, что нет DROP TABLE без IF EXISTS).

На практике

В реальном проекте тесты миграций обычно выносят в отдельную директорию tests/migrations/ и запускают не в каждом PR, а в отдельном CI-джобе с доступом к PostgreSQL. Причины: они медленные (создание базы, применение десятков миграций) и требуют специфического окружения.

Рабочий процесс:

• Для каждой новой миграции пишется тест, который проверяет конкретное изменение: например, что колонка status получила правильный default, или что backfill корректно заполнил created_at из id.
• Используется паттерн "миграция вверх до N-1, вставка данных, миграция до N, проверка".
• Для проверки отката - "миграция до N, откат до N-1, проверка схемы".
• Данные для тестов генерируются фабриками (factory_boy) или SQL-скриптами, но обязательно включают краевые случаи.
• Если миграция затрагивает большие объёмы, пишется отдельный нагрузочный тест, который запускается вручную или по расписанию.

Также полезно добавить проверку, что alembic upgrade head на копии продакшен-базы (анонимизированной) проходит без ошибок. Это делается через отдельный скрипт, который запускается перед релизом.

Пример кода

PYTHON
# tests/migrations/test_add_status_column.py
import pytest
from alembic import command
from alembic.config import Config
from sqlalchemy import create_engine, text

@pytest.fixture()
def migrated_db(tmp_path, postgresql):
    """Применяет миграции до ревизии 'abc123' и возвращает engine."""
    cfg = Config("alembic.ini")
    cfg.set_main_option("sqlalchemy.url", postgresql.url)
    command.upgrade(cfg, "abc123")  # предыдущая ревизия
    engine = create_engine(postgresql.url)
    yield engine
    engine.dispose()

def test_add_status_column_backfill(migrated_db):
    # вставляем данные до миграции
    with migrated_db.begin() as conn:
        conn.execute(text("INSERT INTO orders (id, amount) VALUES (1, 100), (2, 200)"))

    # применяем целевую миграцию
    cfg = Config("alembic.ini")
    cfg.set_main_option("sqlalchemy.url", migrated_db.url)
    command.upgrade(cfg, "def456")

    # проверяем результат
    with migrated_db.connect() as conn:
        rows = conn.execute(text("SELECT id, status FROM orders ORDER BY id")).fetchall()
        assert rows == [(1, "pending"), (2, "pending")]

def test_downgrade_removes_column(migrated_db):
    cfg = Config("alembic.ini")
    cfg.set_main_option("sqlalchemy.url", migrated_db.url)
    command.upgrade(cfg, "def456")
    command.downgrade(cfg, "abc123")

    with migrated_db.connect() as conn:
        # колонки status больше нет
        cols = conn.execute(text(
            "SELECT column_name FROM information_schema.columns WHERE table_name='orders'"
        )).fetchall()
        assert "status" not in [c[0] for c in cols]

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

Начни с трёх уровней тестирования: применимость, откат, целостность данных. Затем упомяни, что в Python стандартный инструмент - Alembic, и тесты строятся на pytest с временной базой. Обязательно скажи про разницу между пустой базой и базой с данными - это ключевой момент, который отличает senior от junior. Добавь про проверку отката и про то, что миграции должны быть идемпотентными. Если спросят про производительность, упомяни отдельный нагрузочный тест и копию продакшен-базы. Не углубляйся в детали реализации, если не просят - лучше показать понимание trade-off между скоростью тестов и их полнотой.

Что проверяет интервьюер

Интервьюер оценивает:

• Понимание, что миграция - это не просто DDL, а изменение данных, которое может сломаться на реальных данных.
• Знание инструментов экосистемы Python: Alembic, pytest, фикстуры, временные базы.
• Умение проектировать тесты: какие данные вставлять, какие инварианты проверять.
• Понимание отката и идемпотентности - это признаки зрелости.
• Умение объяснить, почему тесты миграций не должны быть частью обычного быстрого CI.

Типичные ошибки

• Тестировать только на пустой базе - пропускает ошибки backfill и конфликты с данными.
• Не проверять downgrade - миграция может быть необратимой, и это выяснится в проде.
• Использовать SQLite вместо PostgreSQL - различия в типах, индексах и поведении транзакций дают ложную уверенность.
• Вставлять "идеальные" данные без NULL, дубликатов и длинных строк - миграция падает на реальных данных.
• Не проверять идемпотентность повторных запусков - особенно для миграций с CREATE INDEX или INSERT SELECT.
• Запускать тесты миграций в том же CI-джобе, что и обычные юнит-тесты - это замедляет разработку и маскирует проблемы с окружением.