Главная/Testing/Урок

15.4. Testcontainers for Python

Цели

После этого материала вы сможете:

  • поднимать зависимости из кода теста и объяснить, чем это отличается от Compose;
  • выбрать область действия фикстуры и объяснить, почему это решение стоит в сто раз дороже остальных;
  • объяснить, что такое Ryuk и почему очистка работает даже при убитом процессе тестов;
  • оценить цену переиспользования container'а и назвать, что оно ломает;
  • назвать требование к окружению CI, которое Testcontainers предъявляет, а Compose — нет;
  • обоснованно выбрать между Compose и Testcontainers для конкретного проекта.

Предварительные знания

Ключевые термины

ТерминОбъяснение
TestcontainersБиблиотека, управляющая container'ами из кода теста
RyukСлужебный container, удаляющий созданные объекты
wait strategyВстроенное условие готовности зависимости
scopeОбласть действия фикстуры pytest
reuseПереиспользование container'а между прогонами

О версиях. В примерах ниже зависимость указана как testcontainers[postgres] без закреплённой версии. Актуальную версию возьмите с PyPI и закрепите у себя — в этом уроке она не приводится, поскольку не была сверена при его написании (урок 11.6).


Теория

Кто управляет окружением

text
Compose                          Testcontainers

compose.test.yaml                код теста
      │                                │
      ▼                                ▼
docker compose up  ──► тесты     pytest ──► библиотека ──► Docker API
      │                                              │
   окружение снаружи                        окружение изнутри
СвойствоComposeTestcontainers
Где описано окружениеYAML-файлКод на Python
Кто запускаетОболочка, CIСам тест
Прогон одного тестаПоднимает весь файлПоднимает нужное
Требует доступ к Docker socket из тестовНетДа
Разные версии зависимости в одном набореНеудобноЕстественно
Понятно человеку без PythonДаНет

Строка про socket — главная практическая разница. Тесты под Compose работают в container'е, ничего не зная о Docker. Тесты с Testcontainers сами создают container'ы, а значит, им нужен доступ к API daemon (урок 12.2).

Минимальный пример

python
from testcontainers.postgres import PostgresContainer

with PostgresContainer("postgres:17-alpine") as postgres:
    url = postgres.get_connection_url()
    # container поднят, готовность дождалась библиотека
# здесь container уже удалён

Контекстный менеджер делает четыре вещи: скачивает образ, запускает container, ждёт готовности встроенной стратегией, а на выходе останавливает и удаляет.

Ожидание готовности — то, ради чего библиотеку берут. Для PostgreSQL она сама знает, что нужно дождаться строки в логе и успешного подключения; писать healthcheck вручную не требуется.

Область действия фикстуры: решение ценой в сто раз

Три взаимоисключающих варианта:

text
@pytest.fixture(scope="function")   container на КАЖДЫЙ тест
@pytest.fixture(scope="module")     на файл
@pytest.fixture(scope="session")    один на весь прогон
ОбластьContainer'ов на 100 тестовВремя только на старты
function100Минуты
moduleПо числу файловДесятки секунд
session1Секунды

Разница между function и session — два порядка. При этом изоляция при session не теряется: её обеспечивает отдельный механизм (урок 15.3).

Правильное сочетание:

text
container            → scope="session"   (дорого поднимать)
схема или транзакция → scope="function"  (дёшево создавать)

Ошибка, встречающаяся постоянно: scope="function" для container'а «ради изоляции». Изоляцию это даёт, но платит за неё в сто раз дороже необходимого.

Ryuk: почему очистка не подводит

Testcontainers запускает служебный container — Ryuk. Он получает доступ к Docker API и следит за соединением с процессом тестов.

text
pytest ──(соединение)──► Ryuk ──► Docker API
   │
   × процесс убит
                          │
                          ▼
              соединение разорвано → Ryuk удаляет всё помеченное

Каждый созданный объект помечается меткой сеанса. При обрыве соединения Ryuk удаляет всё с этой меткой.

Это решает задачу, для которой в Compose нужен trap (урок 15.3): очистка происходит даже при kill -9 процесса тестов.

ПеременнаяДействие
TESTCONTAINERS_RYUK_DISABLED=trueОтключает Ryuk
TESTCONTAINERS_RYUK_PRIVILEGED=trueЗапускает Ryuk привилегированным

Отключение иногда требуется в средах, где запуск дополнительного container'а невозможен. Цена: объекты остаются при аварийном завершении, и очистку придётся делать самому.

Переиспользование container'а

python
container = PostgresContainer("postgres:17-alpine").with_reuse()

Container не удаляется после прогона и подхватывается следующим. Экономия — секунды старта на каждом запуске.

Цена:

Что ломаетсяПочему
Чистота состоянияВ базе остаются данные прошлого прогона
ВоспроизводимостьРезультат зависит от истории запусков
Работа в CIКаждая задача получает свою машину — переиспользовать нечего

Требуется явное согласие в ~/.testcontainers.properties:

properties
testcontainers.reuse.enable=true

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

Требование к CI

Тестам нужен доступ к Docker. Варианты:

СпособКакРиск
Монтирование socket-v /var/run/docker.sock:/var/run/docker.sockРавносильно root на host (урок 12.2)
Docker-in-DockerОтдельный сервис docker:dindТребует --privileged
Готовый сборщик с DockerМногие CI дают из коробкиЗависит от платформы
Rootless DockerОтдельный daemon без rootСложнее настроить

Это и есть главная цена подхода. Compose такого требования не предъявляет: тесты выполняются внутри container'а и о Docker не знают.

Когда что выбирать

УсловиеComposeTestcontainers
Окружение одинаково для всех тестов
Нужны разные версии зависимости в одном наборе
Тесты запускают не только разработчики Python
Нужно поднять зависимость только для части тестов
Доступ к Docker socket из тестов недопустим
То же окружение нужно для ручной работы
Набор тестов — библиотека, а не сервис

Смешивать не следует. Два механизма поднятия окружения в одном проекте усложняют его без выгоды: удваивается число мест, где что-то может пойти не так.


Внутренний механизм

Как определяется адрес

Container получает случайный порт на host (-p 0:5432), и библиотека узнаёт фактический через Docker API. Отсюда:

python
url = postgres.get_connection_url()   # postgresql+psycopg2://...@localhost:49173/test

Порт разный при каждом запуске — это и даёт параллельность без конфликтов, о которой в Compose приходится заботиться вручную (урок 15.3).

Хардкод порта в тесте ломает это свойство и приводит к конфликтам при параллельном запуске.

Как работает ожидание готовности

Для каждого типа зависимости в библиотеке описана стратегия: ждать строку в логе, ждать открытия порта, ждать успешного вызова.

Для PostgreSQL это комбинация: сначала строка в логе, затем реальное подключение. То есть ровно то, к чему приходят вручную в уроке 15.3 — проверка должна делать то же, что приложение.

Своя стратегия задаётся явно:

python
from testcontainers.core.waiting_utils import wait_for_logs

container = DockerContainer("myimage:1.0").with_exposed_ports(8080)
container.start()
wait_for_logs(container, "сервис готов", timeout=30)

Команды и примеры

Основа: container из кода теста

bash
mkdir -p /tmp/tcdemo && cd /tmp/tcdemo
mkdir -p src tests

cat > requirements.txt <<'EOF'
psycopg[binary]==3.3.4
EOF
cat > requirements-dev.txt <<'EOF'
pytest==9.1.1
# Версию закрепите по данным PyPI: здесь она намеренно не указана,
# поскольку не сверялась при написании урока
testcontainers[postgres]
EOF

cat > src/__init__.py <<'PY'
PY

cat > src/store.py <<'PY'
"""Хранилище событий на PostgreSQL."""
from __future__ import annotations

import psycopg

SCHEMA = """
CREATE TABLE IF NOT EXISTS events (
    id      SERIAL PRIMARY KEY,
    kind    TEXT NOT NULL,
    payload JSONB NOT NULL,
    at      TIMESTAMPTZ NOT NULL DEFAULT now()
);
"""


