Стратег Тестирования

Стратегия тестирования OrderService

Код ниже написан для pytest 8.x, pytest-asyncio ≥ 0.24, pytest-mock, SQLAlchemy 2.x, redis-py ≥ 5.0.1 и fakeredis ≥ 2.20. В этой сессии я его не запускал. Первый прогон стоит сделать локально: pytest -m "not slow".

0. Анализ кода и риски

0.1. Публичный API, ветвления, зависимости

У модуля нет зависимости от времени (datetime.now() не вызывается), поэтому freezegun не нужен. Детерминизм обеспечивается патчем uuid.uuid4. Поскольку import uuid выполняется внутри метода, патчить нужно именно uuid.uuid4, а не order_service.uuid.

0.2. Риски и найденные дефекты (по приоритету)

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

2. Код тестов

Структура

pyproject.toml
requirements-test.txt
order_service.py
tests/
├── __init__.py
├── helpers.py
├── conftest.py
├── unit/
│   ├── __init__.py
│   ├── test_create_order.py
│   ├── test_apply_discount.py
│   ├── test_get_order_status.py
│   └── test_generate_order_id.py
├── integration/
│   ├── __init__.py
│   ├── conftest.py
│   └── test_order_flow.py
└── performance/
    ├── __init__.py
    └── test_create_order_perf.py

Файлы __init__.py пустые.

requirements-test.txt

pytest>=8.0,<9
pytest-asyncio>=0.24
pytest-mock>=3.12
pytest-cov>=5.0
SQLAlchemy>=2.0
redis>=5.0.1
fakeredis>=2.20
hypothesis>=6.100   # для property-based тестов из раздела 4

pyproject.toml

[tool.pytest.ini_options]
minversion = "8.0"
testpaths = ["tests"]
pythonpath = ["."]
asyncio_mode = "strict"                       # async-тесты только с явным @pytest.mark.asyncio
asyncio_default_fixture_loop_scope = "function"
xfail_strict = true                           # починили дефект -> XPASS -> красный тест, маркер надо снять
markers = [
    "slow: нагрузочные тесты; по умолчанию исключены (pytest -m slow --no-cov)",
    "integration: тесты с SQLite in-memory и fakeredis",
]
addopts = """
    -ra --strict-markers --strict-config
    -m "not slow"
    --cov=order_service --cov-branch
    --cov-report=term-missing
    --cov-report=xml:coverage.xml
    --cov-fail-under=90
"""

[tool.coverage.run]
branch = true
relative_files = true      # относительные пути в coverage.xml — нужно SonarQube в CI/Docker

[tool.coverage.report]
show_missing = true

tests/helpers.py

"""Общие тестовые утилиты: контракты зависимостей, фейки, фабрики данных.

Контракты (Protocol-подобные классы) описывают ДОПУЩЕНИЯ об интерфейсах внешних
зависимостей (см. раздел «Допущения»). create_autospec по ним ловит опечатки
в именах методов и неверные сигнатуры.
"""
from __future__ import annotations

import uuid
from dataclasses import dataclass
from decimal import Decimal
from typing import Any

FIXED_UUID = uuid.UUID("12345678-9abc-4def-8123-456789abcdef")


def expected_order_id(user_id: str) -> str:
    return f"{user_id}-{FIXED_UUID.hex[:8]}"


def make_item(sku: str = "SKU-1", qty: int = 1, price: Any = "10.00") -> dict:
    """Каждый вызов возвращает новый dict: общее изменяемое состояние между тестами исключено."""
    return {"sku": sku, "qty": qty, "price": price}


# ---------- Контракты зависимостей (допущения A1–A3) ----------

@dataclass(frozen=True)
class PaymentResult:
    success: bool
    reason: str | None = None


class PaymentGatewayContract:
    async def charge(self, user_id: str, amount: Decimal) -> PaymentResult: ...


class StockRepositoryContract:
    async def get_available(self, sku: str) -> int: ...
    async def reserve(self, items: list[dict]) -> None: ...
    async def release(self, items: list[dict]) -> None: ...


class CacheClientContract:
    async def get(self, key: str) -> str | None: ...
    async def set(self, key: str, value: str, ttl: int | None = None) -> None: ...


# ---------- Фейки для интеграционного уровня ----------

class RedisCacheAdapter:
    """Адаптер redis.asyncio -> контракт CacheClientContract (ttl -> ex)."""

    def __init__(self, redis_client) -> None:
        self._redis = redis_client

    async def get(self, key: str) -> str | None:
        return await self._redis.get(key)

    async def set(self, key: str, value: str, ttl: int | None = None) -> None:
        await self._redis.set(key, value, ex=ttl)


