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

Асинхронный код тестируется так же, как синхронный, с одной поправкой: тесту нужен цикл событий, а объектам с состоянием (клиентам, пулам) нужна правильная область жизни, иначе тесты ломаются загадочными «attached to a different loop». Разберём инструменты, которые стоят в проекте на FastAPI, и четыре вида тестов, которые ловят именно асинхронные ошибки: отмену, таймаут, гонку и утечку задач.

Обязательно

Запуск корутин в pytest

pytest сам не умеет выполнять async def-тесты. Два плагина это умеют. pytest-asyncio помечает тесты @pytest.mark.asyncio (или выполняет все async def при asyncio_mode = auto) и даёт асинхронные фикстуры. Плагин anyio, который приходит вместе со Starlette, помечает тесты @pytest.mark.anyio и умеет запускать их на разных реализациях цикла. В проекте выбирают один; дальше примеры на pytest-asyncio, они проверены на версии 1.4.

[pytest]
asyncio_mode = strict
asyncio_default_fixture_loop_scope = function
filterwarnings = error::RuntimeWarning

Третья строка превращает coroutine ... was never awaited в ошибку: забытый await роняет тест вместо тихого предупреждения. Режим strict требует явной пометки тестов и фикстур: асинхронная фикстура объявляется через @pytest_asyncio.fixture, а обычный @pytest.fixture над async def в этом режиме даёт ошибку «requested an async fixture with no plugin or hook that handled it» (проверено на pytest-asyncio 1.4). В режиме auto подходит обычный декоратор, но тогда любой async def считается тестом asyncio, что мешает смешивать плагины.

Область цикла событий это главный источник проблем. По умолчанию каждый тест получает свой цикл. Фикстура уровня session, которая создала клиент или пул в одном цикле, при использовании из теста с другим циклом падает: у asyncpg и httpx объекты привязаны к циклу, в котором их создали. Решение одно из двух: либо клиенты создают в фикстуре той же области, что тесты (на каждый тест), либо поднимают область цикла до модуля или сессии через asyncio_default_fixture_loop_scope = session и @pytest.mark.asyncio(loop_scope="session"). Для интеграционных тестов с реальной базой второй вариант быстрее, потому что пул создаётся один раз.

Обработчики FastAPI без сети

Для тестов обработчиков не нужен uvicorn: httpx.ASGITransport вызывает приложение напрямую, через тот же протокол ASGI, что и сервер.

import httpx, pytest, pytest_asyncio
from app.main import app

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

@pytest.mark.asyncio
async def test_get_order(client):
    r = await client.get("/orders/42")
    assert r.status_code == 200

В отличие от синхронного TestClient, этот путь выполняет обработчик в том же цикле, что и тест: можно подменять зависимости через app.dependency_overrides, ставить точки останова, проверять asyncio.all_tasks() после запроса. Чего ASGITransport не делает: не запускает lifespan. Если клиентам из app.state нужен lifespan, его открывают в фикстуре явно через async with app.router.lifespan_context(app) или используют TestClient как контекстный менеджер, который lifespan выполняет. Подробнее о тестировании FastAPI в статье про тесты FastAPI.

Соседние сервисы: respx

Исходящие вызовы через httpx перехватывает respx: он подменяет транспорт и отвечает заранее заданными ответами, без сети и без изменений в коде сервиса.

import respx

@pytest.mark.asyncio
@respx.mock
async def test_order_with_catalog(client):
    route = respx.get("https://catalog.internal/products/42").mock(
        return_value=httpx.Response(200, json={"name": "Кофе"})
    )
    r = await client.get("/orders/42")
    assert r.json()["product"] == "Кофе"
    assert route.called

Проверено на respx 0.23 и httpx 0.28: маршрут отвечает, route.called фиксирует вызов. Тем же способом проверяют поведение при ошибках соседа: side_effect=httpx.ConnectTimeout("...") для таймаута, httpx.Response(503) для отказа, и это единственный честный способ протестировать ветки с повтором и запасным вариантом. Какие внешние вызовы подменять, а какие поднимать в контейнерах, разбирает статья про заглушки внешних HTTP.

Тест отмены

Поведение при отмене проверяют явно, потому что оно ломается незаметно. Шаблон: запустить задачу, отменить, убедиться, что она завершилась отменой за разумное время и освободила ресурсы.

@pytest.mark.asyncio
async def test_worker_stops_on_cancel():
    released = asyncio.Event()
    async def worker():
        try:
            await asyncio.sleep(10)
        finally:
            released.set()
    task = asyncio.create_task(worker())
    await asyncio.sleep(0)
    task.cancel()
    with pytest.raises(asyncio.CancelledError):
        await task
    assert released.is_set()