def init_schema(conn: psycopg.Connection) -> None:
    conn.execute(SCHEMA)
    conn.commit()


def add_event(conn: psycopg.Connection, kind: str, payload: dict) -> int:
    import json
    row = conn.execute(
        "INSERT INTO events (kind, payload) VALUES (%s, %s) RETURNING id",
        (kind, json.dumps(payload)),
    ).fetchone()
    return row[0]


def events_of_kind(conn: psycopg.Connection, kind: str) -> list[dict]:
    rows = conn.execute(
        "SELECT id, kind, payload FROM events WHERE kind = %s ORDER BY id",
        (kind,),
    ).fetchall()
    return [{"id": r[0], "kind": r[1], "payload": r[2]} for r in rows]


def count_events(conn: psycopg.Connection) -> int:
    return conn.execute("SELECT count(*) FROM events").fetchone()[0]
PY

cat > tests/conftest.py <<'PY'
"""Фикстуры Testcontainers.

Ключевое решение: container — на весь прогон (дорого поднимать),
изоляция — на каждый тест (дёшево создавать схему).
Обратный выбор замедлил бы прогон примерно в сто раз.
"""
from __future__ import annotations

import uuid

import psycopg
import pytest
from testcontainers.postgres import PostgresContainer

from src.store import init_schema


@pytest.fixture(scope="session")
def postgres():
    """Один container на весь прогон.

    Библиотека сама дожидается готовности: сначала строка в логе,
    затем реальное подключение. Писать healthcheck не требуется.
    """
    with PostgresContainer("postgres:17-alpine") as container:
        yield container


@pytest.fixture(scope="session")
def dsn(postgres) -> str:
    """Адрес с ФАКТИЧЕСКИМ портом: он случаен при каждом запуске."""
    return (
        f"postgresql://{postgres.username}:{postgres.password}"
        f"@{postgres.get_container_host_ip()}"
        f":{postgres.get_exposed_port(5432)}/{postgres.dbname}"
    )


@pytest.fixture(scope="session")
def connection(dsn: str):
    conn = psycopg.connect(dsn)
    try:
        yield conn
    finally:
        conn.close()


@pytest.fixture
def db(connection):
    """Изоляция схемой — на каждый тест."""
    schema = f"t_{uuid.uuid4().hex[:12]}"
    connection.execute(f'CREATE SCHEMA "{schema}"')
    connection.execute(f'SET search_path TO "{schema}"')
    connection.commit()
    init_schema(connection)
    try:
        yield connection
    finally:
        connection.rollback()
        connection.execute(f'DROP SCHEMA "{schema}" CASCADE')
        connection.commit()
PY

cat > tests/test_store.py <<'PY'
"""Integration-тесты через Testcontainers."""
import pytest

from src.store import add_event, count_events, events_of_kind


def test_add_and_read(db):
    ident = add_event(db, "order.created", {"number": "N-1", "sum": 100})
    found = events_of_kind(db, "order.created")
    assert len(found) == 1
    assert found[0]["id"] == ident
    assert found[0]["payload"]["number"] == "N-1"


def test_jsonb_roundtrip(db):
    """JSONB — тип PostgreSQL; подделка его не воспроизведёт."""
    payload = {"вложенный": {"список": [1, 2, 3], "флаг": True}}
    add_event(db, "test", payload)
    assert events_of_kind(db, "test")[0]["payload"] == payload


def test_jsonb_query(db):
    """Запрос по полю JSONB выполняет БД."""
    add_event(db, "x", {"status": "ok"})
    add_event(db, "x", {"status": "fail"})
    row = db.execute(
        "SELECT count(*) FROM events WHERE payload->>'status' = 'ok'"
    ).fetchone()
    assert row[0] == 1


def test_ordering_by_id(db):
    for i in range(5):
        add_event(db, "seq", {"n": i})
    found = events_of_kind(db, "seq")
    assert [e["payload"]["n"] for e in found] == [0, 1, 2, 3, 4]


def test_isolation(db):
    """Схема на тест: данные прошлых тестов не видны."""
    assert count_events(db) == 0
PY
echo "  файлы созданы"

Ожидаемый вывод:

text
  файлы созданы

Прогон

bash
cd /tmp/tcdemo
cat > run.sh <<'SH'
#!/usr/bin/env bash
# Testcontainers требует доступ к Docker API из процесса тестов.
# Здесь socket монтируется в container с тестами.
set -uo pipefail

docker run --rm \
    -v /var/run/docker.sock:/var/run/docker.sock \
    -v "$PWD:/w" -w /w \
    --network host \
    -e PYTHONPATH=/w \
    -e TESTCONTAINERS_HOST_OVERRIDE=localhost \
    python:3.13-slim sh -c '
        pip install --quiet --no-cache-dir -r requirements.txt -r requirements-dev.txt \
            > /dev/null 2>&1
        python -m pytest tests/ -q --tb=short "$@"
    ' -- "$@"
SH
chmod +x run.sh

echo "═══ прогон тестов ═══"
timeout 300 ./run.sh 2>&1 | tail -6 | sed 's/^/  /'

echo "═══ что происходило ═══"
cat <<'TXT'
  1. pytest запросил фикстуру postgres (scope="session")
  2. Библиотека скачала образ postgres:17-alpine
  3. Запустила container с портом -p 0:5432 (случайный порт)
  4. Дождалась готовности встроенной стратегией
  5. Прогнала все тесты через ОДИН container
  6. Удалила container при завершении сеанса

  Ни healthcheck, ни compose-файла, ни ожидания вручную —
  всё это внутри библиотеки.
TXT

Ожидаемый вывод:

text
═══ прогон тестов ═══
  .....                                                                [100%]
  5 passed in 12.84s
═══ что происходило ═══
  1. pytest запросил фикстуру postgres (scope="session")
  2. Библиотека скачала образ postgres:17-alpine
  3. Запустила container с портом -p 0:5432 (случайный порт)
  4. Дождалась готовности встроенной стратегией
  5. Прогнала все тесты через ОДИН container
  6. Удалила container при завершении сеанса

  Ни healthcheck, ни compose-файла, ни ожидания вручную —
  всё это внутри библиотеки.

Двенадцать секунд на пять тестов — почти всё это старт PostgreSQL. Отсюда следующий вопрос: сколько раз он поднимается.

Область действия фикстуры: цена решения

bash
cd /tmp/tcdemo
cat > tests/conftest_function.py <<'PY'
"""ОШИБОЧНЫЙ вариант: container на каждый тест.

Изоляцию это даёт, но платит за неё в сто раз дороже необходимого:
PostgreSQL поднимается заново для каждого теста.
"""
from __future__ import annotations

import psycopg
import pytest
from testcontainers.postgres import PostgresContainer

from src.store import init_schema


@pytest.fixture(scope="function")          # ← вот здесь ошибка
def db():
    with PostgresContainer("postgres:17-alpine") as container:
        dsn = (
            f"postgresql://{container.username}:{container.password}"
            f"@{container.get_container_host_ip()}"
            f":{container.get_exposed_port(5432)}/{container.dbname}"
        )
        conn = psycopg.connect(dsn)
        init_schema(conn)
        try:
            yield conn
        finally:
            conn.close()
PY

cat > compare_scopes.sh <<'SH'
#!/usr/bin/env bash
# Сравнение scope="session" и scope="function" по времени.
set -uo pipefail

run_variant() {
    local label="$1" conftest="$2"
    cp "$conftest" tests/conftest.py
    local start end
    start="$(python3 -c 'import time; print(time.monotonic())')"
    timeout 600 ./run.sh > "/tmp/scope_$label.log" 2>&1
    local rc=$?
    end="$(python3 -c 'import time; print(time.monotonic())')"
    local seconds
    seconds="$(python3 -c "print(f'{$end - $start:.1f}')")"
    local containers
    containers="$(grep -c 'Pulling image\|Container started' "/tmp/scope_$label.log" 2>/dev/null || echo "?")"
    printf '  %-28s %8s с  (код %s)\n' "$label" "$seconds" "$rc"
    echo "$seconds"
}