class InMemoryStockRepository:
    """Фейк склада с состоянием: позволяет проверить, что reserve/release взаимно компенсируются."""

    def __init__(self, levels: dict[str, int]) -> None:
        self.available: dict[str, int] = dict(levels)

    async def get_available(self, sku: str) -> int:
        return self.available.get(sku, 0)

    async def reserve(self, items: list[dict]) -> None:
        for item in items:
            self.available[item["sku"]] -= item["qty"]

    async def release(self, items: list[dict]) -> None:
        for item in items:
            self.available[item["sku"]] += item["qty"]

tests/conftest.py

"""Базовые фикстуры: все внешние зависимости — autospec-моки, без сети и БД."""
import pytest
from sqlalchemy.orm import Session

from order_service import OrderService
from tests.helpers import (
    FIXED_UUID,
    CacheClientContract,
    PaymentGatewayContract,
    PaymentResult,
    StockRepositoryContract,
)


@pytest.fixture
def fixed_uuid(mocker):
    # uuid импортируется внутри _generate_order_id, поэтому патчим атрибут модуля uuid
    return mocker.patch("uuid.uuid4", return_value=FIXED_UUID)


@pytest.fixture
def payment_gateway(mocker):
    gateway = mocker.create_autospec(PaymentGatewayContract, instance=True)
    gateway.charge.return_value = PaymentResult(success=True)  # AsyncMock: это awaited-значение
    return gateway


@pytest.fixture
def stock_repository(mocker):
    repo = mocker.create_autospec(StockRepositoryContract, instance=True)
    repo.get_available.return_value = 100
    return repo


@pytest.fixture
def cache_client(mocker):
    cache = mocker.create_autospec(CacheClientContract, instance=True)
    cache.get.return_value = None
    return cache


@pytest.fixture
def db_session(mocker):
    # Синхронная SQLAlchemy Session; цепочка query().filter_by().first() — обычные MagicMock
    return mocker.create_autospec(Session, instance=True)


@pytest.fixture
def service(db_session, payment_gateway, cache_client, stock_repository, fixed_uuid):
    return OrderService(
        db_session=db_session,
        payment_gateway=payment_gateway,
        cache_client=cache_client,
        stock_repository=stock_repository,
    )

tests/unit/test_create_order.py

"""Модульные тесты OrderService.create_order."""
import logging
import operator
from decimal import Decimal

import pytest

from order_service import InsufficientStockError, Order, PaymentDeclinedError
from tests.helpers import PaymentResult, expected_order_id, make_item

pytestmark = pytest.mark.asyncio

USER_ID = "user42"
DECLINE_REASON = "insufficient_funds"


@pytest.fixture
def declined_gateway(payment_gateway):
    payment_gateway.charge.return_value = PaymentResult(success=False, reason=DECLINE_REASON)
    return payment_gateway


def _recorder(journal: list[str], name: str, result=None):
    """side_effect, который пишет имя вызова в общий журнал (для проверки порядка)."""
    def _side_effect(*args, **kwargs):
        journal.append(name)
        return result
    return _side_effect


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

async def test_create_order_valid_single_item_returns_paid_order(service):
    # Arrange
    items = [make_item("SKU-1", qty=2, price="10.50")]
    # Act
    order = await service.create_order(USER_ID, items)
    # Assert
    assert order == Order(
        id=expected_order_id(USER_ID), user_id=USER_ID, total=Decimal("21.00"), status="paid"
    )


@pytest.mark.parametrize(
    ("items", "expected_total"),
    [
        pytest.param([make_item("A", 3, 0.1)], Decimal("0.3"), id="float-0.1x3-no-binary-drift"),
        pytest.param([make_item("A", 1, 19.99), make_item("B", 2, 5)], Decimal("29.99"), id="multi-items"),
        pytest.param([make_item("A", 1, "0.01")], Decimal("0.01"), id="min-price"),
    ],
)
async def test_create_order_mixed_price_types_charges_exact_decimal_total(
    service, payment_gateway, items, expected_total
):
    await service.create_order(USER_ID, items)
    payment_gateway.charge.assert_awaited_once_with(USER_ID, expected_total)


async def test_create_order_valid_items_charges_user_once(service, payment_gateway):
    await service.create_order(USER_ID, [make_item("SKU-1", qty=2, price="10.50")])
    payment_gateway.charge.assert_awaited_once_with(USER_ID, Decimal("21.00"))


@pytest.mark.parametrize(
    ("available", "qty"),
    [(5, 5), (6, 5)],
    ids=["available-equals-qty", "available-exceeds-qty"],
)
async def test_create_order_sufficient_stock_reserves_items(service, stock_repository, available, qty):
    stock_repository.get_available.return_value = available
    items = [make_item("SKU-1", qty=qty)]

    await service.create_order(USER_ID, items)

    stock_repository.reserve.assert_awaited_once_with(items)


