Моки и внешние системы в тестах на Python

Java Backend

Когда хватает рукописной заглушки, когда нужен AsyncMock с autospec, а когда respx или pytest-httpserver вместо партнёра: что нельзя подменять, как считать повторы и убрать зависимость от времени.

Опубликовано 3 октября 2026 ~12 мин чтения

← назад к разделу

Самое неприятное в заглушках — не то, что они ломают тесты, а то, что они их не ломают. Партнёр поменял формат ответа, сервис в проде считает отказ успехом, а тест по-прежнему зелёный: он спрашивает не партнёра, а наше представление о партнёре. В Python к этому добавляется собственная ловушка: mock.patch умеет подменить что угодно, где угодно, и именно поэтому им так легко подменить лишнее.

Обязательно

Когда заглушка уместна

В Python зависимость передают через конструктор, а её форму описывают маленьким Protocol — только те методы, которые нужны потребителю. Объявляет его тот, кто потребляет.

Заглушкой закрывают то, что за границей системы и чем вы не управляете; свой код подменять незачем, он и есть предмет теста.

Короткая формула: заглушка уместна на внешней границе — там, где заканчивается ваш код и начинается чужой.

from typing import Protocol


class Charger(Protocol):
    async def charge(self, request: ChargeRequest) -> ChargeResult: ...


class Checkout:
    def __init__(self, payments: Charger, orders: OrderStore) -> None:
        self.payments = payments
        self.orders = orders

Заглушку пишут руками, и это несколько строк:

class StubCharger:
    def __init__(self, result: ChargeResult | None = None, error: Exception | None = None) -> None:
        self.result = result
        self.error = error

    async def charge(self, request: ChargeRequest) -> ChargeResult:
        if self.error:
            raise self.error
        return self.result


async def test_checkout_payment_declined():
    checkout = Checkout(payments=StubCharger(ChargeResult(status=Status.DECLINED)), orders=FakeOrders())

    with pytest.raises(PaymentDeclined):
        await checkout.place(order_request())

Никакого Docker, никакой сети, тест работает мгновенно. Когда заглушек десятки, берут unittest.mock: create_autospec(Charger, instance=True) даёт объект с теми же методами, у асинхронных — AsyncMock, который умеет await. autospec здесь не формальность: обычный MagicMock молча примет вызов chrage с опечаткой и вернёт ещё один мок, а autospec упадёт с AttributeError. Правило простое: пока заглушка умещается в десять строк, пишите её руками; мок с автоспецификацией — когда нужны готовые счётчики вызовов.

Когда заглушка вредит

Короткая формула: подменять свой код — значит тестировать ожидания о поведении, а не само поведение.

Если заглушка стоит между двумя вашими классами (сценарий и репозиторий заказов), тест проходит даже при неправильном взаимодействии: поменяйте смысл метода, и заглушка «не заметит». Для своего репозитория лучше настоящая база в контейнере, а для чистой логики — fake: репозиторий на словаре, который действительно хранит и ищет.

class FakeOrders:
    def __init__(self) -> None:
        self.items: dict[OrderId, Order] = {}

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

    async def by_id(self, order_id: OrderId) -> Order:
        try:
            return self.items[order_id]
        except KeyError:
            raise OrderNotFound(order_id) from None

Fake ведёт себя как настоящий на уровне контракта — «сохранил, потом нашёл», «не нашёл — OrderNotFound», — поэтому тесты с ним переживают рефакторинг сценария. Блокировка здесь не нужна: цикл событий один, а словарь меняется между await. Если сценарий уходит в пул потоков через run_in_executor или anyio.to_thread, fake получает threading.Lock.

Чего нельзя подменять вообще

Чужие классы со сложным поведением. AsyncSession, AIOKafkaProducer, httpx.AsyncClient, клиент boto3. Заглушка на них воспроизводит не библиотеку, а вашу фантазию о ней: вы записали «на третий вызов вернёт пустой список», а настоящая библиотека в этом месте бросает исключение, повторяет запрос сама или отдаёт ленивый курсор. Признак ловушки: в тесте приходится описывать, в каком порядке и что вернут четыре метода чужого класса.

Чем заменять: настоящей реализацией в контейнере (база, брокер), заглушкой на уровне протокола (respx или сервер из следующего раздела, а не мок клиента; для AWS — moto, который отвечает на HTTP-вызовы boto3 как сам AWS) или своей тонкой обёрткой. Последнее — самый недооценённый приём: заводите свой Protocol с двумя-тремя методами на языке задачи, за ним прячете чужую библиотеку и подменяете свой интерфейс. Обёртка проверяется одним интеграционным тестом против настоящей библиотеки.

