Тестирование асинхронного кода в Python: типовые грабли

Асинхронный тест проходит, а в продакшне тот же код виснет или тихо теряет исключения. Причина почти всегда одна из четырёх: забытый await, который не проверяет ничего; общий event loop, из-за которого один тест ломает следующий; Mock вместо AsyncMock, отдающий не то; и незавершённые фоновые задачи, до которых тест не дожидается. Разберём каждую на рабочем коде под Python 3.12 и pytest-asyncio 0.23+, покажем текст ошибки, который вы реально увидите, и как переписать, чтобы тест проверял то, что должен.

Речь только про грабли самих тестов. Отладка гонок в проде, трейсинг и профилирование asyncio — отдельные темы, здесь не трогаем.

Грабля 1. Забытый await: тест проверяет корутину, а не результат

Самая частая и самая тихая. Если убрать await перед вызовом корутины, Python не ругается сразу — он возвращает объект корутины. Сравнение с ним обычно проходит совсем не так, как вы думаете, а иногда проходит случайно.

import asyncio

async def fetch_price(item_id: int) -> int:
    await asyncio.sleep(0)  # имитация запроса
    return item_id * 10

# Тест с ошибкой: await потерян
async def test_price_broken():
    result = fetch_price(5)   # это корутина, а не 50
    assert result != 42       # проходит, но проверяет мусор

# Правильно
async def test_price_ok():
    result = await fetch_price(5)
    assert result == 50

В сломанном тесте result — объект корутины. Он не равен 42, ассерт зелёный, тест бесполезен. При запуске такого файла pytest дополнительно выдаёт предупреждение RuntimeWarning: coroutine 'fetch_price' was never awaited — это ваш главный сигнал. Не глушите предупреждения корутин глобально.

Правило простое: если в теле теста встречается вызов async def-функции без await и без явной обёртки в задачу — это баг, а не оптимизация.

Чтобы такие места падали, а не молчали, включите в pytest.ini превращение предупреждений в ошибки:

[pytest]
filterwarnings =
    error::RuntimeWarning

Теперь coroutine ... was never awaited роняет тест, и забытый await не проскочит в мёрж.

Грабля 2. Общий event loop и утечки состояния между тестами

В pytest-asyncio 0.23 поменялась модель работы с циклом событий. Появился режим, задаваемый через asyncio_mode, и фикстура event_loop, которую раньше переопределяли вручную, теперь deprecated. Если вы тащите старые примеры из интернета с @pytest.fixture def event_loop(), получите предупреждение об устаревании, а в будущих версиях — поломку.

Практическая грабля: если два теста используют один и тот же цикл и один из них оставил висящую задачу или незакрытый клиент, второй тест падает с RuntimeError: Event loop is closed или ведёт себя недетерминированно. Проявляется классически — тест зелёный по одному, красный в общем прогоне.

Как настроить режим

Для большинства проектов достаточно авторежима — тогда не нужно помечать каждый тест декоратором:

[pytest]
asyncio_mode = auto

Область жизни цикла и связанных фикстур в 0.23 задаётся через asyncio_default_fixture_loop_scope. Значение по умолчанию менялось между релизами, поэтому задайте его явно, чтобы не гадать:

[pytest]
asyncio_mode = auto
asyncio_default_fixture_loop_scope = function

Scope function — самый безопасный: свежий цикл на каждый тест, состояние не протекает. Если вам нужен общий дорогой ресурс на весь модуль, поднимайте scope осознанно и следите, чтобы ресурс закрывался. Точную семантику новых опций сверяйте с документацией вашей версии pytest-asyncio — она активно менялась в ветке 0.2x.

Грабля 3. Mock вместо AsyncMock

Если вы мокаете корутину обычным Mock, вызов вернёт Mock, а не awaitable. При попытке await получите TypeError: object Mock can't be used in 'await' expression. С MagicMock ошибка тоньше: он awaitable-совместим не полностью, и поведение зависит от того, что именно вы дёргаете.

