Базовые правила — интеграционные тесты и AAA-структура на Python

Опирается на правила: PYTS-1, PYTS-2, PYTS-3, PYTS-X1, PYTS-X2 из Python Test Strategy → раздел 1. Базовые правила.

Важно знать

• Тест быстрый и детерминированный. asyncio.sleep/polling/tenacity — признак недетерминированного дизайна, не способ «дождаться» результата.
• Интеграционный тест = httpx.AsyncClient(transport=ASGITransport(app)) + реальный PostgreSQL (Testcontainers) + вызов роутера по HTTP.
• Внешние HTTP-сервисы (платёж, каталог, логистика) — мок через pytest-httpserver или respx; не трогаем реальные endpoints.
• Kafka и Redis не поднимаем — профиль integration-test отключает брокер и кеш; события проверяем через Outbox-таблицу.
• Время и UUID фиксированы через app.dependency_overrides на get_clock/get_id_generator.
• Один тест — один сценарий, AAA-структура (# Arrange, # Act, # Assert).
• pytest-asyncio с asyncio_mode=auto; дорогой setup (контейнер, схема) — session-scope, не per-test.

UCP-подход к тестам в Python строится вокруг той же идеи, что и в Java и Go: тест быстрый (миллисекунды, не секунды), детерминированный (одинаковый результат при каждом прогоне), простой (один сценарий). Разница в инструментах: вместо Spring-контекста — FastAPI app через ASGITransport, вместо @MockitoBean — dependency_overrides, вместо Awaitility — его полное отсутствие.

Что входит в интеграционный тест

PYTS-1: полный стек внутри сервиса, минус внешние системы.

Это даёт:

• End-to-end внутри сервиса — от HTTP-запроса до строки в PostgreSQL.
• Без асинхронных рассинхронизаций — нет ожидания Kafka consumer-а.
• Время прогона — один тест занимает 20–150 мс; контейнер поднимается один раз в session-scope фикстуре.

@pytest.mark.asyncio
async def test_create_order_success(client: AsyncClient, db_preparer: OrderDatabasePreparer):
    order_id = uuid.UUID("11111111-1111-1111-1111-111111111111")
    now = datetime(2026, 5, 26, 10, 0, 0, tzinfo=UTC)

    # Arrange
    db_preparer.clear_orders().create_customer("c1").prepare()
    app.dependency_overrides[get_id_generator] = lambda: FixedIdGenerator(next_id=order_id)
    app.dependency_overrides[get_clock] = lambda: FixedClock(at=now)

    # Act
    response = await client.post(
        "/v1/orders",
        json={"customerId": "c1", "amount": 100},
        headers=success_token(),
    )

    # Assert
    assert response.status_code == 201
    body = response.json()
    assert body["orderId"] == str(order_id)

    outbox = await db_preparer.find_outbox_events()
    assert [e["event_type"] for e in outbox] == ["OrderCreated"]

Все тесты детерминированные

PYTS-2: никаких asyncio.sleep, while-poll или tenacity-ожиданий.

# ПЛОХО — тест зависит от скорости CI-машины
await asyncio.sleep(0.5)
order = await find_order(session, order_id)
assert order is not None

# ПЛОХО — polling скрывает недетерминированный дизайн
async def wait_for_order():
    for _ in range(50):
        o = await find_order(session, order_id)
        if o:
            return o
        await asyncio.sleep(0.1)
    raise AssertionError("order not found")

Что плохо:

• Мигающий — на медленном CI падает, на быстром проходит; «починка» — увеличить timeout, снова мигает.
• Медленный — каждый sleep/poll — сотни миллисекунд; сотня тестов превращается в минуты.
• Скрывает баги — асинхронная ошибка «событие опоздало на 80 мс» проходит при poll(5s), но в production клиент уже получил timeout.

Корректный подход — синхронный поток:

1. Clock через dependency_overrides[get_clock] → FixedClock с заранее заданным значением.
2. IdGenerator через dependency_overrides[get_id_generator] → генератор с фиксированной последовательностью.
3. Outbox-relay вызывается явно: await relay.process_pending(), не ждём фонового воркера.
4. Kafka/Redis отключены профилем — нет ожидания consumer-а.

class FixedClock:
    def __init__(self, at: datetime) -> None:
        self._at = at

    def now(self) -> datetime:
        return self._at


class SeqIdGenerator:
    def __init__(self) -> None:
        self._counter = 0

    def next_id(self) -> uuid.UUID:
        self._counter += 1
        return uuid.UUID(int=self._counter)

PYTS-X2: прямые вызовы datetime.now() и uuid.uuid4() в доменном коде (Handler/Service/Aggregate) запрещены — делают тест недетерминированным.

AAA-структура

PYTS-3: один тест — один сценарий, три блока с пустой строкой между ними.

@pytest.mark.asyncio
async def test_confirm_order_when_reservation_failed_returns_409(
    client: AsyncClient,
    db_preparer: OrderDatabasePreparer,
):
    """BR-ORDER-2: нельзя подтвердить заказ со статусом RESERVATION_FAILED."""
    order_id = uuid.UUID("11111111-1111-1111-1111-111111111111")

    # Arrange
    order = OrderObjectGenerator().with_id(order_id).with_status("RESERVATION_FAILED").build()
    db_preparer.clear_orders().create_order(order).prepare()

    # Act
    response = await client.post(
        f"/v1/orders/{order_id}/confirm",
        headers=success_token(),
    )

    # Assert
    assert response.status_code == 409
    body = response.json()
    assert body["code"] == "OUT_OF_STOCK"

    orders = await db_preparer.find_orders(order_id=order_id)
    assert orders[0]["status"] == "RESERVATION_FAILED"

Каждый блок — одна ответственность:

• Arrange — очистка БД, создание фикстур, настройка заглушек времени/UUID.
• Act — один HTTP-вызов.
• Assert — проверки ответа и побочных эффектов (строки в БД, Outbox-события).

Не «проверить весь жизненный цикл в одном тесте»

# ПЛОХО — мега-тест проверяет пять переходов состояния
async def test_order_lifecycle(client, db_preparer):
    # создать заказ
    # подтвердить
    # оплатить
    # отгрузить
    # отменить
    ...

Что плохо:

• Если падает шаг 3 — неизвестно, что не так с шагами 4–5.
• Имя теста не описывает ни одного конкретного сценария.
• Нельзя запустить один сценарий в изоляции.

Корректно — отдельные функции:

async def test_create_order_success(client, db_preparer): ...
async def test_confirm_order_when_draft_returns_200(client, db_preparer): ...
async def test_pay_order_when_confirmed_returns_200(client, db_preparer): ...
async def test_cancel_order_when_confirmed_returns_200(client, db_preparer): ...

Имена тестов

PYTS-16: формат test_<action>_when_<condition>_<expected>, docstring с цитатой BR-кода из спецификации.

async def test_create_order_when_customer_not_found_returns_404(client, db_preparer):
    """BR-ORDER-1: клиент должен существовать."""
    ...

async def test_confirm_order_when_reservation_failed_returns_409(client, db_preparer):
    """BR-ORDER-2: нельзя подтвердить заказ со статусом RESERVATION_FAILED."""
    ...

async def test_get_product_when_not_found_returns_404(client, db_preparer):
    """BR-PROD-5: продукт с несуществующим ID → 404."""
    ...

Docstring с BR-кодом создаёт traceability: спецификация → тест → код. Изменили BR-ORDER-2 — сразу видно, какие тесты нужно обновить.

Что запрещено

Куда дальше

• Фикстуры и конфигурация тестов — conftest.py, PostgresContainer, ASGITransport, session-scope.
• DatabasePreparer — fluent setup БД: clear_*(), create_*(...), prepare().
• ObjectGenerator — fluent builders с разумными дефолтами.
• Один тест — полный пример, имена, docstring с BR-кодом.
• Kafka, Redis, async — по умолчанию НЕТ — Outbox-подход, прямой вызов relay.
• Пирамида тестов — что чем покрывать: unit агрегата, unit контроллера, интеграционный, E2E.