async def test_create_order_payment_success_adds_order_to_session(service, db_session):
    order = await service.create_order(USER_ID, [make_item()])
    db_session.add.assert_called_once_with(order)


async def test_create_order_payment_success_commits_once(service, db_session):
    await service.create_order(USER_ID, [make_item()])
    db_session.commit.assert_called_once_with()


async def test_create_order_payment_success_caches_status_with_ttl(service, cache_client):
    await service.create_order(USER_ID, [make_item()])
    cache_client.set.assert_awaited_once_with(f"order:{expected_order_id(USER_ID)}", "paid", ttl=3600)


async def test_create_order_payment_success_logs_no_errors(service, caplog):
    caplog.set_level(logging.ERROR, logger="order_service")
    await service.create_order(USER_ID, [make_item()])
    assert [r for r in caplog.records if r.name == "order_service"] == []


# ---------------- InsufficientStockError ----------------

async def test_create_order_stock_short_by_one_raises_insufficient_stock(service, stock_repository):
    stock_repository.get_available.return_value = 4
    with pytest.raises(InsufficientStockError, match=r"SKU-1: доступно 4, запрошено 5"):
        await service.create_order(USER_ID, [make_item("SKU-1", qty=5)])


async def test_create_order_zero_stock_raises_insufficient_stock(service, stock_repository):
    stock_repository.get_available.return_value = 0
    with pytest.raises(InsufficientStockError):
        await service.create_order(USER_ID, [make_item(qty=1)])


@pytest.mark.parametrize(
    "dependency_call",
    ["stock_repository.reserve", "payment_gateway.charge", "db_session.commit", "cache_client.set"],
)
async def test_create_order_insufficient_stock_skips_side_effect(service, stock_repository, dependency_call):
    stock_repository.get_available.return_value = 0

    with pytest.raises(InsufficientStockError):
        await service.create_order(USER_ID, [make_item(qty=1)])

    assert operator.attrgetter(dependency_call)(service).call_count == 0


async def test_create_order_second_item_short_stops_stock_checks(service, stock_repository):
    stock_repository.get_available.side_effect = [10, 0, 10]
    items = [make_item("SKU-1"), make_item("SKU-2"), make_item("SKU-3")]

    with pytest.raises(InsufficientStockError, match="SKU-2"):
        await service.create_order(USER_ID, items)

    assert stock_repository.get_available.await_count == 2


# ---------------- PaymentDeclinedError и транзакционность ----------------

async def test_create_order_payment_declined_raises_with_reason(service, declined_gateway):
    with pytest.raises(PaymentDeclinedError, match=f"Платёж отклонён: {DECLINE_REASON}"):
        await service.create_order(USER_ID, [make_item()])


async def test_create_order_payment_declined_rolls_back_session(service, declined_gateway, db_session):
    with pytest.raises(PaymentDeclinedError):
        await service.create_order(USER_ID, [make_item()])
    db_session.rollback.assert_called_once_with()


async def test_create_order_payment_declined_releases_reserved_stock(service, declined_gateway, stock_repository):
    items = [make_item("SKU-1", qty=2)]
    with pytest.raises(PaymentDeclinedError):
        await service.create_order(USER_ID, items)
    stock_repository.release.assert_awaited_once_with(items)


@pytest.mark.parametrize("dependency_call", ["db_session.add", "db_session.commit", "cache_client.set"])
async def test_create_order_payment_declined_skips_persistence(service, declined_gateway, dependency_call):
    with pytest.raises(PaymentDeclinedError):
        await service.create_order(USER_ID, [make_item()])
    assert operator.attrgetter(dependency_call)(service).call_count == 0


async def test_create_order_payment_declined_compensates_in_order(
    service, payment_gateway, stock_repository, db_session
):
    # Arrange: единый журнал вызовов всех зависимостей
    journal: list[str] = []
    stock_repository.reserve.side_effect = _recorder(journal, "reserve")
    payment_gateway.charge.side_effect = _recorder(journal, "charge", PaymentResult(False, DECLINE_REASON))
    stock_repository.release.side_effect = _recorder(journal, "release")
    db_session.rollback.side_effect = _recorder(journal, "rollback")
    # Act
    with pytest.raises(PaymentDeclinedError):
        await service.create_order(USER_ID, [make_item()])
    # Assert
    assert journal == ["reserve", "charge", "release", "rollback"]


async def test_create_order_payment_declined_logs_error_with_order_id(service, declined_gateway, caplog):
    caplog.set_level(logging.ERROR, logger="order_service")

    with pytest.raises(PaymentDeclinedError):
        await service.create_order(USER_ID, [make_item()])

    records = [(r.levelno, r.getMessage()) for r in caplog.records if r.name == "order_service"]
    assert records == [
        (logging.ERROR, f"Оплата заказа {expected_order_id(USER_ID)} отклонена, откат резерва товара")
    ]


