Детерминированные тесты на асинхронность в Python: тестовая среда для таймаутов, ретраев и гонок

Асинхронный код в Python — это не просто async/await. Это целая система допущений: планирование задач в event loop, тайминг, отмены, повторные попытки, обработка исключений, внешние зависимости (сеть, файловая система, очередь), и — главное — то, что без правильной дисциплины тесты становятся непредсказуемыми. В CI они то проходят, то падают. Локально всё зелёное, а в пайплайне — флейк. Классическая история.

Эта статья — практический разбор того, как проектировать детерминированные (воспроизводимые) автотесты для async-кода: как управлять временем, изолировать внешние зависимости, надёжно воспроизводить отмену задач и проверять ретраи без флейка. Параллельно обсудим типичные ошибки и дадим готовые шаблоны.

Почему асинхронность “ломает” тесты

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

1. Event loop планирует задачи не по вашему желанию, а по готовности await‑источников. Это означает, что «когда именно» выполнится конкретная строка зависит от таймингов.
2. Операции ожидания (await asyncio.sleep, I/O, ожидание событий) могут вести себя по-разному в зависимости от нагрузки и скорости CI‑агентов.
3. Отмена задач (task.cancel()) переводит корутину в специфический режим — многие места кода должны корректно обработать CancelledError, и это легко упустить.
4. Ретраи часто завязаны на sleep/backoff — значит, тесты могут стать чувствительными к реальному времени или к очередности планирования.
5. Гонки возникают на уровне логики: например, два параллельных запроса обновляют состояние, а вы ожидаете одно и то же значение. В тестах это проявляется как нестабильность.

Вывод: стабильность тестов нужно создавать внутри тестовой среды, а не надеяться, что “CI случайно не подведёт”.

Что значит “детерминированный” тест

Детерминированность в контексте async‑тестов — это не “всегда одинаковый результат при любом окружении”. Это обычно означает:

• Тест не зависит от реального времени (или зависит строго контролируемо).
• Любые “внешние” эффекты (сеть, очередь, таймеры, случайность) заменены на управляемые фейковые реализации.
• Порядок событий зафиксирован: вы сами управляете, когда корутины становятся готовыми, когда происходит отмена, когда выдаётся ответ.
• Ошибки воспроизводимы: если вы тестируете отмену, вы знаете точно, в какой момент она произошла.

На практике детерминированность строится из трёх кирпичей: контроль времени, контроль зависимостей, контроль планирования/синхронизации.

Контроль времени: как убрать sleep из тестов

Самая частая причина флейка — использование реального asyncio.sleep в коде и ожидание “пока пройдёт время”. В CI задержки непредсказуемы. Поэтому правильный подход — заменить тайм-ауты и backoff на управляемые.

Стратегия 1: “переопределить” sleep на фейк

Если ваша функция использует asyncio.sleep, можно параметризовать sleep (инъекция зависимости). Например:

# app/retry.py
import asyncio
from typing import Awaitable, Callable, TypeVar

T = TypeVar("T")

async def retry_async(
    op: Callable[[], Awaitable[T]],
    *,
    attempts: int = 3,
    backoff: float = 0.1,
    sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
    retry_exceptions: tuple[type[BaseException], ...] = (Exception,),
) -> T:
    last_exc: BaseException | None = None
    for i in range(attempts):
        try:
            return await op()
        except retry_exceptions as exc:
            last_exc = exc
            if i == attempts - 1:
                raise
            await sleep(backoff * (2 ** i))
    assert last_exc is not None
    raise last_exc

Теперь тест может передать собственную sleep, которая не ждёт реально, а только регистрирует “виртуальное” время.

# tests/test_retry_deterministic.py
import pytest
from app.retry import retry_async

class FakeClock:
    def __init__(self):
        self.slept: list[float] = []
        self.time = 0.0

    async def sleep(self, seconds: float):
        self.slept.append(seconds)
        self.time += seconds
        # никаких реальных ожиданий

@pytest.mark.asyncio
async def test_retry_backoff_is_deterministic():
    clock = FakeClock()
    calls = {"n": 0}

    async def op():
        calls["n"] += 1
        if calls["n"] < 3:
            raise ValueError("temporary")
        return "ok"

    res = await retry_async(
        op,
        attempts=3,
        backoff=0.1,
        sleep=clock.sleep,
        retry_exceptions=(ValueError,),
    )

    assert res == "ok"
    assert calls["n"] == 3
    assert clock.slept == [0.1, 0.2]  # backoff*1, backoff*2
    assert clock.time == 0.3

Этот тест детерминирован: он не зависит от нагрузки и скорости CI.

Подводные камни

