Что НЕ покрывается интеграционными тестами (Python / pytest)

Опирается на правила: PYTS-26, PYTS-27, PYTS-28, PYTS-X5, PYTS-X6 из Python Test Strategy → раздел 8. Что НЕ покрывается интеграционными тестами.

Важно знать

• Интеграционный тест (AsyncClient + Testcontainers) — один слой из трёх, не универсальный инструмент.
• Доменная логика агрегата (Order, Product) — unit без фреймворка, самые быстрые тесты.
• Контроллер + сериализация без БД — AsyncClient + app.dependency_overrides на in-memory фейк репозитория.
• E2E через настоящие Kafka / внешние сервисы — отдельная группа @pytest.mark.e2e, ≤ 5–10 на сервис.
• MagicMock на Handler / Aggregate / порт-репозиторий в интеграционном тесте запрещён (PYTS-X5).
• Живой Keycloak или JWT, собранный руками, — тоже запрещён (PYTS-X6); только success_token().
• Три зоны дополняют друг друга: unit → integration → e2e; не заменяют.

Интеграционный тест с реальным Postgres через Testcontainers проверяет самое главное — что весь путь от HTTP до БД работает как единое целое. Но три зоны проверяются вне этого слоя: они либо быстрее, либо изолированнее, либо требуют реальной внешней инфраструктуры. Размывать границы дорого: unit вместо integration медленнее не станет, но integration вместо unit замедлит suite в 10–100 раз.

Зона 1 — чистая доменная логика агрегата (PYTS-26)

Агрегаты (Order, Product, Customer) содержат бизнес-инварианты: что можно делать в каком статусе, как считается итог, какие события публикуются. Эти правила не зависят от FastAPI, SQLAlchemy и Postgres — тестируем как обычные Python-объекты.

# tests/unit/domain/test_order.py

from datetime import datetime, timezone
from uuid import UUID

import pytest

from app.domain.order import Order, OrderStatus


ORDER_ID = UUID("11111111-1111-1111-1111-111111111111")
CUSTOMER_ID = UUID("22222222-2222-2222-2222-222222222222")
NOW = datetime(2026, 5, 26, 10, 0, 0, tzinfo=timezone.utc)


def test_confirm_when_draft_changes_status_to_confirmed():
    """BR-001: подтверждение заказа в статусе DRAFT переводит в CONFIRMED."""
    order = Order(id=ORDER_ID, customer_id=CUSTOMER_ID, status=OrderStatus.DRAFT)

    order.confirm(confirmed_at=NOW)

    assert order.status == OrderStatus.CONFIRMED
    assert order.confirmed_at == NOW


def test_confirm_when_reservation_failed_raises_domain_error():
    """BR-002: подтверждение заказа со статусом RESERVATION_FAILED запрещено."""
    order = Order(
        id=ORDER_ID,
        customer_id=CUSTOMER_ID,
        status=OrderStatus.RESERVATION_FAILED,
    )

    with pytest.raises(ValueError, match="OUT_OF_STOCK"):
        order.confirm(confirmed_at=NOW)


def test_cancel_when_confirmed_publishes_order_cancelled_event():
    """BR-005: отмена подтверждённого заказа публикует событие OrderCancelled."""
    order = Order(id=ORDER_ID, customer_id=CUSTOMER_ID, status=OrderStatus.CONFIRMED)

    order.cancel(reason="customer_request")

    events = order.pull_domain_events()
    assert len(events) == 1
    assert events[0].event_type == "OrderCancelled"
    assert events[0].order_id == ORDER_ID

Unit-тесты агрегата:

• Нет @pytest.mark.asyncio — логика синхронная, pytest запускает мгновенно.
• Нет фикстур с контейнерами — тест занимает миллисекунды.
• Покрывают каждое BR-правило — самый дешёвый способ зафиксировать инвариант.

Их должно быть больше всего в проекте: десятки unit-тестов агрегата на каждый интеграционный тест бизнес-операции.

Зона 2 — контроллер и сериализация без БД (PYTS-27)

Иногда нужно проверить, что контроллер правильно парсит тело запроса, возвращает нужные HTTP-статусы при валидационных ошибках или корректно сериализует ответ. Поднимать Postgres ради этого — лишняя сложность.

Паттерн: AsyncClient(transport=ASGITransport(app)) + app.dependency_overrides на in-memory фейковый репозиторий.

# tests/integration/controller/test_create_order_controller.py

import pytest
from httpx import AsyncClient, ASGITransport

from app.main import app
from app.port.order_repository import OrderRepository
from tests.fixtures.auth import success_token


class InMemoryOrderRepository(OrderRepository):
    def __init__(self):
        self._orders: dict = {}

    async def save(self, order) -> None:
        self._orders[order.id] = order

    async def find_by_id(self, order_id):
        return self._orders.get(order_id)