async def test_create_order_gateway_raises_declined_rolls_back_session(service, payment_gateway, db_session):
    payment_gateway.charge.side_effect = PaymentDeclinedError("fraud_suspected")
    with pytest.raises(PaymentDeclinedError, match="fraud_suspected"):
        await service.create_order(USER_ID, [make_item()])
    db_session.rollback.assert_called_once_with()


async def test_create_order_gateway_timeout_propagates_original_error(service, payment_gateway):
    payment_gateway.charge.side_effect = TimeoutError("PayFlow timeout")
    with pytest.raises(TimeoutError, match="PayFlow timeout"):
        await service.create_order(USER_ID, [make_item()])


@pytest.mark.xfail(reason="DEFECT-1: при сбое шлюза (не PaymentDeclinedError) резерв не освобождается")
async def test_create_order_gateway_timeout_releases_reserved_stock(service, payment_gateway, stock_repository):
    payment_gateway.charge.side_effect = TimeoutError("PayFlow timeout")
    with pytest.raises(TimeoutError):
        await service.create_order(USER_ID, [make_item()])
    stock_repository.release.assert_awaited_once()


@pytest.mark.xfail(reason="DEFECT-2: исключение в release() пропускает rollback() и маскирует отказ оплаты")
async def test_create_order_release_fails_still_rolls_back_session(
    service, declined_gateway, stock_repository, db_session
):
    stock_repository.release.side_effect = ConnectionError("stock service down")
    with pytest.raises(ConnectionError):
        await service.create_order(USER_ID, [make_item()])
    db_session.rollback.assert_called_once_with()


# ---------------- Характеризационные тесты (фиксируют ТЕКУЩЕЕ поведение, см. «Допущения») ----------------

async def test_create_order_cache_unavailable_commits_then_raises(service, cache_client, db_session):
    # RISK-5: заказ зафиксирован в БД, но клиент получит ошибку -> риск повторной оплаты
    cache_client.set.side_effect = ConnectionError("redis down")
    with pytest.raises(ConnectionError):
        await service.create_order(USER_ID, [make_item()])
    db_session.commit.assert_called_once_with()


async def test_create_order_empty_items_charges_zero_total(service, payment_gateway):
    # RISK-6: пустой заказ сейчас не валидируется; ожидаемое поведение нужно уточнить
    await service.create_order(USER_ID, [])
    payment_gateway.charge.assert_awaited_once_with(USER_ID, Decimal("0"))

tests/unit/test_apply_discount.py

"""Модульные тесты OrderService.apply_discount (синхронная чистая функция)."""
from decimal import Decimal

import pytest

from order_service import InvalidDiscountCodeError, OrderService

TOTAL = Decimal("100.00")


@pytest.fixture
def svc():
    # apply_discount не использует зависимости — передаём None, чтобы это было явно видно
    return OrderService(db_session=None, payment_gateway=None, cache_client=None, stock_repository=None)


@pytest.mark.parametrize(
    ("percent", "expected"),
    [
        pytest.param(0, Decimal("100.00"), id="0%-lower-bound"),
        pytest.param(0.01, Decimal("99.99"), id="0.01%-just-above-lower"),
        pytest.param(1, Decimal("99.00"), id="1%"),
        pytest.param(Decimal("12.5"), Decimal("87.50"), id="12.5%-decimal-input"),
        pytest.param(33.3, Decimal("66.70"), id="33.3%-fractional"),
        pytest.param(50, Decimal("50.00"), id="50%"),
        pytest.param(99.99, Decimal("0.01"), id="99.99%-just-below-upper"),
        pytest.param(100, Decimal("0.00"), id="100%-upper-bound"),
    ],
)
def test_apply_discount_valid_percent_returns_discounted_total(svc, percent, expected):
    assert svc.apply_discount(TOTAL, percent) == expected


@pytest.mark.parametrize(
    "percent",
    [
        pytest.param(-0.01, id="-0.01%-just-below-lower"),
        pytest.param(-1, id="-1%"),
        pytest.param(-100, id="-100%"),
        pytest.param(100.01, id="100.01%-just-above-upper"),
        pytest.param(101, id="101%"),
        pytest.param(1000, id="1000%"),
        pytest.param(float("inf"), id="+inf"),
        pytest.param(float("-inf"), id="-inf"),
    ],
)
def test_apply_discount_percent_out_of_range_raises_invalid_discount(svc, percent):
    with pytest.raises(InvalidDiscountCodeError, match="Некорректный процент скидки"):
        svc.apply_discount(TOTAL, percent)