Значения. Money, OrderId, datetime, списки и словари создают настоящими — они для этого и существуют.

Функции модуля. datetime.now(), random.random(), os.environ внутри метода — зависимость, не выраженная в коде. mock.patch("app.orders.datetime") технически работает, и именно поэтому стоит сказать: это признак, что зависимость надо сделать явной — аргументом конструктора. Правило patch «подменяй там, где используется, а не там, где объявлено» существует ровно потому, что патч прибит к имени в конкретном модуле: тест ломается от переноса импорта. А сам datetime.datetime.now подменить и вовсе нельзя: это атрибут неизменяемого типа на C, и patch падает с cannot set 'now' attribute of immutable type.

То, что проверяется только целиком. Транзакция, права доступа, сериализация ответа, порядок middleware: заглушка обходит ровно тот механизм, который вы собирались проверить. Это уровень интеграционного теста.

Проверка вызовов

Всё, что выше, — про ответы заглушки. Но половина проверок звучит иначе: «письмо отправлено», «повтора не было», «в шлюз ушла правильная сумма». Это не результат функции, а факт исходящего вызова. Рукописная заглушка просто записывает, что с ней делали; у AsyncMock это встроено:

spy = create_autospec(Charger, instance=True)
spy.charge.return_value = ChargeResult(status=Status.APPROVED)
checkout = Checkout(payments=spy, orders=FakeOrders())

await checkout.place(order_request(total=Money(500)))

spy.charge.assert_awaited_once()
assert spy.charge.await_args.args[0].amount == Money(500)

spy.charge.assert_not_awaited()          # не вызывали вовсе
assert spy.charge.await_count == 3       # ровно три попытки

Проверка «не вызывали» заслуживает отдельного слова: именно такие ошибки самые дорогие — второе списание, повторное письмо, уведомление при откате. Число вызовов — единственный способ проверить повторы. У асинхронных методов смотрят await_count и assert_awaited_*, а не call_count: вызов корутины без await тоже засчитается как вызов, хотя ничего не произошло.

И граница, за которую лучше не выходить: проверка вызовов привязывает тест к реализации. «Вызвал ли репозиторий save» — плохая проверка: поведение «заказ сохранён» не изменится, если сохранять станут иначе. «Ушёл ли запрос в шлюз» — хорошая: сам вызов и есть наблюдаемое поведение.

respx и pytest-httpserver вместо реального партнёра

Когда ваш код вызывает внешний HTTP-сервис, поднять настоящий партнёрский сервер в тестах невозможно. В Python два инструмента, и они отвечают на разные вопросы.

respx перехватывает запросы httpx на уровне транспорта: сети нет, ответ подставляется в коде теста.

import httpx
import respx


@respx.mock(base_url="https://pay.example")
async def test_payment_client_declined(respx_mock):
    route = respx_mock.post("/charge").respond(402, json={"status": "DECLINED"})
    client = PaymentClient(base_url="https://pay.example", http=httpx.AsyncClient())

    result = await client.charge(ChargeRequest(card="card_123", amount=500))

    assert result.status is Status.DECLINED
    assert route.called

Половина работы здесь — строка PaymentClient(base_url=...): клиент должен брать адрес партнёра из конструктора или настроек, а не держать его константой. respx страхует от второй половины сам: по умолчанию он падает на любом запросе, для которого нет маршрута (AllMockedAssertionError), и на выходе проверяет, что все объявленные маршруты были вызваны. Тест, который «стучится к настоящему партнёру», с ним не напишется.

pytest-httpserver поднимает настоящий HTTP-сервер на свободном порту: запрос идёт через сокет, и клиент ведёт себя точно так, как в проде.

def test_payment_client_declined(httpserver):
    httpserver.expect_request("/charge", method="POST").respond_with_json({"status": "DECLINED"}, status=402)
    client = PaymentClient(base_url=httpserver.url_for("/"), http=httpx.AsyncClient())
    ...

Чем они отличаются, видно на трёх случаях.

Таймаут клиента. Партнёр отвечает медленнее, чем вы готовы ждать. В respx таймаут имитируют: route.mock(side_effect=httpx.ReadTimeout("slow")) — и тест проверяет, что клиент превращает его в свою ошибку. Но он не проверяет, что таймаут вообще настроен. Это проверяет только настоящий сервер с задержкой:

def test_payment_client_times_out(httpserver):
    def slow(request):
        time.sleep(3)
        return Response('{"status":"APPROVED"}', status=200, content_type="application/json")

    httpserver.expect_request("/charge").respond_with_handler(slow)
    client = PaymentClient(base_url=httpserver.url_for("/"), http=httpx.AsyncClient(timeout=0.2))

    with pytest.raises(PaymentUnavailable):
        asyncio.run(client.charge(request))