cp tests/conftest.py /tmp/conftest_session_backup.py
SH
chmod +x compare_scopes.sh

echo "═══ сравнение областей действия ═══"
cp tests/conftest.py /tmp/conftest_session.py

measure_variant() {
    local label="$1"
    local start end
    start="$(python3 -c 'import time; print(time.monotonic())')"
    timeout 600 ./run.sh > "/tmp/v_$label.log" 2>&1
    end="$(python3 -c 'import time; print(time.monotonic())')"
    python3 -c "print(f'{$end - $start:.1f}')"
}

t_session="$(measure_variant session)"
printf '  %-30s %6s с\n' 'scope="session" (один container)' "$t_session"

cp tests/conftest_function.py tests/conftest.py
t_function="$(measure_variant function)"
printf '  %-30s %6s с\n' 'scope="function" (на каждый тест)' "$t_function"
cp /tmp/conftest_session.py tests/conftest.py

python3 -c "
s, f = $t_session, $t_function
n = 5
print(f'  на {n} тестах: разница {f - s:.1f} с, отношение {f / s:.1f}×')
print(f'  на 100 тестах ожидаемо: session ~{s:.0f} с, function ~{f / n * 100:.0f} с')
"

echo "═══ правильное сочетание ═══"
cat <<'TXT'
  container            scope="session"   поднимается один раз
  схема или транзакция scope="function"  создаётся на каждый тест

  Изоляция при этом полная: каждый тест видит свою схему.
  Ошибка «container на каждый тест ради изоляции» даёт ту же
  изоляцию за в сто раз большую цену.
TXT
rm -f tests/conftest_function.py compare_scopes.sh

Ожидаемый вывод:

text
═══ сравнение областей действия ═══
  scope="session" (один container)   12.9 с
  scope="function" (на каждый тест)  48.3 с
  на 5 тестах: разница 35.4 с, отношение 3.7×
  на 100 тестах ожидаемо: session ~13 с, function ~966 с
═══ правильное сочетание ═══
  container            scope="session"   поднимается один раз
  схема или транзакция scope="function"  создаётся на каждый тест

  Изоляция при этом полная: каждый тест видит свою схему.
  Ошибка «container на каждый тест ради изоляции» даёт ту же
  изоляцию за в сто раз большую цену.

Отношение на пяти тестах — всего 3,7×, потому что старт первого container'а входит в оба замера. Экстраполяция на сотню тестов показывает настоящий масштаб: 13 секунд против 16 минут.

Ryuk: очистка при убитом процессе

bash
cd /tmp/tcdemo
echo "═══ служебный container Ryuk ═══"
docker ps -a --filter 'ancestor=testcontainers/ryuk' \
    --format '  {{.Names}}  {{.Image}}  {{.Status}}' 2>/dev/null | head -3
echo "  (появляется во время прогона и исчезает после)"

echo "═══ проверка: убиваем процесс тестов ═══"
cat > slow_test.py <<'PY'
"""Тест, который висит достаточно долго, чтобы его прервать."""
import time

import pytest
from testcontainers.postgres import PostgresContainer


def test_long_running():
    with PostgresContainer("postgres:17-alpine") as container:
        print(f"container поднят: {container.get_exposed_port(5432)}", flush=True)
        time.sleep(120)
PY

docker run -d --name tc-kill-demo \
    -v /var/run/docker.sock:/var/run/docker.sock \
    -v "$PWD:/w" -w /w --network host \
    -e TESTCONTAINERS_HOST_OVERRIDE=localhost \
    python:3.13-slim sh -c '
        pip install --quiet --no-cache-dir pytest testcontainers[postgres] \
            > /dev/null 2>&1
        python -m pytest slow_test.py -q -s
    ' > /dev/null 2>&1

echo "  ждём, пока поднимется PostgreSQL..."
for i in $(seq 1 40); do
    if docker logs tc-kill-demo 2>&1 | grep -q 'container поднят'; then
        break
    fi
    sleep 2
done
before="$(docker ps --filter 'ancestor=postgres:17-alpine' -q | wc -l)"
ryuk="$(docker ps --filter 'ancestor=testcontainers/ryuk' -q | wc -l)"
printf '  до убийства: postgres=%s ryuk=%s\n' "$before" "$ryuk"

echo "  убиваем процесс тестов сигналом KILL:"
docker kill -s KILL tc-kill-demo > /dev/null 2>&1
sleep 15
after="$(docker ps --filter 'ancestor=postgres:17-alpine' -q | wc -l)"
printf '  после убийства: postgres=%s\n' "$after"
docker rm -f tc-kill-demo > /dev/null 2>&1
rm -f slow_test.py

echo "═══ как это работает ═══"
cat <<'TXT'
  Ryuk — служебный container, который держит соединение
  с процессом тестов и следит за ним.

  Каждый созданный объект помечается меткой сеанса.
  Обрыв соединения → Ryuk удаляет всё с этой меткой.

  Это решает ровно ту задачу, для которой в Compose нужен trap
  ([урок 15.3]) — но работает даже при kill -9, когда никакой
  обработчик сигнала уже не сработает.

  Отключение (иногда требуется в ограниченных средах):
    TESTCONTAINERS_RYUK_DISABLED=true

  Цена отключения: объекты остаются при аварийном завершении.
TXT

Ожидаемый вывод:

text
═══ служебный container Ryuk ═══
  (появляется во время прогона и исчезает после)
═══ проверка: убиваем процесс тестов ═══
  ждём, пока поднимется PostgreSQL...
  до убийства: postgres=1 ryuk=1
  убиваем процесс тестов сигналом KILL:
  после убийства: postgres=0
═══ как это работает ═══
  Ryuk — служебный container, который держит соединение
  с процессом тестов и следит за ним.

  Каждый созданный объект помечается меткой сеанса.
  Обрыв соединения → Ryuk удаляет всё с этой меткой.

  Это решает ровно ту задачу, для которой в Compose нужен trap
  ([урок 15.3]) — но работает даже при kill -9, когда никакой
  обработчик сигнала уже не сработает.
  ...

postgres=1 до и postgres=0 после — при том, что процесс тестов был убит SIGKILL и не мог выполнить никакой обработчик.

Это преимущество перед trap: обработчик сигнала при kill -9 не запускается вовсе.

Переиспользование и его цена

bash
cd /tmp/tcdemo
echo "═══ что даёт .with_reuse() ═══"
cat <<'TXT'
  container = PostgresContainer("postgres:17-alpine").with_reuse()

  Container не удаляется после прогона и подхватывается следующим.
  Экономия — секунды старта на каждом запуске.

  Требуется явное согласие в ~/.testcontainers.properties:
    testcontainers.reuse.enable=true

  Без этого файла флаг игнорируется.
TXT

echo "═══ что переиспользование ломает ═══"
python3 - <<'PY'
CASES = [
    ("Чистота состояния", "в базе остаются данные прошлого прогона",
     "изоляция схемой на тест — обязательна, а не желательна"),
    ("Воспроизводимость", "результат зависит от истории запусков",
     "тест может проходить только потому, что до него что-то создали"),
    ("Смена образа", "старый container не пересоздаётся автоматически",
     "обновили версию PostgreSQL — тесты идут на старой"),
    ("Работа в CI", "каждая задача получает свою машину",
     "переиспользовать нечего; выгоды ноль"),
]
print(f"  {'что ломается':<24} {'почему':<46} следствие")
print("  " + "─" * 108)
for what, why, effect in CASES:
    print(f"  {what:<24} {why:<46} {effect}")
print()
print("  Вывод: переиспользование — инструмент ЛОКАЛЬНОЙ разработки.")
print("  В CI оно бесполезно и создаёт ложную уверенность,")
print("  что тесты быстрые.")
PY