@pytest.mark.xfail(reason="DEFECT-3: NaN проходит проверку диапазона и возвращается Decimal('NaN')")
def test_apply_discount_nan_percent_raises_invalid_discount(svc):
    with pytest.raises(InvalidDiscountCodeError):
        svc.apply_discount(TOTAL, float("nan"))


@pytest.mark.parametrize(
    "code",
    [None, "", "SALE10", "sale", "2024", "СКИДКА10"],
    ids=["none", "empty-string", "alnum-upper", "alpha-lower", "digits-only", "cyrillic"],
)
def test_apply_discount_valid_code_format_applies_discount(svc, code):
    assert svc.apply_discount(TOTAL, 10, code) == Decimal("90.00")


@pytest.mark.parametrize(
    "code",
    ["SALE-10", "SALE 10", " ", "10%", "code!", "a_b", "SALE10\n"],
    ids=["hyphen", "space-inside", "whitespace-only", "percent-sign", "exclamation", "underscore", "trailing-newline"],
)
def test_apply_discount_non_alnum_code_raises_invalid_discount(svc, code):
    with pytest.raises(InvalidDiscountCodeError, match="Некорректный формат кода скидки"):
        svc.apply_discount(TOTAL, 10, code)


def test_apply_discount_invalid_percent_and_code_reports_percent_error(svc):
    with pytest.raises(InvalidDiscountCodeError, match="процент"):
        svc.apply_discount(TOTAL, 150, "BAD-CODE")


def test_apply_discount_fractional_result_returns_unrounded_value(svc):
    # Характеризация RISK-9: результат не квантуется до копеек
    assert svc.apply_discount(Decimal("19.99"), 10) == Decimal("17.991")


def test_apply_discount_zero_total_returns_zero(svc):
    assert svc.apply_discount(Decimal("0"), 50) == Decimal("0")


def test_apply_discount_valid_input_returns_decimal_type(svc):
    assert isinstance(svc.apply_discount(TOTAL, 10), Decimal)

tests/unit/test_get_order_status.py

"""Модульные тесты OrderService.get_order_status."""
from decimal import Decimal

import pytest

from order_service import Order, OrderNotFoundError

pytestmark = pytest.mark.asyncio

ORDER_ID = "o-1"


def _db_returns(db_session, order):
    db_session.query.return_value.filter_by.return_value.first.return_value = order


async def test_get_order_status_cache_hit_returns_cached_status(service, cache_client):
    cache_client.get.return_value = "paid"
    assert await service.get_order_status(ORDER_ID) == "paid"


async def test_get_order_status_cache_hit_skips_database(service, cache_client, db_session):
    cache_client.get.return_value = "paid"
    await service.get_order_status(ORDER_ID)
    db_session.query.assert_not_called()


async def test_get_order_status_any_id_uses_prefixed_cache_key(service, cache_client):
    cache_client.get.return_value = "paid"
    await service.get_order_status(ORDER_ID)
    cache_client.get.assert_awaited_once_with(f"order:{ORDER_ID}")


async def test_get_order_status_cache_miss_returns_db_status(service, db_session):
    _db_returns(db_session, Order(id=ORDER_ID, user_id="u1", total=Decimal("1"), status="shipped"))
    assert await service.get_order_status(ORDER_ID) == "shipped"


async def test_get_order_status_cache_miss_filters_by_order_id(service, db_session):
    _db_returns(db_session, Order(id=ORDER_ID, user_id="u1", total=Decimal("1"), status="paid"))
    await service.get_order_status(ORDER_ID)
    db_session.query.return_value.filter_by.assert_called_once_with(id=ORDER_ID)


async def test_get_order_status_unknown_order_raises_order_not_found(service, db_session):
    _db_returns(db_session, None)
    with pytest.raises(OrderNotFoundError, match="Заказ o-404 не найден"):
        await service.get_order_status("o-404")


async def test_get_order_status_empty_cached_value_falls_back_to_db(service, cache_client, db_session):
    # Характеризация: "" считается промахом кэша (if cached)
    cache_client.get.return_value = ""
    _db_returns(db_session, Order(id=ORDER_ID, user_id="u1", total=Decimal("1"), status="paid"))
    assert await service.get_order_status(ORDER_ID) == "paid"


async def test_get_order_status_cache_unavailable_propagates_error(service, cache_client):
    # Характеризация: деградации до БД при недоступном Redis нет
    cache_client.get.side_effect = ConnectionError("redis down")
    with pytest.raises(ConnectionError):
        await service.get_order_status(ORDER_ID)

tests/unit/test_generate_order_id.py

import re

from order_service import OrderService