• Если asyncio.sleep захардкожен внутри функции и вы не можете его инъектировать — придётся рефакторить. Это нормально: тестируемость — часть дизайна.
• Иногда вы используете тайм-ауты через asyncio.wait_for или asyncio.timeout. Тогда лучше параметризовать таймаут‑поведение через обвязку либо применять фейковый “таймер” (см. ниже).

Таймауты: тестируем wait_for, не плавая по времени

Проверка таймаутов часто выглядит как “пусть не успеет и упадёт”. Это почти гарантированно приводит к флейку, если код использует реальное время.

Ключевой приём: вместо ожидания заставить корутину “никогда не завершаться” и управлять отменой через события/синхронизацию.

Пример: функция с таймаутом

Допустим, у вас есть:

# app/timeout.py
import asyncio

async def fetch_with_timeout(fetch_coro, timeout: float):
    return await asyncio.wait_for(fetch_coro, timeout=timeout)

Чтобы тест был детерминированным, не делайте “sleep > timeout”. Сделайте fetch_coro, который ждёт событие, которое тест не будет ставить.

# tests/test_timeout_deterministic.py
import asyncio
import pytest

from app.timeout import fetch_with_timeout

@pytest.mark.asyncio
async def test_timeout_raises_without_real_sleep():
    event = asyncio.Event()

    async def fetch_coro():
        await event.wait()  # никогда не завершится в тесте

    # Таймаут всё равно использует реальное время wait_for.
    # Но мы делаем его минимальным и проверяем тип исключения.
    with pytest.raises(asyncio.TimeoutError):
        await fetch_with_timeout(fetch_coro(), timeout=0.01)

Да, здесь всё ещё есть реальный таймаут. Но флейка можно избежать, если:

• таймаут очень маленький (и допускает погрешность),
• event loop не перегружен тестами,
• тесты не “копят” задержки.

Для особо строгой детерминированности лучше заменить таймаут‑механизм. Практичный вариант — завернуть таймаут в абстракцию, которая позволяет тесту “отстрелить” отмену без реального времени.

Воспроизводимая отмена задач: тестируем CancelledError без случайностей

Отмена в asyncio — тонкая тема. task.cancel() не “останавливает немедленно”. Она инъектит сигнал отмены, а корутина должна встретить точку, где CancelledError будет выброшен. Если корутина игнорирует отмену (или вы делаете долгие CPU‑операции без await), отмена может повести себя неожиданно.

Стратегия 1: контролируйте место отмены через синхронизацию

Допустим, есть функция, которая делает шаги и может быть отменена между ними:

# app/cancel.py
import asyncio

async def worker_stepwise(stop_event: asyncio.Event):
    # шаг 1
    await asyncio.sleep(0)  # точка планирования
    if stop_event.is_set():
        return "stopped"

    # шаг 2: "длинное ожидание"
    await stop_event.wait()
    return "done"

Чтобы детерминированно отменить задачу в “шаге 2”, используйте две сущности:

• событие, показывающее, что корутина дошла до нужного места,
• и task.cancel() из теста.

# tests/test_cancel_deterministic.py
import asyncio
import pytest

from app.cancel import worker_stepwise

@pytest.mark.asyncio
async def test_worker_cancellation_is_handled():
    stop_event = asyncio.Event()
    reached_wait = asyncio.Event()

    async def instrumented_worker():
        await asyncio.sleep(0)
        reached_wait.set()
        if stop_event.is_set():
            return "stopped"
        await stop_event.wait()
        return "done"

    task = asyncio.create_task(instrumented_worker())

    # Ждём сигнал, что корутина дошла до wait()
    await reached_wait.wait()

    task.cancel()
    with pytest.raises(asyncio.CancelledError):
        await task

Это детерминировано: вы отменяете именно в момент, когда корутина уже стоит на await.

Частая ошибка: отмена “слишком рано” или “слишком поздно”

Если отменить задачу до того, как она дошла до точки ожидания, то результат зависит от того, успеет ли корутина обработать отмену и какие ветки пройдут. Поэтому в тестах полезно фиксировать точку синхронизации.

Проверка ретраев и отсутствие флейка: тестируем временные диаграммы логики

Ретраи — это сочетание обработки исключений, количества попыток, backoff и (иногда) “не ретраить при некоторых ошибках”. Главная цель тестов — проверить логическую последовательность, а не “как быстро пройдёт время”.

Базовая модель ретрая: сколько раз, с какими параметрами

У вас уже есть пример retry_async, который мы тестировали с FakeClock. Этого достаточно, чтобы проверить:

• количество вызовов op,
• какие исключения ретраились,
• что backoff вычисляется корректно.

Тест на “не ретраить” (например, для фатальных ошибок)

# tests/test_retry_no_fatal_retry.py
import pytest
from app.retry import retry_async

class FakeClock:
    def __init__(self):
        self.slept = []
    async def sleep(self, seconds: float):
        self.slept.append(seconds)

class FatalError(Exception):
    pass