echo "═══ проверка: файл согласия ═══"
if [ -f "$HOME/.testcontainers.properties" ]; then
    grep -c 'reuse.enable' "$HOME/.testcontainers.properties" 2>/dev/null \
        | sed 's/^/  строк с reuse.enable: /'
else
    echo "  ~/.testcontainers.properties отсутствует — переиспользование выключено"
fi

Ожидаемый вывод:

text
═══ что даёт .with_reuse() ═══
  container = PostgresContainer("postgres:17-alpine").with_reuse()

  Container не удаляется после прогона и подхватывается следующим.
  Экономия — секунды старта на каждом запуске.

  Требуется явное согласие в ~/.testcontainers.properties:
    testcontainers.reuse.enable=true

  Без этого файла флаг игнорируется.
═══ что переиспользование ломает ═══
  что ломается             почему                                         следствие
  ────────────────────────────────────────────────────────────────────────────────────────────────────────────
  Чистота состояния        в базе остаются данные прошлого прогона        изоляция схемой на тест — обязательна, а не желательна
  Воспроизводимость        результат зависит от истории запусков          тест может проходить только потому, что до него что-то создали
  Смена образа             старый container не пересоздаётся автоматически  обновили версию PostgreSQL — тесты идут на старой
  Работа в CI              каждая задача получает свою машину             переиспользовать нечего; выгоды ноль

  Вывод: переиспользование — инструмент ЛОКАЛЬНОЙ разработки.
  В CI оно бесполезно и создаёт ложную уверенность,
  что тесты быстрые.
═══ проверка: файл согласия ═══
  ~/.testcontainers.properties отсутствует — переиспользование выключено

Третья строка таблицы — самая коварная: обновили версию образа в коде, а тесты продолжают идти на старом container'е и ничего не сообщают.

Разные версии зависимости в одном наборе

bash
cd /tmp/tcdemo
cat > tests/test_versions.py <<'PY'
"""То, что через Compose делается неудобно: проверка на нескольких
версиях зависимости в одном наборе тестов.

Каждая версия — свой container, поднимаемый параметризованной фикстурой.
"""
from __future__ import annotations

import psycopg
import pytest
from testcontainers.postgres import PostgresContainer

VERSIONS = ["postgres:16-alpine", "postgres:17-alpine"]


@pytest.fixture(scope="session", params=VERSIONS, ids=lambda v: v.split(":")[1])
def pg_any_version(request):
    with PostgresContainer(request.param) as container:
        yield container


@pytest.fixture
def conn_any(pg_any_version):
    dsn = (
        f"postgresql://{pg_any_version.username}:{pg_any_version.password}"
        f"@{pg_any_version.get_container_host_ip()}"
        f":{pg_any_version.get_exposed_port(5432)}/{pg_any_version.dbname}"
    )
    conn = psycopg.connect(dsn)
    try:
        yield conn
    finally:
        conn.close()


def test_server_version_reported(conn_any):
    row = conn_any.execute("SHOW server_version").fetchone()
    assert row[0], "версия сервера не получена"
    print(f"\n    сервер: {row[0]}")


def test_jsonb_supported_everywhere(conn_any):
    """Проверка совместимости: работает ли код на обеих версиях."""
    conn_any.execute("CREATE TEMP TABLE t (data JSONB)")
    conn_any.execute("""INSERT INTO t VALUES ('{"k": "v"}')""")
    row = conn_any.execute("SELECT data->>'k' FROM t").fetchone()
    assert row[0] == "v"
PY

echo "═══ прогон на двух версиях PostgreSQL ═══"
timeout 400 ./run.sh tests/test_versions.py -v 2>&1 \
    | grep -E 'сервер:|PASSED|passed|failed' | head -8 | sed 's/^/  /'

echo "═══ почему это трудно через Compose ═══"
cat <<'TXT'
  В Compose пришлось бы:
    · описать два сервиса db-16 и db-17
    · поднять оба на каждом прогоне, даже когда нужен один
    · передать тестам два разных DATABASE_URL
    · вручную сопоставить параметр теста с нужным сервисом

  Здесь это одна строка params=VERSIONS в фикстуре.

  Это и есть случай, где Testcontainers однозначно выигрывает:
  окружение зависит от параметра теста.
TXT
rm -f tests/test_versions.py

Ожидаемый вывод:

text
═══ прогон на двух версиях PostgreSQL ═══
      сервер: 16.10
  tests/test_versions.py::test_server_version_reported[16-alpine] PASSED
  tests/test_versions.py::test_jsonb_supported_everywhere[16-alpine] PASSED
      сервер: 17.6
  tests/test_versions.py::test_server_version_reported[17-alpine] PASSED
  tests/test_versions.py::test_jsonb_supported_everywhere[17-alpine] PASSED
  4 passed in 26.71s
═══ почему это трудно через Compose ═══
  В Compose пришлось бы:
    · описать два сервиса db-16 и db-17
    · поднять оба на каждом прогоне, даже когда нужен один
    · передать тестам два разных DATABASE_URL
    · вручную сопоставить параметр теста с нужным сервисом

  Здесь это одна строка params=VERSIONS в фикстуре.
  ...

Один и тот же тест выполнен на двух версиях PostgreSQL; версия видна в имени теста.

Это самый убедительный довод в пользу подхода: окружение становится параметром теста.

Выбор между подходами

bash
cd /tmp/tcdemo
cat > choose.py <<'PY'
"""Выбор между Compose и Testcontainers по свойствам проекта."""
from __future__ import annotations

CRITERIA = [
    ("Окружение одинаково для всех тестов", "compose", 2,
     "один файл проще кода в фикстурах"),
    ("Нужны разные версии зависимости", "testcontainers", 3,
     "окружение становится параметром теста"),
    ("Тесты запускают не только Python-разработчики", "compose", 2,
     "YAML читается без знания языка"),
    ("Зависимость нужна лишь части тестов", "testcontainers", 2,
     "поднимается по требованию, а не всегда"),
    ("Доступ к Docker socket из тестов недопустим", "compose", 3,
     "Testcontainers без доступа к API не работает"),
    ("То же окружение нужно для ручной работы", "compose", 2,
     "docker compose up поднимает то же самое"),
    ("Проект — библиотека, а не сервис", "testcontainers", 2,
     "у библиотеки нет своего окружения развёртывания"),
    ("Требуется гарантия очистки при kill -9", "testcontainers", 1,
     "Ryuk работает там, где trap уже не сработает"),
]


def decide(answers: dict[str, bool]) -> None:
    compose = testcontainers = 0
    print(f"  {'критерий':<46} {'ответ':<7} {'в пользу':<16} вес")
    print("  " + "─" * 84)
    for text, favours, weight, _ in CRITERIA:
        answer = answers.get(text, False)
        mark = "да" if answer else "нет"
        if answer:
            if favours == "compose":
                compose += weight
            else:
                testcontainers += weight
        print(f"  {text:<46} {mark:<7} {favours if answer else '—':<16} "
              f"{weight if answer else 0}")

    print()
    print(f"  Compose:        {compose}")
    print(f"  Testcontainers: {testcontainers}")
    print()
    if compose > testcontainers:
        print("  ВЫВОД: Compose")
    elif testcontainers > compose:
        print("  ВЫВОД: Testcontainers")
    else:
        print("  ВЫВОД: равный счёт — выбирайте то, что команда знает лучше")
    print()
    print("  Смешивать НЕ следует: два механизма поднятия окружения")
    print("  удваивают число мест, где что-то может пойти не так.")


if __name__ == "__main__":
    # Типичный веб-сервис: одно окружение, нужен и для ручной работы,
    # доступ к socket в CI ограничен
    decide({
        "Окружение одинаково для всех тестов": True,
        "Тесты запускают не только Python-разработчики": True,
        "Доступ к Docker socket из тестов недопустим": True,
        "То же окружение нужно для ручной работы": True,
    })
PY

echo "═══ выбор для типичного веб-сервиса ═══"
python3 choose.py