def test_generate_order_id_fixed_uuid_returns_user_prefixed_id(service):
    # service зависит от fixed_uuid -> uuid4 детерминирован
    assert service._generate_order_id("u1") == "u1-12345678"


def test_generate_order_id_real_uuid_matches_expected_format():
    # Значение случайное, но проверяется только формат, поэтому результат теста детерминирован
    svc = OrderService(db_session=None, payment_gateway=None, cache_client=None, stock_repository=None)
    assert re.fullmatch(r"u1-[0-9a-f]{8}", svc._generate_order_id("u1"))

tests/integration/conftest.py

"""Интеграционный уровень: реальная SQLAlchemy Session (SQLite in-memory) + fakeredis.
PaymentGateway остаётся autospec-моком (из корневого conftest), склад — фейк с состоянием."""
import fakeredis
import pytest
import pytest_asyncio
from fakeredis import aioredis as fake_aioredis
from sqlalchemy import Column, Numeric, String, Table, create_engine
from sqlalchemy.orm import Session, clear_mappers, registry

from order_service import Order
from tests.helpers import InMemoryStockRepository, RedisCacheAdapter


@pytest.fixture(scope="module")
def orders_metadata():
    # Order — обычный dataclass (допущение A5): мапим императивно только на время модуля
    reg = registry()
    orders = Table(
        "orders",
        reg.metadata,
        Column("id", String(64), primary_key=True),
        Column("user_id", String(64), nullable=False),
        Column("total", Numeric(12, 2), nullable=False),
        Column("status", String(16), nullable=False),
    )
    reg.map_imperatively(Order, orders)
    yield reg.metadata
    clear_mappers()  # возвращаем Order в исходное состояние для остальных тестов


@pytest.fixture
def db_session(orders_metadata):  # переопределяет мок из корневого conftest
    engine = create_engine("sqlite+pysqlite:///:memory:")
    orders_metadata.create_all(engine)
    with Session(engine) as session:
        yield session
    engine.dispose()


@pytest_asyncio.fixture
async def fake_redis():
    # Отдельный FakeServer на каждый тест: данные между тестами не пересекаются
    client = fake_aioredis.FakeRedis(server=fakeredis.FakeServer(), decode_responses=True)
    yield client
    await client.aclose()


@pytest.fixture
def cache_client(fake_redis):  # переопределяет мок из корневого conftest
    return RedisCacheAdapter(fake_redis)


@pytest.fixture
def stock_repository():  # переопределяет мок из корневого conftest
    return InMemoryStockRepository({"SKU-1": 10})

tests/integration/test_order_flow.py

"""Интеграция OrderService с SQLAlchemy (SQLite) и Redis (fakeredis)."""
import pytest
from sqlalchemy import text

from order_service import InsufficientStockError, OrderNotFoundError, PaymentDeclinedError
from tests.helpers import PaymentResult, make_item

pytestmark = [
    pytest.mark.asyncio,
    pytest.mark.integration,
    pytest.mark.filterwarnings("ignore:.*support Decimal objects natively:sqlalchemy.exc.SAWarning"),
]


async def test_create_order_success_persists_paid_row(service, db_session):
    order = await service.create_order("user1", [make_item("SKU-1", 2, "10.00")])
    status = db_session.execute(
        text("SELECT status FROM orders WHERE id = :id"), {"id": order.id}
    ).scalar_one()
    assert status == "paid"


async def test_create_order_success_stores_status_in_redis(service, fake_redis):
    order = await service.create_order("user1", [make_item()])
    assert await fake_redis.get(f"order:{order.id}") == "paid"


async def test_create_order_success_sets_redis_ttl_one_hour(service, fake_redis):
    order = await service.create_order("user1", [make_item()])
    assert 3590 < await fake_redis.ttl(f"order:{order.id}") <= 3600


async def test_get_order_status_cache_evicted_reads_status_from_db(service, fake_redis):
    order = await service.create_order("user1", [make_item()])
    await fake_redis.flushall()
    assert await service.get_order_status(order.id) == "paid"


async def test_get_order_status_cache_and_db_differ_returns_cached(service, fake_redis):
    order = await service.create_order("user1", [make_item()])
    await fake_redis.set(f"order:{order.id}", "shipped")
    assert await service.get_order_status(order.id) == "shipped"


async def test_create_order_payment_declined_leaves_no_order_rows(service, payment_gateway, db_session):
    payment_gateway.charge.return_value = PaymentResult(success=False, reason="card_expired")
    with pytest.raises(PaymentDeclinedError):
        await service.create_order("user1", [make_item("SKU-1", 3)])
    assert db_session.execute(text("SELECT COUNT(*) FROM orders")).scalar_one() == 0