Начиная с Python 3.8 в стандартной библиотеке есть unittest.mock.AsyncMock — используйте его для всего асинхронного.

from unittest.mock import AsyncMock

async def get_user(repo, user_id: int) -> dict:
    return await repo.fetch(user_id)

async def test_get_user():
    repo = AsyncMock()
    repo.fetch.return_value = {"id": 7, "name": "Ann"}

    result = await get_user(repo, 7)

    assert result == {"id": 7, "name": "Ann"}
    repo.fetch.assert_awaited_once_with(7)  # именно awaited, не called

Обратите внимание на assert_awaited_once_with, а не assert_called_once_with. Первый проверяет, что мок именно ждали через await. Если в коде потеряли await (грабля 1), assert_called_* может пройти, а assert_awaited_* честно упадёт. Это ваш способ поймать первую граблю через мок.

Когда патчите через patch, autospec=True сам подставит AsyncMock для корутин, если мокаемый объект действительно объявлен как async def. Без autospec придётся указывать new_callable=AsyncMock вручную.

Грабля 4. Незавершённые фоновые задачи

Код создаёт задачу через asyncio.create_task и не ждёт её. Тест завершается раньше, чем задача успела выполниться или упасть. Итог: либо ассерт проверяет состояние до того, как задача его изменила, либо исключение внутри задачи теряется и всплывает как Task was destroyed but it is pending уже в чужом тесте.

import asyncio

class Notifier:
    def __init__(self):
        self.sent = []
        self._tasks: set[asyncio.Task] = set()

    def schedule(self, msg: str) -> None:
        task = asyncio.create_task(self._send(msg))
        # держим ссылку, иначе задачу может собрать GC
        self._tasks.add(task)
        task.add_done_callback(self._tasks.discard)

    async def _send(self, msg: str) -> None:
        await asyncio.sleep(0)
        self.sent.append(msg)

    async def drain(self) -> None:
        # дождаться всех запланированных задач
        if self._tasks:
            await asyncio.gather(*self._tasks)

async def test_notifier():
    n = Notifier()
    n.schedule("hello")
    # без drain ассерт ниже упадёт: sent ещё пустой
    await n.drain()
    assert n.sent == ["hello"]

Два вывода. Первый: храните ссылки на задачи — create_task не удерживает их сам, и сборщик мусора может убить задачу до завершения (это описано в документации asyncio). Второй: в тесте дожидайтесь фоновой работы явно через gather или await task, а не через asyncio.sleep наугад — сон гадает и делает тест хрупким.

Тайм-ауты, чтобы тест не висел вечно

Если фоновая задача из-за бага не завершается, тест зависнет. Ограничьте ожидание. В Python 3.11+ есть asyncio.timeout:

async def test_with_timeout():
    async with asyncio.timeout(2):  # Python 3.11+
        await do_something()

На уровне всего прогона добавьте плагин pytest-timeout — он прибьёт зависший тест по общему лимиту вместо того, чтобы вешать CI.

Сводка: грабля, симптом, лечение

Что настроить дальше

Минимальный порядок действий на новом проекте:

1. Пропишите в pytest.ini asyncio_mode = auto и явный asyncio_default_fixture_loop_scope = function.
2. Включите filterwarnings = error::RuntimeWarning, чтобы забытые await роняли тесты, а не прятались.
3. Поставьте pytest-timeout и задайте общий лимит на тест — зависший асинхронный тест не должен вешать CI.
4. Проверьте, что все моки корутин — AsyncMock, и что вы используете assert_awaited_*, а не assert_called_*.

После этого смотрите на флаки: тест, красный только в полном прогоне и зелёный в одиночку, почти всегда указывает на утечку состояния через loop или незавершённую задачу. Прогоняйте набор с -p no:randomly и наоборот со случайным порядком, чтобы такие зависимости всплыли до прода.