Промпт: интеграционные тесты для API

1. Таблица тест-кейсов

---

2. Пример реализации тестов

Фикстуры и фабрики данных (tests/conftest.py)

import asyncio
from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from fastapi.testclient import TestClient
from testcontainers.postgres import PostgresContainer
from app.main import app
from app.models import Base
from app.db import get_db


@pytest.fixture(scope="session")
def event_loop() -> Generator[asyncio.AbstractEventLoop, None, None]:
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()


@pytest.fixture(scope="session")
async def postgres_container() -> AsyncGenerator[str, None]:
    with PostgresContainer("postgres:latest") as pg:
        uri = f"postgresql+asyncpg://{pg.get_username()}:{pg.get_password()}@{pg.get_hostname()}:{pg.get_port()}/{pg.get_db_name()}"
        engine = create_async_engine(uri)
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
        yield uri
        await engine.dispose()


@pytest.fixture(scope="function")
async def session(postgres_container):
    engine = create_async_engine(postgres_container)
    SessionLocal = async_sessionmaker(bind=engine, expire_on_commit=False)
    async with SessionLocal() as session:
        yield session
    await engine.dispose()


@pytest.fixture(scope="module")
def client(session):
    # Замещаем зависимость get_db для тестирования
    def override_get_db():
        return session

    app.dependency_overrides[get_db] = override_get_db
    return TestClient(app)

Примеры тестов (tests/test_endpoints.py)

import pytest
from datetime import timedelta, datetime
from httpx import AsyncClient
from app.schemas.promo import ApplyPromo
from app.models import User, Cart, Promo
from tests.factories import PromoFactory, UserFactory, CartFactory


# Вводим дополнительную фикстуру для работы с асинхронностью
@pytest.mark.anyio
class TestApplyPromoEndpoint:
    @pytest.fixture(autouse=True)
    async def setup(self, session):
        self.user = await UserFactory.create()
        self.cart = await CartFactory.create(user=self.user)
        self.promo = await PromoFactory.create(max_uses=5, percent=10, expires_at=datetime.utcnow() + timedelta(days=7))

    async def test_valid_promocode(self, client):
        response = await client.post(
            "/promo/apply",
            json={"code": self.promo.code},
            headers={"Authorization": f"Bearer {self.user.token}"},  # Добавляем заголовки аутентификации
        )
        assert response.status_code == 200
        data = response.json()
        assert "discount" in data and isinstance(data["discount"], float)
    
    async def test_expired_promocode(self, client):
        expired_promo = await PromoFactory.create(expires_at=datetime.utcnow() - timedelta(hours=1))
        
        response = await client.post(
            "/promo/apply",
            json={"code": expired_promo.code},
            headers={"Authorization": f"Bearer {self.user.token}"},
        )
        assert response.status_code == 404
        assert response.json()["detail"] == "Промокод не найден"

    async def test_used_up_promocode(self, client):
        fully_used_promo = await PromoFactory.create(max_uses=0)
        
        response = await client.post(
            "/promo/apply",
            json={"code": fully_used_promo.code},
            headers={"Authorization": f"Bearer {self.user.token}"},
        )
        assert response.status_code == 409
        assert response.json()["detail"] == "Промокод закончился"

    async def test_duplicate_application(self, client):
        first_response = await client.post(
            "/promo/apply",
            json={"code": self.promo.code},
            headers={"Authorization": f"Bearer {self.user.token}"},
        )
        second_response = await client.post(
            "/promo/apply",
            json={"code": self.promo.code},
            headers={"Authorization": f"Bearer {self.user.token}"},
        )
        assert first_response.status_code == 200
        assert second_response.status_code == 200  # Используется повторно, but count должен остаться прежним
        applied_promo = await session.get(Promo, self.promo.id)
        assert applied_promo.used_count == 1

    async def test_unauthorized_access(self, client):
        response = await client.post(
            "/promo/apply",
            json={"code": self.promo.code},
        )  # Без Authorization header
        assert response.status_code == 401

Фабрики данных (tests/factories.py)

from factory.alchemy import AsyncSQLAlchemyModelFactory
from app.models import User, Cart, Promo
from tests.conftest import session


class UserFactory(AsyncSQLAlchemyModelFactory):
    class Meta:
        model = User
        sqlalchemy_session = session
        sqlalchemy_session_persistence = 'commit'

    id = factory.Sequence(lambda n: n)
    email = factory.Faker('email')
    token = factory.Faker('uuid4')  # Простая заглушка для токенов


class CartFactory(AsyncSQLAlchemyModelFactory):
    class Meta:
        model = Cart
        sqlalchemy_session = session
        sqlalchemy_session_persistence = 'commit'

    id = factory.Sequence(lambda n: n)
    total = 100.00
    discount = 0.00
    user_id = factory.SelfAttribute('..id')


class PromoFactory(AsyncSQLAlchemyModelFactory):
    class Meta:
        model = Promo
        sqlalchemy_session = session
        sqlalchemy_session_persistence = 'commit'

    code = factory.Sequence(lambda n: f"PROMO{n}")
    max_uses = 5
    percent = 10
    max_discount = 100.00
    expires_at = factory.LazyFunction(lambda: datetime.utcnow() + timedelta(days=7))

---

3. Запуск тестов в CI и изоляция данных