async def test_create_order_payment_declined_restores_stock_level(service, payment_gateway, stock_repository):
    payment_gateway.charge.return_value = PaymentResult(success=False, reason="card_expired")
    with pytest.raises(PaymentDeclinedError):
        await service.create_order("user1", [make_item("SKU-1", 3)])
    assert stock_repository.available["SKU-1"] == 10


async def test_create_order_payment_declined_writes_nothing_to_redis(service, payment_gateway, fake_redis):
    payment_gateway.charge.return_value = PaymentResult(success=False, reason="card_expired")
    with pytest.raises(PaymentDeclinedError):
        await service.create_order("user1", [make_item()])
    assert await fake_redis.dbsize() == 0


async def test_create_order_insufficient_stock_keeps_stock_level(service, stock_repository):
    with pytest.raises(InsufficientStockError):
        await service.create_order("user1", [make_item("SKU-1", 11)])
    assert stock_repository.available["SKU-1"] == 10


async def test_get_order_status_unknown_order_raises_order_not_found(service):
    with pytest.raises(OrderNotFoundError):
        await service.get_order_status("missing")


async def test_raw_redis_client_set_with_ttl_kwarg_raises_type_error(fake_redis):
    # Контрактный тест (RISK-4): «голый» redis.asyncio не принимает ttl=, нужен адаптер или ex=
    with pytest.raises(TypeError):
        await fake_redis.set("order:x", "paid", ttl=3600)

tests/performance/test_create_order_perf.py

"""Нагрузочные тесты. Не входят в основной прогон; запуск: pytest -m slow --no-cov
(трассировка coverage замедляет код в разы и искажает замеры)."""
import time
import uuid

import pytest

from tests.helpers import make_item

pytestmark = [pytest.mark.slow, pytest.mark.asyncio]

ORDERS_COUNT = 500
TOTAL_LIMIT_S = 3.0
SINGLE_CALL_LIMIT_S = 0.050


@pytest.fixture
def unique_uuids(mocker, fixed_uuid):
    # Зависимость от fixed_uuid гарантирует, что этот патч ляжет поверх него.
    # UUID(int=i << 96) даёт hex[:8] == f"{i:08x}": ID уникальны и детерминированы.
    return mocker.patch("uuid.uuid4", side_effect=[uuid.UUID(int=i << 96) for i in range(ORDERS_COUNT)])


async def test_create_order_500_sequential_orders_complete_within_3_seconds(service, unique_uuids):
    items = [make_item("SKU-1", 1, "9.99")]

    start = time.perf_counter()
    for i in range(ORDERS_COUNT):
        await service.create_order(f"user{i}", items)
    elapsed = time.perf_counter() - start

    assert elapsed < TOTAL_LIMIT_S, f"{ORDERS_COUNT} заказов заняли {elapsed:.3f} с"


async def test_create_order_single_call_completes_within_50ms(service, unique_uuids):
    items = [make_item("SKU-1", 1, "9.99")]
    durations = []
    for i in range(ORDERS_COUNT):
        t0 = time.perf_counter()
        await service.create_order(f"user{i}", items)
        durations.append(time.perf_counter() - t0)

    assert max(durations) < SINGLE_CALL_LIMIT_S, f"max={max(durations) * 1000:.2f} мс"

3. Оценка покрытия

Цифры ниже оценены по анализу кода; их нужно подтвердить первым прогоном.

Точки ветвления order_service.py (coverage.py, --cov-branch):

Ожидаемый результат:

• строки: около 100% (все исполняемые операторы достижимы и покрыты уже unit-уровнем);
• ветвления: около 100% (14 из 14 дуг).

Это с запасом выше порогов 90% / 85%.

Что coverage.py не измеряет и как это закрыто:

• Покрытие условий (отдельные операнды в or/and). Каждый операнд покрыт явными параметрами: отрицательные значения и >100; None, "", валидный и невалидный код.
• Дефектные пути (DEFECT-1…3). Строки исполняются, но корректность поведения не подтверждена: такие тесты помечены xfail(strict=True).
• xfail и XPASS. Код в xfail-тестах всё равно исполняется и учитывается в покрытии. После исправления дефекта XPASS станет красным благодаря xfail_strict = true, и маркер придётся снять.
• Нагрузочные тесты в покрытие не входят (--no-cov) и на него не влияют.

Раздельный порог для ветвлений. --cov-fail-under проверяет только общий процент. Порог 85% по ветвлениям проверяйте отдельным шагом CI:

python - <<'EOF'
import sys, xml.etree.ElementTree as ET
root = ET.parse("coverage.xml").getroot()
line, branch = float(root.get("line-rate")), float(root.get("branch-rate"))
print(f"lines={line:.1%} branches={branch:.1%}")
sys.exit(0 if line >= 0.90 and branch >= 0.85 else 1)
EOF