httpx.AsyncClient() без timeout в этом тесте прождёт три секунды и вернёт успех: у httpx таймаут по умолчанию пять секунд, и так вы увидите, что в проде клиент ждал бы партнёра по пять секунд на каждый вызов — дольше любого бюджета запроса. Такой тест заодно отвечает на вопрос, который иначе проверить нечем: соответствует ли таймаут бюджету.

Обрыв соединения. Не всякий отказ — код ответа. Сброшенное соединение обрабатывается другим кодом, чем 500, и ломает клиентов чаще. В respx это side_effect=httpx.ConnectError("reset"); с настоящим сервером — просто неверный порт, на котором никто не слушает.

Последовательность ответов и число повторов. Самая полезная возможность: «первые два раза плохо, третий хорошо» и проверка, что клиент сделал именно три попытки:

@respx.mock(base_url="https://pay.example")
async def test_payment_client_retries(respx_mock):
    route = respx_mock.post("/charge").mock(side_effect=[
        httpx.Response(503),
        httpx.Response(503),
        httpx.Response(200, json={"status": "APPROVED"}),
    ])
    client = PaymentClient(base_url="https://pay.example", http=httpx.AsyncClient())

    result = await client.charge(request)

    assert result.status is Status.APPROVED
    assert route.call_count == 3

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

Как выбрать: respx для логики клиента — разбор ответа, коды, повторы, обрыв; pytest-httpserver для того, что видно только через сокет, — таймауты, заголовки как они уходят на провод, потоковые тела. За пределами обоих остаётся сам контракт партнёра: заглушка отвечает так, как мы думаем, что отвечает партнёр. Расхождение ловят контрактными тестами или записью настоящих ответов — об этом раздел «Глубже» ниже.

Изоляция тестовых данных

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

Откат транзакции. Сессия на внешнем соединении с join_transaction_mode="create_savepoint", откат в конце фикстуры. Благодаря dependency_overrides годится и для сквозных тестов через ASGITransport; не годится для кода, который берёт соединение сам.

Очистка таблиц. TRUNCATE всех таблиц одним запросом перед тестом. Не годится при параллельном прогоне: пока один рабочий процесс чистит, второй читает свои данные и не находит их.

Уникальные идентификаторы. uuid4() на покупателя прямо в тесте, чтобы тесты не конкурировали за одни записи:

async def test_orders_of_customer(repo):
    customer = f"cust-{uuid4()}"
    await repo.save(order(customer, Status.PENDING))
    await repo.save(order(customer, Status.PAID))

    got = await repo.by_customer(customer)

    assert len(got) == 2

Уникальные ключи не ломаются ни при HTTP, ни при pytest-xdist. Цена одна: проверка «а сколько всего» перестаёт работать, считать надо всегда с условием по своему ключу.

Тест падает в полночь: время и случайность

Код с datetime.now() внутри недетерминирован: каждый прогон даёт другой результат, и точную проверку не написать. Решение простое — функция как зависимость:

from collections.abc import Callable
from datetime import UTC, datetime


class OrderService:
    def __init__(self, orders: OrderStore, now: Callable[[], datetime] = lambda: datetime.now(UTC)) -> None:
        self.orders = orders
        self.now = now

    async def place(self, request: OrderRequest) -> Order:
        order = Order.new(request, created_at=self.now())
        await self.orders.save(order)
        return order

async def test_place_stamps_created_at():
    fixed = datetime(2025, 1, 15, 10, 0, tzinfo=UTC)
    service = OrderService(orders=FakeOrders(), now=lambda: fixed)

    order = await service.place(order_request())

    assert order.created_at == fixed