cd /tmp && rm -rf /tmp/tcdemo

Ожидаемый вывод:

text
═══ выбор для типичного веб-сервиса ═══
  критерий                                       ответ   в пользу         вес
  ────────────────────────────────────────────────────────────────────────────────────
  Окружение одинаково для всех тестов            да      compose          2
  Нужны разные версии зависимости                нет     —                0
  Тесты запускают не только Python-разработчики  да      compose          2
  Зависимость нужна лишь части тестов            нет     —                0
  Доступ к Docker socket из тестов недопустим    да      compose          3
  То же окружение нужно для ручной работы        да      compose          2
  Проект — библиотека, а не сервис               нет     —                0
  Требуется гарантия очистки при kill -9         нет     —                0

  Compose:        9
  Testcontainers: 0

  ВЫВОД: Compose

Для типичного веб-сервиса счёт получается односторонним. Testcontainers выигрывает в других обстоятельствах — у библиотеки, у набора тестов совместимости, там, где окружение зависит от параметра.


Практическое упражнение

Задание. Перепишите набор тестов на Testcontainers и сравните подходы.

Требования:

  1. Перенести тесты из урока 15.3 на Testcontainers, сохранив изоляцию.
  2. Измерить разницу между scope="session" и scope="function" для container'а.
  3. Показать, что очистка происходит даже при kill -9 процесса тестов.
  4. Показать случай, где Testcontainers выигрывает однозначно: тесты на нескольких версиях зависимости.
  5. Назвать требование к CI, которого нет у Compose, и указать способы его закрытия.
  6. Обосновать выбор подхода для конкретного проекта числами, а не предпочтениями.

Подсказки

Подсказка 1

Правильное сочетание: container — scope="session", изоляция схемой — scope="function".

Подсказка 2

Для пункта 3 запустите тесты в фоне, дождитесь появления container'а зависимости и убейте процесс сигналом KILL.

Подсказка 3

Пункт 4 решается параметризованной фикстурой: @pytest.fixture(params=["postgres:16-alpine", "postgres:17-alpine"]).

Решение

Показать решение
bash
mkdir -p /tmp/tclab && cd /tmp/tclab
mkdir -p src tests

cat > requirements.txt <<'EOF'
psycopg[binary]==3.3.4
EOF
cat > requirements-dev.txt <<'EOF'
pytest==9.1.1
# ВАЖНО: версию закрепите по данным PyPI.
# Здесь она не указана намеренно — не сверялась при написании.
testcontainers[postgres]
EOF

cat > src/__init__.py <<'PY'
PY

cat > src/inventory.py <<'PY'
"""Учёт остатков — тот же код, что в уроке 15.3."""
from __future__ import annotations

import psycopg

SCHEMA = """
CREATE TABLE IF NOT EXISTS items (
    id       SERIAL PRIMARY KEY,
    sku      TEXT NOT NULL UNIQUE,
    quantity INTEGER NOT NULL CHECK (quantity >= 0),
    updated  TIMESTAMPTZ NOT NULL DEFAULT now()
);
"""


def init_schema(conn: psycopg.Connection) -> None:
    conn.execute(SCHEMA)
    conn.commit()


def add_item(conn: psycopg.Connection, sku: str, quantity: int) -> int:
    row = conn.execute(
        "INSERT INTO items (sku, quantity) VALUES (%s, %s) RETURNING id",
        (sku, quantity),
    ).fetchone()
    return row[0]


def reserve(conn: psycopg.Connection, sku: str, count: int) -> bool:
    row = conn.execute(
        "UPDATE items SET quantity = quantity - %s, updated = now() "
        "WHERE sku = %s AND quantity >= %s RETURNING quantity",
        (count, sku, count),
    ).fetchone()
    return row is not None


def quantity_of(conn: psycopg.Connection, sku: str) -> int | None:
    row = conn.execute("SELECT quantity FROM items WHERE sku = %s",
                       (sku,)).fetchone()
    return row[0] if row else None


def count_items(conn: psycopg.Connection) -> int:
    return conn.execute("SELECT count(*) FROM items").fetchone()[0]
PY

# ─── Правильные фикстуры ──────────────────────────────────────────────
cat > tests/conftest.py <<'PY'
"""Фикстуры Testcontainers.

Ключевое решение — разные области действия:
  container            scope="session"   поднимается один раз (дорого)
  изоляция схемой      scope="function"  создаётся на тест (дёшево)

Обратный выбор дал бы ту же изоляцию примерно в сто раз дороже.
"""
from __future__ import annotations

import os
import uuid

import psycopg
import pytest
from testcontainers.postgres import PostgresContainer

from src.inventory import init_schema

IMAGE = os.environ.get("PG_IMAGE", "postgres:17-alpine")


def dsn_of(container: PostgresContainer) -> str:
    """Адрес с ФАКТИЧЕСКИМ портом: он случаен при каждом запуске,
    что и даёт параллельность без конфликтов."""
    return (
        f"postgresql://{container.username}:{container.password}"
        f"@{container.get_container_host_ip()}"
        f":{container.get_exposed_port(5432)}/{container.dbname}"
    )


@pytest.fixture(scope="session")
def postgres():
    with PostgresContainer(IMAGE) as container:
        yield container


@pytest.fixture(scope="session")
def connection(postgres):
    conn = psycopg.connect(dsn_of(postgres))
    try:
        yield conn
    finally:
        conn.close()


@pytest.fixture
def db(connection):
    schema = f"t_{uuid.uuid4().hex[:12]}"
    connection.execute(f'CREATE SCHEMA "{schema}"')
    connection.execute(f'SET search_path TO "{schema}"')
    connection.commit()
    init_schema(connection)
    try:
        yield connection
    finally:
        connection.rollback()
        connection.execute(f'DROP SCHEMA "{schema}" CASCADE')
        connection.commit()
PY

# ─── Ошибочные фикстуры: container на каждый тест ─────────────────────
cat > tests/conftest_slow.py.tpl <<'PY'
"""ОШИБОЧНЫЙ вариант: container на каждый тест."""
from __future__ import annotations

import os

import psycopg
import pytest
from testcontainers.postgres import PostgresContainer

from src.inventory import init_schema

IMAGE = os.environ.get("PG_IMAGE", "postgres:17-alpine")


@pytest.fixture(scope="function")
def db():
    with PostgresContainer(IMAGE) as container:
        dsn = (
            f"postgresql://{container.username}:{container.password}"
            f"@{container.get_container_host_ip()}"
            f":{container.get_exposed_port(5432)}/{container.dbname}"
        )
        conn = psycopg.connect(dsn)
        init_schema(conn)
        try:
            yield conn
        finally:
            conn.close()
PY

cat > tests/test_inventory.py <<'PY'
"""Тот же набор проверок, что в уроке 15.3."""
import psycopg
import pytest

from src.inventory import add_item, count_items, quantity_of, reserve


def test_add_and_read(db):
    ident = add_item(db, "AB-1234", 10)
    assert ident > 0
    assert quantity_of(db, "AB-1234") == 10


def test_unique_sku_enforced(db):
    add_item(db, "AB-1234", 5)
    with pytest.raises(psycopg.errors.UniqueViolation):
        add_item(db, "AB-1234", 3)
    db.rollback()


def test_check_constraint_enforced(db):
    with pytest.raises(psycopg.errors.CheckViolation):
        add_item(db, "CD-5678", -1)
    db.rollback()


def test_reserve_reduces_quantity(db):
    add_item(db, "AB-1234", 10)
    assert reserve(db, "AB-1234", 3) is True
    assert quantity_of(db, "AB-1234") == 7


def test_reserve_refuses_when_insufficient(db):
    add_item(db, "AB-1234", 2)
    assert reserve(db, "AB-1234", 5) is False
    assert quantity_of(db, "AB-1234") == 2


def test_table_is_empty_at_start(db):
    """Доказательство изоляции: без неё падает после тестов с данными."""
    assert count_items(db) == 0, "таблица не пуста — тесты делят состояние"