4. Рекомендации

4.1. Покрытие и качество тестов

• SonarQube: sonar.python.coverage.reportPaths=coverage.xml. Благодаря relative_files = true пути корректны и внутри Docker-раннеров.
• Мутационное тестирование (mutmut run --paths-to-mutate order_service.py) показывает, действительно ли тесты ловят подмену < на <= или 100 на 99. Это важнее, чем 100% покрытия строк.
• pytest-randomly перемешивает порядок тестов и подтверждает их независимость.
• pytest-xdist -n auto ускоряет прогон: фикстуры изолированы, общего состояния нет.

4.2. Property-based тесты (Hypothesis) для apply_discount

from decimal import Decimal
from hypothesis import given, strategies as st
from order_service import OrderService

svc = OrderService(None, None, None, None)
totals = st.decimals(min_value=0, max_value=10**6, places=2, allow_nan=False, allow_infinity=False)
percents = st.floats(min_value=0, max_value=100, allow_nan=False)

@given(total=totals, percent=percents)
def test_apply_discount_any_valid_input_stays_within_zero_and_total(total, percent):
    assert Decimal(0) <= svc.apply_discount(total, percent) <= total

@given(total=totals, p1=percents, p2=percents)
def test_apply_discount_higher_percent_never_increases_price(total, p1, p2):
    lo, hi = sorted((p1, p2))
    assert svc.apply_discount(total, hi) <= svc.apply_discount(total, lo)

Для create_order подойдёт hypothesis.stateful.RuleBasedStateMachine поверх InMemoryStockRepository. Инвариант: сумма остатков плюс проданные товары постоянна при любой последовательности успешных и отклонённых оплат.

4.3. CI (пример GitHub Actions)

jobs:
  tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.11" }
      - run: pip install -r requirements-test.txt
      - run: pytest                       # addopts: -m "not slow", coverage, fail-under=90
      - run: python scripts/check_branch_coverage.py   # скрипт из раздела 3
      - uses: SonarSource/sonarqube-scan-action@v3
        env: { SONAR_TOKEN: "${{ secrets.SONAR_TOKEN }}" }

  perf:
    if: github.event_name == 'schedule'   # nightly, вне основного прогона
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.11" }
      - run: pip install -r requirements-test.txt
      - run: pytest -m slow --no-cov

Для трендов производительности подойдёт pytest-benchmark с --benchmark-compare-fail=mean:20%. Он надёжнее единичного порога на шумных раннерах.

4.4. Рефакторинг кода для тестируемости и надёжности

1. Ловить Exception в блоке оплаты: except Exception: + release + rollback + raise. Это исправляет DEFECT-1. Компенсацию обернуть в try/finally, чтобы rollback выполнялся всегда (DEFECT-2).
2. Добавить math.isfinite(discount_percent) в валидацию (DEFECT-3). Квантовать деньги: .quantize(Decimal("0.01"), ROUND_HALF_UP).
3. Внедрять id_factory через конструктор вместо uuid внутри метода. Увеличить длину суффикса ID (сейчас 32 бита).
4. Логировать через logger.exception(...) или с reason, чтобы в логе были причина и traceback.
5. Сделать запись в кэш после commit best-effort (try/except + warning) и добавить идемпотентность оплаты (idempotency key в PayFlow).
6. Перейти на AsyncSession: синхронные вызовы БД блокируют event loop.

5. Допущения и тесты после уточнения

Допущения

Тесты, которые стоит написать после уточнения требований

1. items=[]: ошибка валидации или допустимый нулевой заказ? Сейчас U-CO-24 фиксирует списание 0.
2. qty <= 0 и отрицательная цена: сейчас возможен отрицательный total и списание отрицательной суммы.
3. Две строки с одним SKU (qty 3 + 3 при остатке 5): нужно ли агрегировать позиции перед проверкой остатка?
4. Отсутствующие ключи sku/qty/price: сейчас будет KeyError. Нужно ли доменное исключение?
5. Правила округления денег (банковское или half-up) и минимальная итоговая сумма после скидки.
6. Нужно ли прогревать кэш при промахе в get_order_status и какова политика инвалидации при смене статуса?
7. Поведение при недоступном Redis: деградация до БД в get_order_status и best-effort запись в create_order.
8. Нужно ли переводить заказ в статус failed и сохранять его после отказа оплаты для аудита, вместо полного отката?
9. Атомарность reserve (TOCTOU): контрактный тест на конкурентные create_order через asyncio.gather с общим фейком склада.
10. Регистрозависимость кодов скидки и допустимость Unicode-букв и цифр ("СКИДКА", полноширинные цифры "１０" проходят isalnum()).