Если await task не поднимает CancelledError или зависает, в задаче проглочена отмена. Такой тест пишут на каждого фонового потребителя и на каждый долгоживущий цикл: он занимает миллисекунды и ловит класс ошибок, который иначе виден только при выкате.

Тест таймаута и бюджета

Таймауты тестируют двумя способами. Первый: соседу через respx задают медленный ответ (side_effect с await asyncio.sleep), а у обработчика проверяют, что он отвечает ошибкой или запасным значением в пределах бюджета, измеряя время теста. Второй, для чистых корутин: asyncio.timeout с маленьким значением и pytest.raises(TimeoutError). Важно использовать маленькие реальные задержки, а не подменять часы: цикл событий asyncio нельзя ускорить подменой time, и попытки заморозить время ломают его таймеры. Тест на 10 миллисекунд достаточно быстр, а на 100 миллисекунд достаточно надёжен в CI.

Тест гонки

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

@pytest.mark.asyncio
async def test_rate_fetched_once_under_concurrency(monkeypatch):
    calls = 0
    async def fake_fetch(currency):
        nonlocal calls
        calls += 1
        await asyncio.sleep(0.01)
        return Rate(currency, 100)
    monkeypatch.setattr(rates, "fetch_rate", fake_fetch)
    await asyncio.gather(*(rates.get_rate("USD") for _ in range(20)))
    assert calls == 1

Без замка или Future на ключ calls будет 20. await asyncio.sleep(0.01) в подделке обязателен: именно он создаёт точку переключения, в которой проявляется гонка; подделка без await гонку скроет.

Тест утечки задач

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

@pytest_asyncio.fixture(autouse=True)
async def no_leaked_tasks():
    yield
    leaked = {t for t in asyncio.all_tasks() if t is not asyncio.current_task()}
    assert not leaked, [t.get_name() for t in leaked]

Она ловит обработчики, которые создают задачу и не ждут её, и фоновые циклы, которые не остановились. Для тестов с lifespan фикстуру ставят после закрытия приложения.

Интеграционные тесты с реальной базой и брокером

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

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

Глубже: режим отладки и порядок в тестахрасширенное

Режим отладки asyncio в тестах стоит держать включённым постоянно: PYTHONASYNCIODEBUG=1 в окружении CI или asyncio_debug через фикстуру, которая ставит loop.set_debug(True) и loop.slow_callback_duration = 0.05. Тогда блокирующий вызов в обработчике даст предупреждение Executing <Task> took ... прямо в прогоне тестов, а с filterwarnings = error ещё и уронит его. Это дешёвый способ не пропустить time.sleep или синхронный драйвер в код, который должен быть асинхронным.

Про порядок: тесты асинхронного кода особенно чувствительны к общему состоянию между тестами, потому что модульные одиночки (клиент, пул, кеш _inflight) переживают тест. Симптом: тест зелёный в одиночку и красный в наборе. Лекарство структурное: одиночки создаются в lifespan и живут в app.state, а не в модулях, и тогда каждый тест, который открывает приложение заново, получает чистое состояние. Там, где модульное состояние неизбежно (словарь замков по ключу), фикстура очищает его перед тестом.

И последнее про pytest-asyncio 1.x: старая фикстура event_loop удалена, цикл не подменяют руками; если тесту нужен цикл, его берут через asyncio.get_running_loop() внутри теста, а область задают маркером.

Коротко

  • Тесты async def запускает pytest-asyncio (@pytest.mark.asyncio, фикстуры через @pytest_asyncio.fixture в режиме strict) или плагин anyio; в проекте один из них; filterwarnings = error::RuntimeWarning ловит забытый await.
  • Область цикла событий у фикстур должна совпадать с областью клиентов: либо клиенты на тест, либо loop_scope="session" для дорогих пулов.
  • Обработчики FastAPI тестируют через httpx.ASGITransport в том же цикле; lifespan при этом открывают явно или через TestClient.
  • Соседние сервисы подменяет respx: ответы, ошибки, таймауты через side_effect; это единственный способ протестировать ветки повтора и деградации.
  • Тест отмены: cancel, pytest.raises(CancelledError), проверка освобождения ресурсов; пишется на каждого фонового потребителя.
  • Таймауты тестируют маленькими реальными задержками, часы не подменяют; гонку воспроизводят gather одинаковых операций с await внутри подделки.
  • Автофикстура проверяет, что после теста нет живых задач; реальные база и брокер в контейнерах один раз на сессию.
  • Режим отладки asyncio в CI включён; одиночки живут в app.state, а не в модулях, иначе тесты зависят от порядка.

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