@pytest.mark.asyncio
async def test_does_not_retry_on_fatal_error():
    clock = FakeClock()
    calls = {"n": 0}

    async def op():
        calls["n"] += 1
        raise FatalError("fatal")

    with pytest.raises(FatalError):
        await retry_async(
            op,
            attempts=3,
            backoff=0.1,
            sleep=clock.sleep,
            retry_exceptions=(ValueError,),  # ретраим только ValueError
        )

    assert calls["n"] == 1
    assert clock.slept == []

Что проверять дополнительно в сложных ретраях

Если ваш код делает ретраи с “jitter” (случайность) или корректирует backoff с учётом заголовков/контекста — детерминируйте и это:

• внедряйте генератор случайных чисел,
• внедряйте функцию вычисления задержки,
• либо тестируйте результат через паттерны (диапазоны), но не через точные значения, если генератор нельзя стабилизировать.

Гонки: как тестировать параллельный код без случайности

Гонки — самая “враждебная” область. Если вы тестируете “ожидаемое итоговое значение после двух параллельных задач”, а внутри есть конкуренция за общий ресурс, то флейк почти неизбежен, пока вы не управляете порядком выполнения.

Правило: тесты на гонки должны управлять очередностью, а не надеяться.

Стратегия 1: управляемый план выполнения через asyncio.Event/барьеры

Рассмотрим условную проблему: две задачи одновременно увеличивают счётчик без блокировки — и вы ловите неверное итоговое значение (или проверяете, что блокировка помогает).

Пусть у вас есть функция:

# app/race.py
import asyncio

async def unsafe_increment(counter: dict, key: str):
    # имитируем read-modify-write
    v = counter[key]
    await asyncio.sleep(0)  # уступаем управление
    counter[key] = v + 1

Чтобы детерминированно воспроизвести гонку, мы уберём sleep(0) из фактора реального времени и заменим его точкой синхронизации, которой управляет тест.

# tests/test_race_deterministic.py
import asyncio
import pytest

@pytest.mark.asyncio
async def test_race_can_lose_updates_with_controlled_interleaving():
    counter = {"x": 0}
    t1_can_write = asyncio.Event()
    t2_can_write = asyncio.Event()

    async def t1():
        v = counter["x"]
        t1_can_write.set()          # t1 завершил read
        await t2_can_write.wait()  # ждём, пока t2 тоже прочтёт
        counter["x"] = v + 1       # write 1

    async def t2():
        await t1_can_write.wait()  # ждём read t1
        v = counter["x"]           # read (теперь оба прочли одно значение)
        t2_can_write.set()         # разрешаем t1 write
        await asyncio.sleep(0)     # точка планирования, можно заменить на событие
        counter["x"] = v + 1       # write 2 (переопишет то же значение)

    await asyncio.gather(asyncio.create_task(t1()), asyncio.create_task(t2()))
    assert counter["x"] == 1  # потеря обновления

Этот тест не “ловит гонку случайно”. Он моделирует конкретное чередование операций: read/read/write/write, поэтому результат фиксирован.

Стратегия 2: тестировать фиксы гонок (locks/transactions)

Если вы исправляете код, тест должен проверять, что при правильной синхронизации итог стабилен. Например, используйте asyncio.Lock и повторите ту же схему управления событиями — но уже ожидайте 2.

Идея такая: структура теста остаётся, меняется только механизм синхронизации.

Изоляция внешних зависимостей: фейки вместо настоящего I/O

Асинхронный код особенно часто “утекает” в интеграционные тесты: реальные HTTP-запросы, реальные очереди, реальные базы. В CI это нестабильно по многим причинам: тайминги сети, параллельность контейнеров, лимиты ресурсов.

Детерминизм достигается через фейки:

• Фейковый клиент (например, HTTP) возвращает заранее подготовленные ответы.
• Фейковый таймер — как минимум вместо sleep используется управляемая функция.
• Фейковый стор (репозиторий/очередь) — синхронизирует события, которые тесту нужно контролировать.

Пример: ретраи при временных ошибках без сети

Допустим, функция делает HTTP call и ретраит по 5xx:

# app/http_retry.py
import asyncio
from dataclasses import dataclass

class Http5xx(Exception):
    pass

@dataclass
class FakeResponse:
    status: int

async def get_with_retry(client, url: str, *, attempts: int, sleep=asyncio.sleep):
    for i in range(attempts):
        try:
            resp: FakeResponse = await client.get(url)
            if resp.status >= 500:
                raise Http5xx(f"status={resp.status}")
            return resp
        except Http5xx:
            if i == attempts - 1:
                raise
            await sleep(0.1)

Тестируем поведение с фейковым клиентом и FakeClock:

# tests/test_http_retry_deterministic.py
import pytest
from app.http_retry import get_with_retry, FakeResponse, Http5xx