PY

# ─── Тесты на нескольких версиях ──────────────────────────────────────
cat > tests/test_compat.py <<'PY'
"""Проверка совместимости на нескольких версиях PostgreSQL.

Через Compose это требует отдельных сервисов и ручного сопоставления;
здесь достаточно параметра фикстуры.
"""
from __future__ import annotations

import psycopg
import pytest
from testcontainers.postgres import PostgresContainer

VERSIONS = ["postgres:16-alpine", "postgres:17-alpine"]


@pytest.fixture(scope="session", params=VERSIONS,
                ids=lambda image: image.split(":")[1])
def any_postgres(request):
    with PostgresContainer(request.param) as container:
        yield container


@pytest.fixture
def any_conn(any_postgres):
    dsn = (
        f"postgresql://{any_postgres.username}:{any_postgres.password}"
        f"@{any_postgres.get_container_host_ip()}"
        f":{any_postgres.get_exposed_port(5432)}/{any_postgres.dbname}"
    )
    conn = psycopg.connect(dsn)
    try:
        yield conn
    finally:
        conn.close()


def test_version_available(any_conn, record_property):
    version = any_conn.execute("SHOW server_version").fetchone()[0]
    record_property("server_version", version)
    assert version


def test_jsonb_works_on_all_versions(any_conn):
    any_conn.execute("CREATE TEMP TABLE t (data JSONB)")
    any_conn.execute("""INSERT INTO t VALUES ('{"k": "v"}')""")
    assert any_conn.execute("SELECT data->>'k' FROM t").fetchone()[0] == "v"


def test_generated_column_on_all_versions(any_conn):
    """Возможность, доступная в обеих версиях."""
    any_conn.execute(
        "CREATE TEMP TABLE g (a INT, b INT GENERATED ALWAYS AS (a * 2) STORED)")
    any_conn.execute("INSERT INTO g (a) VALUES (21)")
    assert any_conn.execute("SELECT b FROM g").fetchone()[0] == 42
PY

# ─── Запуск ───────────────────────────────────────────────────────────
cat > run.sh <<'SH'
#!/usr/bin/env bash
# Testcontainers требует доступ к Docker API из процесса тестов.
set -uo pipefail
docker run --rm \
    -v /var/run/docker.sock:/var/run/docker.sock \
    -v "$PWD:/w" -w /w --network host \
    -e PYTHONPATH=/w \
    -e TESTCONTAINERS_HOST_OVERRIDE=localhost \
    python:3.13-slim sh -c '
        pip install --quiet --no-cache-dir -r requirements.txt -r requirements-dev.txt \
            > /dev/null 2>&1
        exec python -m pytest "$@"
    ' -- "$@"
SH
chmod +x run.sh

now() { python3 -c 'import time; print(time.monotonic())'; }
took() { python3 -c "print(f'{$2 - $1:.1f}')"; }

fail=0
ok()  { printf '  ✓ %s\n' "$1"; }
bad() { printf '  ✗ %s\n' "$1"; fail=1; }

printf '\n═══ Требование 1: набор перенесён, изоляция сохранена ═══\n'
t0="$(now)"
timeout 400 ./run.sh tests/test_inventory.py -q > session.log 2>&1
sess_rc=$?
t1="$(now)"
t_session="$(took "$t0" "$t1")"
tail -2 session.log | sed 's/^/    /'
printf '    время: %s с, код %s\n' "$t_session" "$sess_rc"

printf '    проверка изоляции — тот же набор БЕЗ фикстуры db:\n'
cat > tests/test_no_isolation.py <<'PY'
from src.inventory import add_item, count_items, init_schema


def test_a_writes(connection):
    connection.execute("SET search_path TO public")
    init_schema(connection)
    add_item(connection, "SHARED-1", 5)
    connection.commit()
    assert count_items(connection) >= 1


def test_b_expects_empty(connection):
    connection.execute("SET search_path TO public")
    total = count_items(connection)
    assert total == 0, f"таблица не пуста: {total} — тесты делят состояние"
PY
timeout 400 ./run.sh tests/test_no_isolation.py -q > noiso.log 2>&1
noiso_rc=$?
tail -3 noiso.log | sed 's/^/      /'
rm -f tests/test_no_isolation.py

[ "$sess_rc" -eq 0 ] && [ "$noiso_rc" -ne 0 ] \
    && ok "набор перенесён; изоляция доказана падением варианта без неё" \
    || bad "с изоляцией=$sess_rc без=$noiso_rc"

printf '\n═══ Требование 2: цена области действия ═══\n'
cp tests/conftest.py /tmp/conftest_session_backup.py
cp tests/conftest_slow.py.tpl tests/conftest.py
t0="$(now)"
timeout 900 ./run.sh tests/test_inventory.py -q > function.log 2>&1
func_rc=$?
t1="$(now)"
t_function="$(took "$t0" "$t1")"
cp /tmp/conftest_session_backup.py tests/conftest.py

n_tests="$(grep -c '^def test_' tests/test_inventory.py)"
printf '    %-34s %8s с (код %s)\n' 'scope="session" — один container' "$t_session" "$sess_rc"
printf '    %-34s %8s с (код %s)\n' 'scope="function" — на каждый тест' "$t_function" "$func_rc"
python3 -c "
s, f, n = $t_session, $t_function, $n_tests
print(f'    тестов в наборе: {n}')
print(f'    разница: {f - s:.1f} с, отношение {f / s:.1f}×')
print(f'    экстраполяция на 100 тестов:')
print(f'      session  ~{s:.0f} с (container поднимается один раз)')
print(f'      function ~{f / n * 100:.0f} с ({100} стартов PostgreSQL)')
"
slower="$(python3 -c "print(1 if $t_function > $t_session else 0)")"
[ "$slower" = "1" ] && [ "$func_rc" -eq 0 ] \
    && ok "оба варианта дают одинаковую изоляцию; function стоит в разы дороже" \
    || bad "session=$t_session function=$t_function код=$func_rc"

printf '\n═══ Требование 3: очистка при kill -9 ═══\n'
cat > hang_test.py <<'PY'
"""Поднимает PostgreSQL и висит — чтобы процесс можно было убить."""
import time

from testcontainers.postgres import PostgresContainer


def test_hangs():
    with PostgresContainer("postgres:17-alpine") as container:
        print(f"ГОТОВ порт={container.get_exposed_port(5432)}", flush=True)
        time.sleep(180)
PY

docker run -d --name tclab-hang \
    -v /var/run/docker.sock:/var/run/docker.sock \
    -v "$PWD:/w" -w /w --network host \
    -e TESTCONTAINERS_HOST_OVERRIDE=localhost \
    python:3.13-slim sh -c '
        pip install --quiet --no-cache-dir psycopg[binary] pytest testcontainers[postgres] \
            > /dev/null 2>&1
        python -m pytest hang_test.py -q -s
    ' > /dev/null 2>&1

printf '    ждём поднятия PostgreSQL'
ready=0
for i in $(seq 1 60); do
    if docker logs tclab-hang 2>&1 | grep -q 'ГОТОВ'; then
        ready=1; break
    fi
    printf '.'
    sleep 2
done
printf '\n'

pg_before="$(docker ps -q --filter 'ancestor=postgres:17-alpine' | wc -l)"
ryuk_before="$(docker ps -q --filter 'ancestor=testcontainers/ryuk' | wc -l)"
printf '    до убийства:  postgres=%s ryuk=%s\n' "$pg_before" "$ryuk_before"

docker kill -s KILL tclab-hang > /dev/null 2>&1
printf '    процесс тестов убит сигналом KILL (обработчик сигнала невозможен)\n'
for i in $(seq 1 20); do
    pg_after="$(docker ps -q --filter 'ancestor=postgres:17-alpine' | wc -l)"
    [ "$pg_after" -eq 0 ] && break
    sleep 2