@pytest.fixture
def in_memory_repo():
    return InMemoryOrderRepository()


@pytest.fixture
async def client_with_fake_repo(in_memory_repo):
    app.dependency_overrides[OrderRepository] = lambda: in_memory_repo
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        yield c, in_memory_repo
    app.dependency_overrides.clear()


@pytest.mark.asyncio
async def test_create_order_returns_400_when_customer_id_missing(client_with_fake_repo):
    """Валидация: тело без customer_id → 422 Unprocessable Entity."""
    client, _ = client_with_fake_repo

    response = await client.post("/v1/orders", json={"items": []})

    assert response.status_code == 422
    errors = response.json()["detail"]
    assert any(e["loc"] == ["body", "customer_id"] for e in errors)


@pytest.mark.asyncio
async def test_create_order_returns_201_with_location_header(client_with_fake_repo):
    """Контроллер возвращает Location-заголовок с URL созданного заказа."""
    client, _ = client_with_fake_repo

    response = await client.post(
        "/v1/orders",
        json={"customer_id": "22222222-2222-2222-2222-222222222222", "items": []},
        headers=success_token(),
    )

    assert response.status_code == 201
    assert "location" in response.headers
    assert response.headers["location"].startswith("/v1/orders/")

Что не делаем здесь:

• Не поднимаем PostgresContainer.
• Не применяем миграции.
• Не тестируем бизнес-логику Handler-а — только HTTP-контракт контроллера.

Зона 3 — E2E через настоящие Kafka и внешние сервисы (PYTS-28)

E2E-тесты проверяют конечный сценарий через несколько сервисов: заказ создан → событие попало в Kafka → Consumer обработал → статус изменился. Это отдельная группа, не часть дефолтного integration suite.

# tests/e2e/test_order_created_saga.py

import pytest

pytestmark = pytest.mark.e2e


async def test_order_confirmed_event_reaches_fulfillment_consumer(
    real_kafka_producer,
    real_kafka_consumer,
    http_client,
):
    """E2E: подтверждение заказа → событие OrderConfirmed → Fulfillment Consumer обновляет статус."""
    response = await http_client.post(
        "/v1/orders/11111111-1111-1111-1111-111111111111/confirm",
        headers={"Authorization": "Bearer real-token"},
    )
    assert response.status_code == 200

    message = await real_kafka_consumer.poll(timeout=10.0)
    assert message is not None
    assert message.value["event_type"] == "OrderConfirmed"

Отдельный CI-этап для E2E:

# .github/workflows/e2e.yml
- name: Run E2E tests
  run: pytest tests/e2e -m e2e --timeout=120

Правила E2E-группы:

• ≤ 5–10 тестов на сервис — только критичные пути, не полное покрытие.
• Отдельный CI-этап — не блокирует PR merge; запускается после деплоя на staging.
• Разрешены asyncio.sleep / polling — это smoke по природе, не бизнес-логика.
• Маркер @pytest.mark.e2e — дефолтный pytest.ini исключает их (addopts = -m "not e2e").

Конфигурация маркера:

# pytest.ini
[pytest]
asyncio_mode = auto
addopts = -m "not e2e"
markers =
    e2e: End-to-end tests requiring real Kafka, external services and staging environment

Почему MagicMock на бизнес-логику запрещён (PYTS-X5)

# ЗАПРЕЩЕНО — интеграционный тест с мок-Handler-ом не проверяет реальный путь
@pytest.mark.asyncio
async def test_confirm_order_wrong(client, mocker):
    mock_handler = mocker.patch("app.usecase.confirm_order.ConfirmOrderHandler.handle")
    mock_handler.return_value = ConfirmOrderResponse(order_id=ORDER_ID, status="CONFIRMED")

    response = await client.post(f"/v1/orders/{ORDER_ID}/confirm")

    assert response.status_code == 200

Что не так:

• Тест не проверил ни одну строку бизнес-логики.
• MagicMock возвращает заглушку — реальная ошибка в Handler пройдёт незамеченной.
• Это unit-тест контроллера, замаскированный под integration.

MagicMock в интеграционном тесте допустим только для внешней системы (платёжный шлюз, каталог, логистика), и только через respx или pytest-httpserver — не через патч бизнес-объекта.

Итог: три слоя, три инструмента

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

Куда дальше

• python/basics.md — базовые правила: детерминизм, AsyncClient, AAA-структура.
• python/database-preparer.md — fluent setup БД для integration-тестов.
• python/no-kafka-redis-async.md — почему Kafka/Redis не поднимаем в базовом suite.
• Внешние HTTP — respx / pytest-httpserver — стабы внешних REST-вызовов.