class FakeClock:
    def __init__(self):
        self.slept = []
    async def sleep(self, seconds: float):
        self.slept.append(seconds)

class FakeClient:
    def __init__(self, statuses):
        self.statuses = list(statuses)
        self.calls = 0
    async def get(self, url: str):
        self.calls += 1
        status = self.statuses.pop(0)
        return FakeResponse(status=status)

@pytest.mark.asyncio
async def test_retry_on_5xx_without_flaky_timing():
    clock = FakeClock()
    client = FakeClient([500, 502, 200])

    resp = await get_with_retry(client, "http://x", attempts=3, sleep=clock.sleep)

    assert resp.status == 200
    assert client.calls == 3
    assert clock.slept == [0.1, 0.1]

Организация тестовой среды для async: важные практики

Используйте воспроизводимые паттерны синхронизации

Если тесту нужно “дождаться момента X”, то обычно нужен:

• asyncio.Event,
• или asyncio.Queue с контролируемым количеством сообщений,
• или asyncio.Semaphore для ограничения параллелизма.

Не полагайтесь на asyncio.sleep(...) как на способ синхронизироваться. Это работает “почти всегда”, пока не перестаёт.

Учитывайте очистку задач

В async-тестах часто остаются фоновые задачи. В зависимости от раннера (pytest-asyncio, anyio) это может:

• приводить к предупреждениям,
• портить состояние event loop,
• создавать редкие падения.

Паттерн: держите ссылки на Task и гарантируйте завершение/отмену в finally.

Минимизируйте общие глобальные стейты

Гонки между тестами — частая причина “случайных” падений. Убедитесь, что тесты:

• не используют общий глобальный объект без изоляции,
• не оставляют изменённые синглтоны,
• не зависимы от порядка запуска.

Набор готовых “строительных блоков” для детерминированных тестов

Ниже — небольшая “библиотека идей” на уровне паттернов, которую можно перенести в проект.

1) FakeClock для контроля backoff

• Инъектируйте sleep.
• Регистрируйте аргументы и “виртуальное время”.

2) Instrumentation Events для контроля точек выполнения

• reached_wait, before_write, after_read — любые “маркеры” шага.

3) Фейковый I/O

• HTTP-клиент возвращает подготовленные статусы.
• Очередь отдаёт сообщения по запросу.
• База возвращает заранее сформированные данные.

4) Управляемое чередование для гонок

• Два шага read, write выстраиваются через события.
• В тесте вы проверяете конкретное состояние, ожидаемое при данном interleaving.

Типичные причины флейка и как их устранить

1. Реальное ожидание вместо синхронизации

   • Было: await asyncio.sleep(0.2) чтобы “дать задаче стартовать”.
   • Стало: await started_event.wait().

2. Слишком большие допуски

   • Если таймауты рассчитаны “на глаз”, в CI они могут не успеть завершиться.
   • Решение: детерминируйте отмену/события или уменьшите таймаут и сократите “плавающую” нагрузку.

3. Игнорирование отмены

   • Корутинa ловит Exception и случайно скрывает CancelledError.
   • Решение: явно обработать asyncio.CancelledError или не перехватывать её неявно.

4. Непредсказуемая очередность запросов

   • Если вы запускаете несколько задач и ожидаете порядок, нужна явная синхронизация.

5. Оставленные фоновые задачи

   • Решение: отмена и ожидание task в finally, использование asyncio.gather(..., return_exceptions=True) где уместно.

Итоги: как сделать тесты на async стабильными в CI

Детерминированные тесты для асинхронного кода — это не “магия”, а инженерная дисциплина. Вам нужно:

• Контролировать время: вместо asyncio.sleep в тестах используйте инъектируемый sleep (FakeClock) или управляемые задержки.
• Изолировать зависимости: любые сетевые/дисковые операции заменяйте фейками, которые заранее знают сценарий.
• Управлять точками выполнения: тест должен инициировать отмену/продолжение корутины через события, а не “угадать” тайминг.
• Тестировать гонки через interleaving, а не через шанс: фиксируйте порядок read/write через Event/Queue.
• Следить за очисткой задач: отмена и ожидание фоновых задач убирают массу “редких” падений.

Если вы хотите системно закрепить подходы — от структуры тестов до практик для CI — полезно пройти материал по автоматизации тестирования на Python: «Автоматизация тестирования на Python». Это не заменит понимание нюансов (оно как раз важно), но поможет собрать в одну систему лучшие практики из разных кейсов.

В следующий раз, когда ваш async‑тест начнёт флейкать, не увеличивайте таймауты “на всякий случай”. Сначала проверьте, где вы опираетесь на реальное время, где допускаете неконтролируемую очередность и есть ли у теста явные точки синхронизации. Обычно этого достаточно, чтобы вернуть предсказуемость.