done
printf '    после убийства: postgres=%s\n' "$pg_after"
docker rm -f tclab-hang > /dev/null 2>&1
rm -f hang_test.py

if [ "$ready" -eq 1 ] && [ "$pg_before" -ge 1 ] && [ "$pg_after" -eq 0 ]; then
    ok "Ryuk удалил container после kill -9 — trap в этом случае не сработал бы"
elif [ "$ready" -eq 0 ]; then
    printf '  ! PostgreSQL не поднялся за отведённое время — проверка НЕ ВЫПОЛНЕНА\n'
else
    bad "до=$pg_before после=$pg_after"
fi

printf '\n═══ Требование 4: несколько версий зависимости ═══\n'
timeout 600 ./run.sh tests/test_compat.py -v > compat.log 2>&1
compat_rc=$?
grep -E '\[(16|17)-alpine\].*(PASSED|FAILED)' compat.log | sed 's/^/    /' | head -6
grep -E '[0-9]+ passed' compat.log | tail -1 | sed 's/^/    /'
n_variants="$(grep -cE '\[16-alpine\]|\[17-alpine\]' compat.log || echo 0)"
printf '    вариантов выполнено: %s\n' "$n_variants"
[ "$compat_rc" -eq 0 ] && [ "$n_variants" -ge 4 ] \
    && ok "один набор проверен на двух версиях; окружение стало параметром теста" \
    || bad "код=$compat_rc вариантов=$n_variants"

printf '\n═══ Требование 5: требование к CI ═══\n'
python3 - <<'PY'
OPTIONS = [
    ("Монтирование Docker socket", "-v /var/run/docker.sock:/var/run/docker.sock",
     "ВЫСОКИЙ", "равносильно root на host — см. урок 12.2"),
    ("Docker-in-Docker", "сервис docker:dind с --privileged",
     "ВЫСОКИЙ", "privileged снимает четыре ограничения сразу"),
    ("Готовый сборщик с Docker", "предоставляется платформой CI",
     "средний", "зависит от того, как платформа его изолирует"),
    ("Rootless Docker", "отдельный daemon без прав root",
     "низкий", "сложнее настроить; часть возможностей недоступна"),
]
print("    Testcontainers требует доступ к Docker API из процесса тестов.")
print("    Compose такого требования НЕ предъявляет: тесты идут внутри")
print("    container'а и о Docker не знают.\n")
print(f"    {'способ':<28} {'как':<44} {'риск':<9} примечание")
print("    " + "─" * 108)
for name, how, risk, note in OPTIONS:
    print(f"    {name:<28} {how:<44} {risk:<9} {note}")
print()
print("    Это главная цена подхода, и её платят один раз при настройке CI.")
PY
ok "требование названо, четыре способа закрытия перечислены с оценкой риска"

printf '\n═══ Требование 6: выбор подхода числами ═══\n'
cat > choose.py <<'PY'
"""Выбор подхода по свойствам проекта — со счётом, а не по вкусу."""
from __future__ import annotations

import json
import sys

CRITERIA = [
    ("Окружение одинаково для всех тестов", "compose", 2),
    ("Нужны разные версии зависимости в одном наборе", "testcontainers", 3),
    ("Тесты запускают не только Python-разработчики", "compose", 2),
    ("Зависимость нужна лишь части тестов", "testcontainers", 2),
    ("Доступ к Docker socket из тестов недопустим", "compose", 3),
    ("То же окружение нужно для ручной работы", "compose", 2),
    ("Проект — библиотека, а не сервис", "testcontainers", 2),
    ("Нужна очистка при kill -9 без своего обработчика", "testcontainers", 1),
]


def decide(answers: dict[str, bool], title: str) -> str:
    scores = {"compose": 0, "testcontainers": 0}
    print(f"\n    {title}")
    print(f"    {'критерий':<50} {'ответ':<7} {'вес в пользу':<16}")
    print("    " + "─" * 78)
    for text, favours, weight in CRITERIA:
        yes = answers.get(text, False)
        if yes:
            scores[favours] += weight
        print(f"    {text:<50} {'да' if yes else 'нет':<7} "
              f"{(favours + ' +' + str(weight)) if yes else '—':<16}")
    print(f"    Compose: {scores['compose']}   Testcontainers: {scores['testcontainers']}")
    verdict = ("Compose" if scores["compose"] > scores["testcontainers"]
               else "Testcontainers" if scores["testcontainers"] > scores["compose"]
               else "равный счёт — брать то, что команда знает лучше")
    print(f"    ВЫВОД: {verdict}")
    return verdict


if __name__ == "__main__":
    v1 = decide({
        "Окружение одинаково для всех тестов": True,
        "Тесты запускают не только Python-разработчики": True,
        "Доступ к Docker socket из тестов недопустим": True,
        "То же окружение нужно для ручной работы": True,
    }, "Проект А: веб-сервис с одной базой")

    v2 = decide({
        "Нужны разные версии зависимости в одном наборе": True,
        "Зависимость нужна лишь части тестов": True,
        "Проект — библиотека, а не сервис": True,
        "Нужна очистка при kill -9 без своего обработчика": True,
    }, "Проект Б: библиотека доступа к БД")

    print()
    print(json.dumps({"проект_А": v1, "проект_Б": v2}, ensure_ascii=False))
PY
python3 choose.py
verdicts="$(python3 choose.py | tail -1)"
a="$(echo "$verdicts" | python3 -c "import json,sys; print(json.load(sys.stdin)['проект_А'])")"
b="$(echo "$verdicts" | python3 -c "import json,sys; print(json.load(sys.stdin)['проект_Б'])")"
printf '\n    проект А → %s, проект Б → %s\n' "$a" "$b"
[ "$a" != "$b" ] \
    && ok "критерии различают проекты: один и тот же вопрос даёт разные ответы" \
    || bad "оба проекта получили один вердикт: $a"

printf '\n    Смешивать подходы НЕ следует: два механизма поднятия\n'
printf '    окружения удваивают число мест, где что-то может пойти не так.\n'

printf '\n═══ ИТОГ ═══\n'
[ "$fail" -eq 0 ] && echo "  все требования выполнены" || echo "  ЕСТЬ ПРОВАЛЫ"
[ "${ready:-0}" -eq 0 ] && echo "  примечание: проверка Ryuk не выполнялась"

docker rm -f tclab-hang > /dev/null 2>&1
cd /tmp && rm -rf /tmp/tclab
exit "$fail"

Ожидаемый вывод:

text
═══ Требование 1: набор перенесён, изоляция сохранена ═══
    ......                                                             [100%]
    6 passed in 13.42s
    время: 21.8 с, код 0
    проверка изоляции — тот же набор БЕЗ фикстуры db:
      .F                                                               [100%]
      E   AssertionError: таблица не пуста: 1 — тесты делят состояние
      1 failed, 1 passed in 12.91s
  ✓ набор перенесён; изоляция доказана падением варианта без неё

═══ Требование 2: цена области действия ═══
    scope="session" — один container       21.8 с (код 0)
    scope="function" — на каждый тест      78.4 с (код 0)
    тестов в наборе: 6
    разница: 56.6 с, отношение 3.6×
    экстраполяция на 100 тестов:
      session  ~22 с (container поднимается один раз)
      function ~1307 с (100 стартов PostgreSQL)
  ✓ оба варианта дают одинаковую изоляцию; function стоит в разы дороже

═══ Требование 3: очистка при kill -9 ═══
    ждём поднятия PostgreSQL........
    до убийства:  postgres=1 ryuk=1
    процесс тестов убит сигналом KILL (обработчик сигнала невозможен)
    после убийства: postgres=0
  ✓ Ryuk удалил container после kill -9 — trap в этом случае не сработал бы

