testcontainers-python: реальные БД в pytest

Самая частая ошибка junior’а при тестировании DE-кода: моки. Ты пишешь функцию load_users_to_db(conn, df), пишешь тест с mock_conn = MagicMock(), тест зелёный, ты счастлив. В проде функция падает с psycopg2.errors.UniqueViolation — потому что mock не проверял, что таблица уже содержит этот user_id, и реальный INSERT упал.

Решение: тестируй на реальной БД. Не staging, не “общая dev-БД”, а свежезапущенная Postgres-контейнер, который существует только во время теста. Эту магию даёт testcontainers — Python-библиотека, которая стартует контейнеры из тестов и убивает после.

В этом уроке: установка, базовый паттерн с Postgres, миграции, fixture’ы pytest, и почему testcontainers заменяет 90% ваших mock’ов.

Что не так с моками

Допустим, у тебя ETL-функция:

def upsert_users(conn, users: list[dict]) -> int:
    with conn.cursor() as cur:
        cur.executemany("""
            INSERT INTO users (id, name, email)
            VALUES (%(id)s, %(name)s, %(email)s)
            ON CONFLICT (id) DO UPDATE
            SET name = EXCLUDED.name, email = EXCLUDED.email
        """, users)
    conn.commit()
    return len(users)

С моками:

def test_upsert_users_mock():
    conn = MagicMock()
    result = upsert_users(conn, [{"id": 1, "name": "Alice", "email": "[email protected]"}])
    assert result == 1
    conn.cursor.assert_called()  # OK

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

• Что SQL правильный (опечатка в SQL прошла бы тест).
• Что ON CONFLICT работает (тебе нужна реальная БД с UNIQUE constraint).
• Что batch на 10000 строк не зависает.
• Что Unicode не ломает encoding.
• Что транзакция действительно коммитится.

Для DE-кода mock-тесты часто хуже отсутствия тестов — они дают false confidence.

testcontainers: реальная БД в тесте

pip install testcontainers[postgres]==4.8.1 psycopg2-binary

Базовый паттерн через контекст-менеджер:

from testcontainers.postgres import PostgresContainer
import psycopg2