Чтобы запустить эти тесты в CI-системе, мы используем testcontainers для запуска контейнера базы данных PostgreSQL, а также создаём отдельную схему БД для каждого теста, чтобы обеспечить полную изоляцию данных между тестовыми кейсами.

Пример конфигурационного файла .github/workflows/test.yml, запускающий тесты в GitHub Actions:

name: Run Integration Tests

on:
  push:
    branches: [main]
  pull_request:
    types: [opened, synchronize, reopened]

jobs:
  integration_tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      
      - name: Set up Python 3.12
        uses: actions/setup-python@v4
        with:
          python-version: 3.12
          
      - name: Install dependencies
        run: pip install -r requirements.txt
      
      - name: Run tests
        env:
          DATABASE_URL: ${{ secrets.DATABASE_URL }}
        run: pytest --asyncio-mode=auto tests/

Изоляция данных:
Каждый тест получает свою собственную сессию баз данных благодаря использованию отдельного механизма сессии в фикстурах. Это гарантирует, что каждый тест работает в чистой среде без взаимозависимости друг от друга.

Таким образом, после выполнения всех шагов, база данных очищается автоматически, и последующие тесты выполняются независимо.

Ниже — набор интеграционных тестов для POST /promo/apply на реальной PostgreSQL через testcontainers.

Я буду исходить из типичной структуры проекта:

app/
  main.py              # app = FastAPI()
  api/deps.py          # get_db, current_user
  models.py            # User, Cart, Promo
  schemas.py
tests/
  conftest.py
  test_apply_promo.py

Импорты в примерах нужно будет подправить под ваш проект.

---

1. Тест-кейсы

---

2. Код тестов

tests/conftest.py

import asyncio
from collections.abc import AsyncGenerator, Generator
from datetime import datetime, timedelta, timezone
from decimal import Decimal
from typing import Any