Два замечания. datetime.now(UTC) в коде, а не в тесте: datetime.now() без аргумента возвращает наивное местное время машины, и одна и та же запись получит разное время на ноутбуке и на сервере; в базе отметка лежит в timestamptz, то есть это момент на общей шкале. И сравнивать нужно осведомлённые даты с осведомлёнными: Python откажется сравнивать наивную дату с датой с поясом (TypeError: can't compare offset-naive and offset-aware datetimes), и это полезная ошибка.

Когда время читает чужой код, который не переписать, есть time_machine: with time_machine.travel(fixed, tick=False): замораживает и datetime.now(), и time.time() на уровне интерпретатора, и в отличие от freezegun не замедляет прогон перебором модулей. Это лечит тесты с повторами по таймеру и сроками действия токенов, которые иначе спят секундами или мигают.

То же с генераторами случайных чисел: random.Random(seed) передают через конструктор, а зерно печатают при падении, чтобы повторить прогон.

Дополнительно: при первом чтении можно пропустить

Глубже: пять видов дублей: dummy, stub, fake, spy, mockрасширенное

Слово «заглушка» обозначает пять разных вещей, и спор «уместна или вредит» без разделения не решается.

Dummy передают, потому что параметр обязателен, и никогда не используют: None на месте уведомлений в тесте расчёта цены (если метод к ним не обращается). Stub отвечает заранее заданным — StubCharger выше: он нужен, чтобы тест дошёл до проверяемого места. Fake — работающая упрощённая реализация: FakeOrders на словаре. Spy записывает вызовы — create_autospec с проверкой await_count. Mock в строгом смысле знает ожидания заранее и падает при отклонении — respx с его «все маршруты должны быть вызваны».

Разница определяет, что тест проверяет. Stub и fake проверяют результат: что вернул метод, что оказалось в хранилище. Spy и mock проверяют взаимодействие: что метод вызвал соседа. Первое устойчиво, второе хрупко. Правило: проверять взаимодействие только там, где оно и есть результат, — у исходящих команд без возвращаемого значения («письмо отправлено», «событие опубликовано»); всё остальное проверять по состоянию. MagicMock на каждую зависимость с assert_called_with на каждый вызов — признак, что тест повторяет реализацию.

Глубже: контрактные тесты и записанные ответырасширенное

Тест на respx остаётся зелёным, когда партнёр меняет формат: заглушку никто не сверяет с настоящим сервисом. Есть три способа закрыть это.

Контракт со стороны потребителя (Pact, библиотека pact-python). Потребитель описывает в тесте, что отправляет и что ожидает получить; Pact поднимает заглушку из этого описания и на выходе даёт файл контракта. Его публикуют в брокер контрактов, а сборка поставщика проигрывает контракты всех потребителей против настоящего кода. Изменил поставщик поле, которое читает мобильное приложение, — его сборка красная с именем потребителя, до выката.

Проверка по спецификации. Если у партнёра есть OpenAPI, ответ заглушки и запрос клиента сверяют с ней в тесте: openapi-core валидирует запрос и ответ по схеме. Это не заменяет контракт, но ловит расхождение, когда заглушка написана «по памяти».

Записанные ответы. vcrpy записывает настоящие ответы партнёрского тестового контура в файл-кассету при первом прогоне и воспроизводит их дальше. Заглушка перестаёт быть вашим представлением о партнёре и становится его настоящим ответом на конкретную дату; перезапись по расписанию находит изменения формата. Кассеты чистят от токенов до коммита: фильтр заголовков в настройках vcrpy для этого и существует.

Для событий в очереди контракт важнее, чем для HTTP: там нет статуса 400, который скажет о расхождении схемы, — об этом в статье про Kafka.

Коротко

Заглушки уместны только на внешней границе: маленький Protocol объявляет потребитель, заглушка — рукописный класс; create_autospec и AsyncMock — когда их десятки, и autospec обязателен, иначе опечатка в имени метода пройдёт молча.
Подменять свой репозиторий — тестировать ожидания: для логики берут fake на словаре, для запросов — настоящую базу в контейнере.
Чужие классы (AsyncSession, httpx.AsyncClient, клиент брокера), значения и функции модуля не подменяют: своя тонкая обёртка, контейнер или заглушка на уровне протокола; patch «там, где используется» — симптом скрытой зависимости.
Факт вызова проверяют assert_awaited_* и await_count: ноль вызовов, ровно три попытки, аргументы; взаимодействие проверяют только у исходящих команд.
respx заменяет партнёра на уровне транспорта и сам падает на незамоканном или невызванном маршруте; pytest-httpserver — настоящий сокет, и только он проверяет, что таймаут настроен.
Повторы проверяют последовательностью side_effect и call_count; обрыв — httpx.ConnectError; паузы между повторами в тесте обнуляют.
Изоляция данных: откат через create_savepoint, TRUNCATE для фиксации, уникальные идентификаторы для pytest-xdist.
Время внедряют функцией now, пишут datetime.now(UTC), сравнивают осведомлённые даты; чужой код замораживают time_machine; случайность — random.Random(seed) с зерном.
Контракт сверяет заглушку с партнёром: pact-python через брокер, openapi-core по спецификации, vcrpy записанными ответами без токенов в кассетах.

Что почитать дальше

Интеграционные тесты на Python — testcontainers, миграции и изоляция данных подробно.
Тестирование FastAPI — TestClient, httpx.AsyncClient и подмена зависимостей с нуля.
Пирамида тестирования — какой уровень тестов за что отвечает.
Паттерны отказоустойчивости на Python — таймауты, повторы и предохранитель, которые эти тесты проверяют.