def test_upsert_users_real():
    with PostgresContainer("postgres:16") as postgres:
        conn = psycopg2.connect(postgres.get_connection_url())

        # Создаём схему
        with conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE users (
                    id INT PRIMARY KEY,
                    name TEXT NOT NULL,
                    email TEXT NOT NULL
                )
            """)
        conn.commit()

        # Тестируем функцию
        result = upsert_users(conn, [
            {"id": 1, "name": "Alice", "email": "[email protected]"},
            {"id": 1, "name": "Alice Updated", "email": "[email protected]"},
        ])

        # Проверяем РЕАЛЬНОЕ состояние БД
        with conn.cursor() as cur:
            cur.execute("SELECT name, email FROM users WHERE id = 1")
            assert cur.fetchone() == ("Alice Updated", "[email protected]")

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

1. PostgresContainer("postgres:16") создал объект контейнера.
2. with запустил его (docker run под капотом). Дождался healthy.
3. get_connection_url() вернул что-то типа postgresql+psycopg2://test:test@localhost:55321/test — порт случайный, чтобы не конфликтовать с другими тестами.
4. Внутри with ты работаешь с реальной БД.
5. После выхода из with контейнер удалён (docker rm -f).

pytest fixture для переиспользования

Каждый тест стартует Postgres за ~5 секунд. Если у тебя 20 тестов — это 100 секунд только на старты. Не годится.

Решение: scope=“session” fixture — один контейнер на всю сессию pytest, очистка между тестами через TRUNCATE.

# conftest.py
import pytest
import psycopg2
from testcontainers.postgres import PostgresContainer

@pytest.fixture(scope="session")
def postgres_container():
    """Один контейнер на всю сессию pytest."""
    with PostgresContainer("postgres:16") as postgres:
        yield postgres

@pytest.fixture(scope="session")
def db_url(postgres_container):
    return postgres_container.get_connection_url().replace(
        "postgresql+psycopg2://", "postgresql://"
    )

@pytest.fixture(scope="session")
def db_schema(db_url):
    """Создаёт схему один раз."""
    conn = psycopg2.connect(db_url)
    with conn.cursor() as cur:
        cur.execute("""
            CREATE TABLE users (
                id INT PRIMARY KEY,
                name TEXT NOT NULL,
                email TEXT NOT NULL
            );
            CREATE TABLE events (
                id SERIAL PRIMARY KEY,
                user_id INT REFERENCES users(id),
                event_type TEXT,
                event_time TIMESTAMP DEFAULT now()
            );
        """)
    conn.commit()
    yield
    conn.close()

@pytest.fixture
def db_conn(db_url, db_schema):
    """Свежий connection на каждый тест, TRUNCATE до начала."""
    conn = psycopg2.connect(db_url)
    with conn.cursor() as cur:
        cur.execute("TRUNCATE users, events RESTART IDENTITY CASCADE")
    conn.commit()
    yield conn
    conn.close()

Теперь тест:

# test_users.py
def test_upsert_users(db_conn):
    result = upsert_users(db_conn, [
        {"id": 1, "name": "Alice", "email": "[email protected]"},
    ])
    assert result == 1

def test_upsert_with_conflict(db_conn):
    upsert_users(db_conn, [{"id": 1, "name": "Alice", "email": "[email protected]"}])
    upsert_users(db_conn, [{"id": 1, "name": "Alice2", "email": "[email protected]"}])

    with db_conn.cursor() as cur:
        cur.execute("SELECT count(*) FROM users")
        assert cur.fetchone()[0] == 1  # один user, не два

Контейнер стартует один раз, тесты гоняются параллельно по чистой схеме.

Миграции через Alembic / SQL-файлы

В реальном проекте схема описана не inline, а в Alembic-миграциях или .sql файлах. Подключаем их к testcontainer:

import subprocess
from pathlib import Path

@pytest.fixture(scope="session")
def db_with_migrations(postgres_container):
    db_url = postgres_container.get_connection_url()

    # Запускаем Alembic upgrade
    subprocess.run(
        ["alembic", "upgrade", "head"],
        env={"DATABASE_URL": db_url, **os.environ},
        check=True,
    )
    yield postgres_container

Или прямо через psql:

@pytest.fixture(scope="session")
def db_with_schema(postgres_container):
    conn = psycopg2.connect(postgres_container.get_connection_url())
    schema_sql = Path("migrations/001_init.sql").read_text()
    with conn.cursor() as cur:
        cur.execute(schema_sql)
    conn.commit()
    yield postgres_container

Теперь тесты используют ту же схему, что и production, потому что миграции одни.

Не только Postgres

testcontainers умеет десятки сервисов:

from testcontainers.kafka import KafkaContainer
from testcontainers.localstack import LocalStackContainer
from testcontainers.redis import RedisContainer

# Kafka
with KafkaContainer() as kafka:
    bootstrap = kafka.get_bootstrap_server()
    # bootstrap = "PLAINTEXT://localhost:55421"
    # ... твой producer/consumer

# S3 через LocalStack
with LocalStackContainer(image="localstack/localstack:3.7").with_services("s3") as ls:
    endpoint = ls.get_url()
    # ... boto3 client с endpoint_url=endpoint

# Redis
with RedisContainer("redis:7") as redis:
    port = redis.get_exposed_port(6379)

Один pattern, разные сервисы. Идеально, когда твой ETL pipeline трогает несколько источников: реальные Postgres + Kafka + S3 в каждом интеграционном тесте.

CI: testcontainers в GitHub Actions

testcontainers требует Docker daemon. В GitHub Actions это работает из коробки на ubuntu-latest:

# .github/workflows/test.yml
name: tests
on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - name: Install deps
        run: |
          pip install -r requirements.txt
          pip install pytest testcontainers[postgres,kafka]
      - name: Run tests
        run: pytest -v

Docker уже установлен на runner’е, testcontainers использует его. Никакой дополнительной настройки.

Почему лучше mock’ов

testcontainers платит цену в скорости (на порядок медленнее) и в требовании Docker — но взамен ловит реальные баги. Для DE-кода, который работает с БД и брокерами, это критично.

Real-life пример: ETL-pipeline

# etl/pipeline.py
import psycopg2
from confluent_kafka import Consumer
import json

def kafka_to_postgres(kafka_bootstrap: str, db_url: str, topic: str, max_messages: int = 100):
    consumer = Consumer({
        "bootstrap.servers": kafka_bootstrap,
        "group.id": "etl-test",
        "auto.offset.reset": "earliest",
    })
    consumer.subscribe([topic])

    conn = psycopg2.connect(db_url)
    count = 0
    try:
        while count < max_messages:
            msg = consumer.poll(timeout=1.0)
            if msg is None: break
            if msg.error(): continue
            event = json.loads(msg.value())
            with conn.cursor() as cur:
                cur.execute(
                    "INSERT INTO events (user_id, event_type) VALUES (%s, %s)",
                    (event["user_id"], event["type"]),
                )
            count += 1
        conn.commit()
    finally:
        consumer.close()
        conn.close()
    return count

# tests/test_etl.py
def test_kafka_to_postgres(postgres_container, kafka_container):
    db_url = postgres_container.get_connection_url()
    kafka_bs = kafka_container.get_bootstrap_server()

    # Setup schema
    conn = psycopg2.connect(db_url)
    with conn.cursor() as cur:
        cur.execute("CREATE TABLE events (id SERIAL, user_id INT, event_type TEXT)")
    conn.commit()
    conn.close()

    # Produce test messages
    from confluent_kafka import Producer
    producer = Producer({"bootstrap.servers": kafka_bs})
    for i in range(5):
        producer.produce("events", value=json.dumps({"user_id": i, "type": "click"}))
    producer.flush()

    # Run ETL
    count = kafka_to_postgres(kafka_bs, db_url, "events", max_messages=5)
    assert count == 5

    # Assert state in Postgres
    conn = psycopg2.connect(db_url)
    with conn.cursor() as cur:
        cur.execute("SELECT count() FROM events")
        assert cur.fetchone()[0] == 5

Этот тест проверяет end-to-end pipeline: реальный Kafka, реальный JSON-encoding, реальный Postgres INSERT. Если в продакшене pipeline сломается из-за encoding-бага в Kafka message — этот тест поймает.

Попробуй сам

# 1. Установка
pip install pytest testcontainers[postgres] psycopg2-binary

# 2. Создай tests/conftest.py с fixture'ами выше

# 3. Создай tests/test_users.py
cat > tests/test_users.py <<'EOF'
def test_simple(db_conn):
    with db_conn.cursor() as cur:
        cur.execute("INSERT INTO users (id, name, email) VALUES (1, 'A', 'a@x')")
        cur.execute("SELECT count() FROM users")
        assert cur.fetchone()[0] == 1
EOF

# 4. Запусти
pytest -v tests/
# Запустится Postgres-контейнер, тест пройдёт, контейнер удалится

# 5. Посмотри: docker ps во время теста увидит контейнер
# (запусти pytest в одном терминале, docker ps -- в другом)