import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy import text
from sqlalchemy.ext.asyncio import (
    AsyncEngine,
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from testcontainers.postgres import PostgresContainer

# Подправьте импорты под ваш проект
from app.main import app
from app.api.deps import get_db, current_user
from app.models import Base, User, Cart, Promo


@pytest.fixture(scope="session")
def event_loop() -> Generator[asyncio.AbstractEventLoop, None, None]:
    """
    Для pytest-asyncio.
    Если у вас pytest-asyncio >= 0.23 и настроен asyncio_mode=auto,
    эта фикстура может быть не нужна.
    """
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()


@pytest.fixture(scope="session")
def postgres_container() -> Generator[PostgresContainer, None, None]:
    """
    Реальная PostgreSQL в Docker.
    """
    with PostgresContainer("postgres:16-alpine") as postgres:
        yield postgres


@pytest.fixture(scope="session")
def database_url(postgres_container: PostgresContainer) -> str:
    """
    testcontainers отдаёт sync URL вида postgresql+psycopg2://...
    Для async SQLAlchemy нужен asyncpg.
    """
    url = postgres_container.get_connection_url()
    return url.replace("postgresql+psycopg2://", "postgresql+asyncpg://")


@pytest_asyncio.fixture(scope="session")
async def engine(database_url: str) -> AsyncGenerator[AsyncEngine, None]:
    engine = create_async_engine(
        database_url,
        echo=False,
        pool_pre_ping=True,
    )

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    yield engine

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

    await engine.dispose()


@pytest_asyncio.fixture
async def db_session(engine: AsyncEngine) -> AsyncGenerator[AsyncSession, None]:
    """
    На каждый тест отдельная сессия.
    Данные чистятся truncate'ом после теста.
    """
    session_factory = async_sessionmaker(
        engine,
        expire_on_commit=False,
        autoflush=False,
    )

    async with session_factory() as session:
        yield session

    async with engine.begin() as conn:
        # Под PostgreSQL. RESTART IDENTITY сбрасывает sequence.
        # CASCADE чистит зависимые таблицы.
        await conn.execute(
            text(
                """
                TRUNCATE TABLE
                    carts,
                    promos,
                    users
                RESTART IDENTITY CASCADE
                """
            )
        )


@pytest_asyncio.fixture
async def app_db_override(engine: AsyncEngine):
    """
    Переопределяем get_db так, чтобы приложение ходило в тестовую БД.
    Важно: создаём новую сессию на каждый request,
    как в реальном приложении.
    """
    session_factory = async_sessionmaker(
        engine,
        expire_on_commit=False,
        autoflush=False,
    )

    async def override_get_db() -> AsyncGenerator[AsyncSession, None]:
        async with session_factory() as session:
            yield session

    app.dependency_overrides[get_db] = override_get_db
    yield
    app.dependency_overrides.pop(get_db, None)


@pytest_asyncio.fixture
async def user_factory(db_session: AsyncSession):
    async def create_user(**kwargs: Any) -> User:
        user = User(
            email=kwargs.get("email", "user@example.com"),
            # Добавьте обязательные поля вашей модели:
            # hashed_password=kwargs.get("hashed_password", "test"),
            # is_active=kwargs.get("is_active", True),
        )
        db_session.add(user)
        await db_session.commit()
        await db_session.refresh(user)
        return user

    return create_user


@pytest_asyncio.fixture
async def cart_factory(db_session: AsyncSession):
    async def create_cart(user: User, **kwargs: Any) -> Cart:
        cart = Cart(
            user_id=user.id,
            total=kwargs.get("total", Decimal("1000.00")),
            discount=kwargs.get("discount", Decimal("0.00")),
            # Если в модели есть promo_id/applied_promo_code — заполните при необходимости.
        )
        db_session.add(cart)
        await db_session.commit()
        await db_session.refresh(cart)
        return cart

    return create_cart


@pytest_asyncio.fixture
async def promo_factory(db_session: AsyncSession):
    async def create_promo(**kwargs: Any) -> Promo:
        promo = Promo(
            code=kwargs.get("code", "SALE10"),
            percent=kwargs.get("percent", Decimal("10.00")),
            max_discount=kwargs.get("max_discount", Decimal("500.00")),
            used_count=kwargs.get("used_count", 0),
            max_uses=kwargs.get("max_uses", 10),
            expires_at=kwargs.get(
                "expires_at",
                datetime.now(timezone.utc) + timedelta(days=1),
            ),
        )
        db_session.add(promo)
        await db_session.commit()
        await db_session.refresh(promo)
        return promo

    return create_promo


@pytest_asyncio.fixture
async def auth_user(user_factory, cart_factory):
    user = await user_factory(email="auth@example.com")
    await cart_factory(user=user, total=Decimal("1000.00"))
    return user


@pytest_asyncio.fixture
async def authorized_client(app_db_override, auth_user: User) -> AsyncGenerator[AsyncClient, None]:
    """
    Авторизованный клиент.
    Внешнюю авторизацию/сервис пользователя мокируем через dependency override.
    """
    async def override_current_user():
        return auth_user

    app.dependency_overrides[current_user] = override_current_user

    transport = ASGITransport(app=app, raise_app_exceptions=False)

    async with AsyncClient(
        transport=transport,
        base_url="http://test",
    ) as client:
        yield client

    app.dependency_overrides.pop(current_user, None)


@pytest_asyncio.fixture
async def anonymous_client(app_db_override) -> AsyncGenerator[AsyncClient, None]:
    """
    Клиент без override current_user.
    Нужен для проверки прав доступа.
    """
    transport = ASGITransport(app=app, raise_app_exceptions=False)

    async with AsyncClient(
        transport=transport,
        base_url="http://test",
    ) as client:
        yield client

---

tests/test_apply_promo.py

import asyncio
from datetime import datetime, timedelta, timezone
from decimal import Decimal

import pytest
from sqlalchemy import select

from app.models import Cart, Promo


pytestmark = pytest.mark.asyncio


async def test_apply_promo_success(
    authorized_client,
    db_session,
    auth_user,
    promo_factory,
):
    promo = await promo_factory(
        code="SALE10",
        percent=Decimal("10.00"),
        max_discount=Decimal("500.00"),
        used_count=0,
        max_uses=10,
        expires_at=datetime.now(timezone.utc) + timedelta(days=1),
    )

    response = await authorized_client.post(
        "/promo/apply",
        json={"code": "SALE10"},
    )

    assert response.status_code == 200
    assert Decimal(str(response.json()["discount"])) == Decimal("100.00")

    await db_session.refresh(promo)

    cart = await db_session.scalar(
        select(Cart).where(Cart.user_id == auth_user.id)
    )

    assert cart is not None
    assert cart.discount == Decimal("100.00")
    assert promo.used_count == 1


async def test_apply_promo_discount_is_limited_by_max_discount(
    authorized_client,
    db_session,
    auth_user,
    promo_factory,
):
    cart = await db_session.scalar(
        select(Cart).where(Cart.user_id == auth_user.id)
    )
    cart.total = Decimal("10000.00")
    await db_session.commit()

    promo = await promo_factory(
        code="BIGSALE",
        percent=Decimal("50.00"),
        max_discount=Decimal("1000.00"),
        used_count=0,
        max_uses=10,
    )

    response = await authorized_client.post(
        "/promo/apply",
        json={"code": "BIGSALE"},
    )

    assert response.status_code == 200
    assert Decimal(str(response.json()["discount"])) == Decimal("1000.00")

    await db_session.refresh(promo)
    await db_session.refresh(cart)

    assert cart.discount == Decimal("1000.00")
    assert promo.used_count == 1


async def test_apply_promo_code_is_case_insensitive(
    authorized_client,
    db_session,
    promo_factory,
):
    promo = await promo_factory(code="SALE10")

    response = await authorized_client.post(
        "/promo/apply",
        json={"code": "sale10"},
    )

    assert response.status_code == 200

    await db_session.refresh(promo)
    assert promo.used_count == 1


async def test_apply_promo_not_found(
    authorized_client,
):
    response = await authorized_client.post(
        "/promo/apply",
        json={"code": "UNKNOWN"},
    )

    assert response.status_code == 404
    assert response.json()["detail"] == "Промокод не найден"


async def test_apply_promo_expired(
    authorized_client,
    promo_factory,
):
    await promo_factory(
        code="OLD",
        expires_at=datetime.now(timezone.utc) - timedelta(seconds=1),
    )

    response = await authorized_client.post(
        "/promo/apply",
        json={"code": "OLD"},
    )

    assert response.status_code == 404
    assert response.json()["detail"] == "Промокод не найден"


async def test_apply_promo_no_uses_left(
    authorized_client,
    promo_factory,
):
    await promo_factory(
        code="LIMITED",
        used_count=5,
        max_uses=5,
        expires_at=datetime.now(timezone.utc) + timedelta(days=1),
    )

    response = await authorized_client.post(
        "/promo/apply",
        json={"code": "LIMITED"},
    )

    assert response.status_code == 409
    assert response.json()["detail"] == "Промокод закончился"


async def test_apply_promo_requires_authentication(
    anonymous_client,
    promo_factory,
):
    await promo_factory(code="SALE10")

    response = await anonymous_client.post(
        "/promo/apply",
        json={"code": "SALE10"},
    )

    assert response.status_code in {401, 403}


async def test_apply_promo_validation_error_when_code_is_missing(
    authorized_client,
):
    response = await authorized_client.post(
        "/promo/apply",
        json={},
    )

    assert response.status_code == 422


async def test_apply_promo_validation_error_when_code_has_wrong_type(
    authorized_client,
):
    response = await authorized_client.post(
        "/promo/apply",
        json={"code": 123},
    )

    # В Pydantic v2 обычный str по умолчанию не приводит int к str.
    # Если в вашей схеме включено приведение типов, этот тест нужно адаптировать.
    assert response.status_code == 422


@pytest.mark.xfail(
    reason=(
        "Бизнес-правило не реализовано в эндпоинте: "
        "повторное применение того же промокода к той же корзине "
        "не должно увеличивать used_count, но текущий код увеличивает."
    ),
    strict=True,
)
async def test_reapplying_same_promo_to_same_cart_does_not_increment_used_count(
    authorized_client,
    db_session,
    promo_factory,
):
    promo = await promo_factory(
        code="ONCE",
        percent=Decimal("10.00"),
        max_discount=Decimal("500.00"),
        used_count=0,
        max_uses=10,
    )

    first_response = await authorized_client.post(
        "/promo/apply",
        json={"code": "ONCE"},
    )
    second_response = await authorized_client.post(
        "/promo/apply",
        json={"code": "ONCE"},
    )

    assert first_response.status_code == 200
    assert second_response.status_code == 200

    await db_session.refresh(promo)

    assert promo.used_count == 1


@pytest.mark.xfail(
    reason=(
        "Бизнес-правило 'один промокод на корзину' в текущем коде не реализовано: "
        "эндпоинт не хранит применённый промокод и позволяет применять другой."
    ),
    strict=True,
)
async def test_only_one_promo_can_be_applied_to_cart(
    authorized_client,
    db_session,
    auth_user,
    promo_factory,
):
    promo1 = await promo_factory(
        code="PROMO1",
        percent=Decimal("10.00"),
        max_discount=Decimal("500.00"),
    )
    promo2 = await promo_factory(
        code="PROMO2",
        percent=Decimal("20.00"),
        max_discount=Decimal("500.00"),
    )

    first_response = await authorized_client.post(
        "/promo/apply",
        json={"code": "PROMO1"},
    )
    second_response = await authorized_client.post(
        "/promo/apply",
        json={"code": "PROMO2"},
    )

    assert first_response.status_code == 200

    # Предположение по бизнес-правилу: второй промокод должен быть отклонён.
    # Если ожидается замена промокода, нужно проверить корректный rollback used_count у PROMO1.
    assert second_response.status_code in {400, 409}

    await db_session.refresh(promo1)
    await db_session.refresh(promo2)

    assert promo1.used_count == 1
    assert promo2.used_count == 0


@pytest.mark.xfail(
    reason=(
        "Конкурентный сценарий небезопасен: чтение used_count и инкремент "
        "выполняются без SELECT FOR UPDATE/atomic UPDATE/constraint. "
        "При max_uses=1 оба запроса могут пройти."
    ),
    strict=False,
)
async def test_concurrent_apply_promo_with_single_available_use_allows_only_one_success(
    authorized_client,
    db_session,
    promo_factory,
):
    promo = await promo_factory(
        code="RACE",
        percent=Decimal("10.00"),
        max_discount=Decimal("500.00"),
        used_count=0,
        max_uses=1,
        expires_at=datetime.now(timezone.utc) + timedelta(days=1),
    )

    responses = await asyncio.gather(
        authorized_client.post("/promo/apply", json={"code": "RACE"}),
        authorized_client.post("/promo/apply", json={"code": "RACE"}),
    )

    statuses = sorted(response.status_code for response in responses)

    assert statuses == [200, 409]

    await db_session.refresh(promo)
    assert promo.used_count == 1


@pytest.mark.xfail(
    reason=(
        "expires_at хранится как timestamptz и обычно возвращается timezone-aware, "
        "а endpoint сравнивает его с datetime.utcnow(), то есть naive datetime. "
        "Это может приводить к TypeError и 500."
    ),
    strict=False,
)
async def test_apply_promo_with_timestamptz_expires_at_does_not_fail_with_500(
    authorized_client,
    promo_factory,
):
    await promo_factory(
        code="TZ",
        expires_at=datetime.now(timezone.utc) + timedelta(days=1),
    )

    response = await authorized_client.post(
        "/promo/apply",
        json={"code": "TZ"},
    )

    assert response.status_code == 200

---

3. Как запускать тесты в CI

Зависимости

Пример requirements-test.txt:

pytest
pytest-asyncio
httpx
testcontainers[postgresql]
asyncpg
SQLAlchemy
psycopg2-binary

psycopg2-binary нужен не приложению, а testcontainers, потому что он может отдавать sync connection URL и делать health-check через psycopg2.

---

pytest.ini

[pytest]
asyncio_mode = auto
testpaths = tests

---

GitHub Actions пример

name: tests

on:
  pull_request:
  push:
    branches:
      - main

jobs:
  integration-tests:
    runs-on: ubuntu-latest

    services:
      docker:
        image: docker:dind
        options: --privileged

    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt
          pip install -r requirements-test.txt

      - name: Run tests
        env:
          TESTCONTAINERS_RYUK_DISABLED: "true"
        run: |
          pytest -vv

На GitHub-hosted runners Docker уже доступен, отдельный service docker:dind обычно не нужен. Более простой вариант:

name: tests

on:
  pull_request:
  push:
    branches:
      - main

jobs:
  integration-tests:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt
          pip install -r requirements-test.txt

      - name: Run tests
        env:
          TESTCONTAINERS_RYUK_DISABLED: "true"
        run: pytest -vv

---

Изоляция данных между тестами

В примере используется такая стратегия:

1. Один PostgreSQL-контейнер на всю test session.
2. Один AsyncEngine на всю test session.
3. Таблицы создаются один раз через:

await conn.run_sync(Base.metadata.create_all)

1. После каждого теста выполняется:

TRUNCATE TABLE carts, promos, users RESTART IDENTITY CASCADE

Плюсы:

• быстро;
• данные не протекают между тестами;
• sequence/id сбрасываются;
• тесты используют настоящую PostgreSQL, а не SQLite.

Если у вас много таблиц, лучше не перечислять их вручную, а генерировать список из metadata:

table_names = ", ".join(
    f'"{table.name}"'
    for table in reversed(Base.metadata.sorted_tables)
)
await conn.execute(
    text(f"TRUNCATE TABLE {table_names} RESTART IDENTITY CASCADE")
)

Но такой вариант нужно аккуратно использовать, если есть схемы, quoted identifiers или служебные таблицы миграций.

---

Найденные проблемы в эндпоинте

1. Повторное применение промокода увеличивает used_count

Бизнес-правило:

> один промокод на корзину; повторное применение того же кода не должно увеличивать used_count.

В текущем коде нет проверки, был ли промокод уже применён к корзине:

cart.discount = min(cart.total * promo.percent / 100, promo.max_discount)
promo.used_count += 1
await db.commit()

Тест, который это ловит:

test_reapplying_same_promo_to_same_cart_does_not_increment_used_count

Он помечен xfail, потому что сейчас endpoint, скорее всего, его провалит.

---

2. Гонка при конкурентных запросах

Код делает:

if promo.used_count >= promo.max_uses:
    raise HTTPException(409, "Промокод закончился")

promo.used_count += 1
await db.commit()

Это небезопасно при параллельных запросах. Два запроса могут одновременно прочитать used_count=0, оба пройти проверку и оба вернуть 200.

Тест:

test_concurrent_apply_promo_with_single_available_use_allows_only_one_success

---

3. Возможная ошибка timezone-aware vs timezone-naive

Бизнес-правило:

> expires_at хранится как timestamptz.

PostgreSQL timestamptz обычно возвращается в Python как timezone-aware datetime.

В коде:

promo.expires_at < datetime.utcnow()

datetime.utcnow() возвращает timezone-naive datetime.

Сравнение aware и naive datetime может привести к:

TypeError: can't compare offset-naive and offset-aware datetimes

Тест:

test_apply_promo_with_timestamptz_expires_at_does_not_fail_with_500

Корректнее было бы сравнивать с:

datetime.now(timezone.utc)

Но код эндпоинта я не менял, только отметил баг и добавил тест.

Тесты для POST /promo/apply

Перед тем как перейти к коду, зафиксирую допущения о доменной модели, так как в задаче дан только сам хендлер. В реальном проекте нужно поправить импорты/имена полей под фактический код:

# app/models.py (предполагаемая структура)
class User(Base):
    id: Mapped[int]
    email: Mapped[str]

class Promo(Base):
    id: Mapped[int]
    code: Mapped[str]
    percent: Mapped[float]
    max_discount: Mapped[float]
    max_uses: Mapped[int]
    used_count: Mapped[int]
    expires_at: Mapped[datetime]  # timestamptz -> DateTime(timezone=True)

class Cart(Base):
    id: Mapped[int]
    user_id: Mapped[int]
    total: Mapped[float]
    discount: Mapped[float]

# app/schemas.py
class ApplyPromo(BaseModel):
    code: str  # min_length=1 предполагается

# app/deps.py
async def current_user(...) -> User: ...  # кидает 401, если не аутентифицирован

# app/db.py
async def get_db() -> AsyncSession: ...

---

1. Таблица тест-кейсов

---

2. Код тестов

tests/factories.py

from datetime import datetime, timedelta, timezone

from app.models import Promo, Cart


def make_promo(
    *,
    code: str = "SAVE10",
    percent: float = 10,
    max_discount: float = 100,
    max_uses: int = 5,
    used_count: int = 0,
    expires_in: timedelta = timedelta(days=1),
) -> Promo:
    return Promo(
        code=code.upper(),
        percent=percent,
        max_discount=max_discount,
        max_uses=max_uses,
        used_count=used_count,
        expires_at=datetime.now(timezone.utc) + expires_in,  # timestamptz
    )


def make_cart(*, user_id: int, total: float = 1000, discount: float = 0) -> Cart:
    return Cart(user_id=user_id, total=total, discount=discount)

tests/conftest.py

import asyncio
from types import SimpleNamespace

import pytest
import pytest_asyncio
from fastapi import Request
from httpx import ASGITransport, AsyncClient
from sqlalchemy import text
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from testcontainers.postgres import PostgresContainer

from app.main import app
from app.db import Base, get_db
from app.deps import current_user
from app.models import User

# ---
# PostgreSQL в Docker — один раз на весь прогон
# ---

@pytest.fixture(scope="session")
def postgres_container():
    with PostgresContainer("postgres:16-alpine") as pg:
        yield pg


@pytest_asyncio.fixture(scope="session")
async def async_engine(postgres_container):
    dsn = postgres_container.get_connection_url().replace(
        "postgresql+psycopg2", "postgresql+asyncpg"
    )
    engine = create_async_engine(dsn, pool_pre_ping=True)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()


# ---
# Изоляция данных: внешняя транзакция + SAVEPOINT на каждый тест.
# join_transaction_mode="create_savepoint" гарантирует, что commit()
# внутри тестируемого кода схлопывает только savepoint, а не внешнюю
# транзакцию — после теста всё откатывается.
# ---

@pytest_asyncio.fixture
async def db_session(async_engine):
    connection = await async_engine.connect()
    outer_trans = await connection.begin()

    session_factory = async_sessionmaker(
        bind=connection,
        expire_on_commit=False,
        join_transaction_mode="create_savepoint",
    )
    session = session_factory()

    yield session

    await session.close()
    if outer_trans.is_active:
        await outer_trans.rollback()
    await connection.close()


@pytest_asyncio.fixture
async def override_get_db(db_session):
    async def _get_db():
        yield db_session

    app.dependency_overrides[get_db] = _get_db
    yield
    app.dependency_overrides.pop(get_db, None)


@pytest_asyncio.fixture
async def test_user(db_session):
    user = User(email="test@example.com", hashed_password="x")
    db_session.add(user)
    await db_session.flush()
    return user


@pytest_asyncio.fixture
async def override_current_user(test_user):
    async def _current_user():
        return test_user

    app.dependency_overrides[current_user] = _current_user
    yield
    app.dependency_overrides.pop(current_user, None)


@pytest_asyncio.fixture
async def client(override_get_db, override_current_user):
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac


@pytest_asyncio.fixture
async def anon_client(override_get_db):
    """Клиент без переопределения current_user — должен ловить реальный 401."""
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac


@pytest_asyncio.fixture
async def client_no_raise(override_get_db, override_current_user):
    """Клиент, который не пробрасывает исключения наружу, а превращает
    их в HTTP 500 — нужен, чтобы ловить необработанные исключения в хендлере."""
    transport = ASGITransport(app=app, raise_app_exceptions=False)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac


# ---
# Отдельная инфраструктура для теста конкурентности: без savepoint-обёртки,
# реальные commit'ы в БД + ручная очистка после теста. Пользователь
# передаётся через заголовок, чтобы не было гонки за dependency_overrides.
# ---

@pytest_asyncio.fixture
async def real_session_factory(async_engine):
    return async_sessionmaker(bind=async_engine, expire_on_commit=False)


@pytest_asyncio.fixture
async def concurrency_client(real_session_factory):
    async def _get_db():
        async with real_session_factory() as session:
            yield session

    async def _current_user_by_header(request: Request):
        return SimpleNamespace(id=int(request.headers["X-Test-User-Id"]))

    app.dependency_overrides[get_db] = _get_db
    app.dependency_overrides[current_user] = _current_user_by_header

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac

    app.dependency_overrides.pop(get_db, None)
    app.dependency_overrides.pop(current_user, None)

tests/test_promo_apply.py

import asyncio
from datetime import timedelta

import pytest
from sqlalchemy import text

from app.models import User
from tests.factories import make_promo, make_cart

pytestmark = pytest.mark.asyncio


async def _seed(db_session, user_id, **promo_kwargs):
    promo = make_promo(**promo_kwargs)
    cart = make_cart(user_id=user_id)
    db_session.add_all([promo, cart])
    await db_session.commit()
    return promo, cart


# ---
# Успешные сценарии
# ---

async def test_apply_promo_success(client, db_session, test_user):
    await _seed(db_session, test_user.id, code="SAVE10", percent=10, max_discount=500)

    resp = await client.post("/promo/apply", json={"code": "SAVE10"})

    assert resp.status_code == 200
    assert resp.json()["discount"] == pytest.approx(100.0)


async def test_apply_promo_code_case_insensitive(client, db_session, test_user):
    await _seed(db_session, test_user.id, code="SAVE10")

    resp = await client.post("/promo/apply", json={"code": "save10"})

    assert resp.status_code == 200


async def test_discount_capped_by_max_discount(client, db_session, test_user):
    await _seed(db_session, test_user.id, code="BIG50", percent=50, max_discount=100)
    # total=1000 -> 50% = 500, но ограничение max_discount=100

    resp = await client.post("/promo/apply", json={"code": "BIG50"})

    assert resp.status_code == 200
    assert resp.json()["discount"] == pytest.approx(100.0)


async def test_used_count_increments_after_apply(client, db_session, test_user):
    promo, _ = await _seed(db_session, test_user.id, code="ONEUSE", used_count=0, max_uses=5)

    resp = await client.post("/promo/apply", json={"code": "ONEUSE"})
    assert resp.status_code == 200

    await db_session.refresh(promo)
    assert promo.used_count == 1


# ---
# Граничные случаи
# ---

async def test_promo_at_last_allowed_use_succeeds(client, db_session, test_user):
    await _seed(db_session, test_user.id, code="LASTONE", used_count=4, max_uses=5)

    resp = await client.post("/promo/apply", json={"code": "LASTONE"})

    assert resp.status_code == 200


async def test_promo_exhausted_returns_409(client, db_session, test_user):
    await _seed(db_session, test_user.id, code="GONE", used_count=5, max_uses=5)

    resp = await client.post("/promo/apply", json={"code": "GONE"})

    assert resp.status_code == 409


async def test_promo_expired_one_second_ago_returns_404(client, db_session, test_user):
    await _seed(db_session, test_user.id, code="OLD", expires_in=timedelta(seconds=-1))

    resp = await client.post("/promo/apply", json={"code": "OLD"})

    assert resp.status_code == 404


async def test_promo_valid_for_one_more_second_succeeds(client, db_session, test_user):
    await _seed(db_session, test_user.id, code="FRESH", expires_in=timedelta(seconds=30))

    resp = await client.post("/promo/apply", json={"code": "FRESH"})

    assert resp.status_code == 200


# ---
# Валидация и доступ
# ---

async def test_unknown_promo_returns_404(client, db_session, test_user):
    resp = await client.post("/promo/apply", json={"code": "NOPE"})
    assert resp.status_code == 404


async def test_empty_code_returns_422(client):
    resp = await client.post("/promo/apply", json={"code": ""})
    assert resp.status_code == 422


async def test_missing_code_field_returns_422(client):
    resp = await client.post("/promo/apply", json={})
    assert resp.status_code == 422


async def test_unauthenticated_user_gets_401(anon_client, db_session, test_user):
    await _seed(db_session, test_user.id, code="SAVE10")
    resp = await anon_client.post("/promo/apply", json={"code": "SAVE10"})
    assert resp.status_code == 401


# ---
# Тесты, фиксирующие найденные баги (см. раздел "Баги" ниже)
# ---

async def test_bug_reapply_same_code_does_not_increment_used_count_twice(
    client, db_session, test_user
):
    """Бизнес-правило: повторное применение ОДНОГО И ТОГО ЖЕ кода
    к той же корзине не должно повторно увеличивать used_count."""
    promo, _ = await _seed(db_session, test_user.id, code="TWICE", used_count=0, max_uses=5)

    resp1 = await client.post("/promo/apply", json={"code": "TWICE"})
    resp2 = await client.post("/promo/apply", json={"code": "TWICE"})

    assert resp1.status_code == 200
    assert resp2.status_code == 200

    await db_session.refresh(promo)
    assert promo.used_count == 1, (
        f"БАГ: used_count вырос до {promo.used_count} при повторном "
        f"применении того же промокода к той же корзине"
    )


async def test_bug_second_different_promo_is_rejected(client, db_session, test_user):
    """Бизнес-правило: один промокод на корзину."""
    await _seed(db_session, test_user.id, code="FIRST", percent=10, max_discount=1000)
    promo2 = make_promo(code="SECOND", percent=20, max_discount=1000)
    db_session.add(promo2)
    await db_session.commit()

    resp1 = await client.post("/promo/apply", json={"code": "FIRST"})
    resp2 = await client.post("/promo/apply", json={"code": "SECOND"})

    assert resp1.status_code == 200
    assert resp2.status_code == 409, (
        f"БАГ: применение второго промокода к корзине с уже применённым "
        f"первым вернуло {resp2.status_code} вместо 409 — скидка молча "
        f"перезаписывается"
    )


async def test_bug_valid_not_expired_promo_should_return_200_not_500(
    client_no_raise, db_session, test_user
):
    """expires_at хранится как timestamptz (offset-aware), а в коде
    сравнивается с datetime.utcnow() (offset-naive):

        promo.expires_at < datetime.utcnow()

    На реальной Postgres/asyncpg это приводит к TypeError и 500
    на абсолютно валидном, не истёкшем промокоде."""
    await _seed(db_session, test_user.id, code="TZBUG", expires_in=timedelta(days=1))

    resp = await client_no_raise.post("/promo/apply", json={"code": "TZBUG"})

    assert resp.status_code == 200, (
        "БАГ: сравнение offset-aware expires_at (timestamptz) с "
        "offset-naive datetime.utcnow() приводит к TypeError -> 500"
    )


async def test_bug_concurrent_requests_can_exceed_max_uses(
    concurrency_client, real_session_factory
):
    """Между проверкой used_count >= max_uses и записью used_count += 1
    нет блокировки строки (SELECT ... FOR UPDATE) и нет атомарного
    UPDATE ... WHERE used_count < max_uses. При параллельных запросах
    промокод может быть применён больше раз, чем max_uses."""
    async with real_session_factory() as session:
        user1 = User(email="race1@example.com", hashed_password="x")
        user2 = User(email="race2@example.com", hashed_password="x")
        session.add_all([user1, user2])
        await session.flush()

        promo = make_promo(code="RACE10", max_uses=1, used_count=0)
        cart1 = make_cart(user_id=user1.id)
        cart2 = make_cart(user_id=user2.id)
        session.add_all([promo, cart1, cart2])
        await session.commit()
        user1_id, user2_id = user1.id, user2.id

    async def call_as(user_id: int):
        return await concurrency_client.post(
            "/promo/apply",
            json={"code": "RACE10"},
            headers={"X-Test-User-Id": str(user_id)},
        )

    try:
        responses = await asyncio.gather(call_as(user1_id), call_as(user2_id))
        success_count = sum(1 for r in responses if r.status_code == 200)

        assert success_count == 1, (
            f"БАГ (race condition / lost update): промокод с max_uses=1 "
            f"был успешно применён {success_count} раз(а) при "
            f"параллельных запросах"
        )
    finally:
        async with real_session_factory() as session:
            await session.execute(
                text("DELETE FROM carts WHERE user_id IN (:a, :b)"),
                {"a": user1_id, "b": user2_id},
            )
            await session.execute(text("DELETE FROM promos WHERE code = 'RACE10'"))
            await session.execute(
                text("DELETE FROM users WHERE id IN (:a, :b)"),
                {"a": user1_id, "b": user2_id},
            )
            await session.commit()

---

Найденные баги (код эндпоинта не менялся, см. тесты выше)

1. Race condition / lost update на used_count.

Между SELECT promo и promo.used_count += 1; commit нет ни SELECT ... FOR UPDATE, ни атомарного UPDATE promos SET used_count = used_count + 1 WHERE id = ... AND used_count < max_uses RETURNING .... При параллельных запросах несколько транзакций читают одно и то же старое значение used_count и обе проходят проверку лимита → промокод может быть использован больше раз, чем max_uses.
Тест: test_bug_concurrent_requests_can_exceed_max_uses.

1. Сравнение offset-aware и offset-naive datetime.

expires_at — timestamptz, из БД приходит как aware datetime. datetime.utcnow() — naive. Выражение promo.expires_at < datetime.utcnow() на боевой Postgres/asyncpg бросает TypeError: can't compare offset-naive and offset-aware datetimes, из-за чего падает 500 даже на абсолютно валидном промокоде. Нужно datetime.now(timezone.utc).
Тест: test_bug_valid_not_expired_promo_should_return_200_not_500.

1. Нет идемпотентности повторного применения одного и того же кода.

Нет связи «какой промокод уже применён к корзине», поэтому повторный вызов с тем же кодом снова инкрементирует used_count, нарушая правило «повторное применение не увеличивает used_count».
Тест: test_bug_reapply_same_code_does_not_increment_used_count_twice.

1. Правило «один промокод на корзину» не реализовано.

cart.discount просто перезаписывается при каждом вызове — второй (другой) промокод молча вытесняет первый вместо явного отказа.
Тест: test_bug_second_different_promo_is_rejected.

---

3. CI и изоляция данных

Зависимости (pyproject.toml / requirements-test.txt)

pytest
pytest-asyncio
httpx
testcontainers[postgres]
asyncpg
sqlalchemy>=2.0.30

Конфиг pytest (pyproject.toml)

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
markers = [
    "integration: интеграционные тесты с реальной БД в Docker",
]

Изоляция данных между тестами

• Для подавляющего большинства тестов — паттерн «внешняя транзакция + SAVEPOINT» (db_session фикстура): каждый тест получает своё соединение, открытую транзакцию и session с join_transaction_mode="create_savepoint". Даже await db.commit() внутри хендлера схлопывает только savepoint — после теста внешняя транзакция откатывается, и БД возвращается в исходное состояние. Тесты можно гонять параллельно (pytest-xdist -n auto) без взаимного влияния, т.к. postgres_container/async_engine — session-scoped, а данные каждого теста живут в своей, откатываемой транзакции.
• Для теста конкурентности (test_bug_concurrent_requests_can_exceed_max_uses) savepoint-изоляция не подходит — нужны реальные конкурирующие коммиты в одну и ту же строку. Используется отдельная фикстура concurrency_client без внешней транзакции, с явной очисткой тестовых данных в finally/teardown (DELETE ... WHERE по известным ключам/кодам). Именно поэтому такие тесты стоит держать отдельно и помечать (@pytest.mark.integration), чтобы при необходимости их можно было исключать из параллельного прогона с -n (гонка потоков теста и так есть внутри самого теста, доп. parallelism от xdist не нужен).

Пример GitHub Actions

name: tests

on: [push, pull_request]

jobs:
  integration-tests:
    runs-on: ubuntu-latest   # Docker daemon доступен "из коробки"
    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Install deps
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt -r requirements-test.txt

      - name: Run tests
        env:
          TESTCONTAINERS_RYUK_DISABLED: "false"
        run: |
          pytest -v -m "" tests/ \
            -p no:cacheprovider \
            --maxfail=1

Примечания:

• Testcontainers сам поднимает и гасит контейнер Postgres через Docker daemon хоста раннера — отдельный services: в workflow не нужен.
• Ryuk (служебный контейнер testcontainers) сам подчищает контейнеры после падения джобы — не отключайте его в CI без необходимости.
• Если нужно распараллелить обычные (не конкурентные) тесты — добавить pytest-xdist и запускать pytest -n auto; тест на race condition стоит исключать отдельным маркером и гонять последовательно, чтобы не путать «параллелизм теста» с «параллелизмом внутри теста».