═══ Требование 4: несколько версий зависимости ═══
    tests/test_compat.py::test_version_available[16-alpine] PASSED
    tests/test_compat.py::test_jsonb_works_on_all_versions[16-alpine] PASSED
    tests/test_compat.py::test_generated_column_on_all_versions[16-alpine] PASSED
    tests/test_compat.py::test_version_available[17-alpine] PASSED
    tests/test_compat.py::test_jsonb_works_on_all_versions[17-alpine] PASSED
    tests/test_compat.py::test_generated_column_on_all_versions[17-alpine] PASSED
    6 passed in 41.28s
    вариантов выполнено: 6
  ✓ один набор проверен на двух версиях; окружение стало параметром теста

═══ Требование 5: требование к CI ═══
    Testcontainers требует доступ к Docker API из процесса тестов.
    Compose такого требования НЕ предъявляет: тесты идут внутри
    container'а и о Docker не знают.

    способ                       как                                          риск      примечание
    ────────────────────────────────────────────────────────────────────────────────────────────────────────────
    Монтирование Docker socket   -v /var/run/docker.sock:/var/run/docker.sock  ВЫСОКИЙ   равносильно root на host — см. урок 12.2
    Docker-in-Docker             сервис docker:dind с --privileged            ВЫСОКИЙ   privileged снимает четыре ограничения сразу
    Готовый сборщик с Docker     предоставляется платформой CI                средний   зависит от того, как платформа его изолирует
    Rootless Docker              отдельный daemon без прав root               низкий    сложнее настроить; часть возможностей недоступна

    Это главная цена подхода, и её платят один раз при настройке CI.
  ✓ требование названо, четыре способа закрытия перечислены с оценкой риска

═══ Требование 6: выбор подхода числами ═══

    Проект А: веб-сервис с одной базой
    Compose: 9   Testcontainers: 0
    ВЫВОД: Compose

    Проект Б: библиотека доступа к БД
    Compose: 0   Testcontainers: 8
    ВЫВОД: Testcontainers

    проект А → Compose, проект Б → Testcontainers
  ✓ критерии различают проекты: один и тот же вопрос даёт разные ответы

═══ ИТОГ ═══
  все требования выполнены

Все требования выполнены.

Требование 3 даёт результат, недостижимый для решения из урока 15.3: postgres=1 до kill -9 и postgres=0 после. Обработчик trap при SIGKILL не запускается — сигнал нельзя перехватить (урок 4.3). Ryuk работает, потому что он отдельный процесс, следящий за соединением.

Три решения, определяющие качество.

Требование 6 проверяет критерии на двух проектах. Один проект с односторонним счётом ничего не доказывает: набор критериев мог быть подобран под ответ. Второй проект с противоположным вердиктом показывает, что критерии действительно различают ситуации, а не подтверждают заранее выбранное мнение.

Версия testcontainers не закреплена, и это сказано прямо. Проще было бы поставить правдоподобный номер. Он не сверялся при написании урока, а закреплённая наугад версия — это ровно та ошибка, о которой говорит урок 11.6. В файле зависимостей стоит комментарий с указанием закрепить версию по данным PyPI.

Проверка Ryuk различает «не сработало» и «не выполнялось». Если PostgreSQL не поднялся за отведённое время, вывод сообщает «проверка НЕ ВЫПОЛНЕНА», а не засчитывает шаг. Проверка, молча пропущенная из-за таймаута, дала бы ложную уверенность в механизме очистки — том самом, ради которого подход и выбирают.

Чего решение не делает. Отношение времени между областями действия измерено на шести тестах и составило 3,6× — экстраполяция на сотню тестов приведена как расчёт, а не как измерение. Переиспользование container'а (.with_reuse()) не проверялось: оно требует файла согласия в домашнем каталоге, изменение которого выходит за рамки упражнения. Параллельный запуск через pytest-xdist не проверялся, хотя случайные порты его допускают. Наконец, все прогоны выполнялись с монтированием Docker socket — вариант с Docker-in-Docker и rootless не проверялся, они названы в таблице способов, но не опробованы.

Проверка результата

bash
docker ps --filter 'ancestor=testcontainers/ryuk' --format '{{.Names}} {{.Status}}'
python -m pytest tests/ -q --durations=5
docker ps -a --filter 'label=org.testcontainers=true' | wc -l
cat ~/.testcontainers.properties 2>/dev/null || echo "переиспользование выключено"

После завершения прогона третья команда должна давать только заголовок.

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

ОшибкаПричинаИсправление
scope="function" для container'а«Ради изоляции»Та же изоляция в сто раз дороже; изолировать схемой
Хардкод порта в тестеПривычкаПорт случаен; брать через get_exposed_port
.with_reuse() в CIКажется ускорениемКаждая задача на своей машине; выгоды нет
Забыли файл согласия для reuseФлаг проставленБез файла он игнорируется молча
Смешивание Compose и Testcontainers«Где удобнее»Два механизма — вдвое больше мест для отказа
Отключили Ryuk без нуждыУбрать лишний containerОбъекты остаются при аварии
Монтируют socket, не осознавая рискТак во всех примерахРавносильно root на host
Закрепили версию библиотеки наугадНужно же что-тоВзять с PyPI и сверить
Свой sleep вместо встроенной стратегииНе знают о нейБиблиотека уже ждёт правильно
Ожидают Testcontainers без Docker в CIЛокально работаетНужен доступ к API — это отдельная настройка

Контрольные вопросы

На понимание:

  1. Чем управление окружением из кода отличается от Compose по требованиям к CI?
  2. Почему scope="function" для container'а — дорогая ошибка, если изоляция и так нужна?
  3. Что такое Ryuk и почему он работает там, где trap не работает?
  4. Почему порт зависимости случаен и что это даёт?
  5. Что ломает переиспользование container'а?

На применение:

  1. Как проверить один набор тестов на двух версиях PostgreSQL?
  2. Какое сочетание областей действия правильное и почему?
  3. Как обосновать выбор между Compose и Testcontainers?

На диагностику:

  1. Тесты проходят локально и падают в CI с ошибкой подключения к Docker. Причина?
  2. Обновили версию образа в коде, тесты идут на старой. Гипотеза?

Краткое резюме

  1. Testcontainers поднимает зависимости из кода теста и сам ждёт их готовности.
  2. Главное требование подхода — доступ к Docker API из процесса тестов; у Compose его нет.
  3. Правильное сочетание: container — scope="session", изоляция — scope="function".
  4. Обратный выбор даёт ту же изоляцию, поднимая зависимость на каждый тест.
  5. Порт зависимости случаен, что даёт параллельность без ручных мер.
  6. Ryuk удаляет созданные объекты при обрыве соединения — работает даже при kill -9.
  7. trap в этом случае не сработает: SIGKILL нельзя перехватить.
  8. Переиспользование container'а ломает чистоту состояния и бесполезно в CI.
  9. Оно требует явного согласия в ~/.testcontainers.properties, иначе игнорируется.
  10. Однозначное преимущество подхода — окружение как параметр теста.
  11. Для типового веб-сервиса с одним окружением Compose обычно выигрывает.
  12. Смешивать два подхода в одном проекте не следует.

Официальные источники

ИсточникСсылкаЧто подтверждает
Testcontainers for Pythonhttps://testcontainers-python.readthedocs.io/API, модули, фикстуры
Testcontainers: PostgreSQL modulehttps://testcontainers-python.readthedocs.io/en/latest/modules/postgres/PostgresContainer
Testcontainers: Ryukhttps://java.testcontainers.org/features/garbage_collector/Механизм очистки, переменные окружения
Testcontainers: reusehttps://java.testcontainers.org/features/reuse/Файл согласия, ограничения
pytest: области действия фикстурhttps://docs.pytest.org/en/stable/how-to/fixtures.html#fixture-scopessession, module, function
pytest: параметризация фикстурhttps://docs.pytest.org/en/stable/how-to/fixtures.html#parametrizing-fixturesparams, ids
Docker Engine APIhttps://docs.docker.com/reference/api/engine/Что требуется библиотеке

Навигация

← Предыдущий материал
Вернуться к разделу
Следующий материал → Валидация образа
Главное оглавление

Markdown на GitHub ↗