Мокирование времени в async: asyncio.sleep и retry-логика

Этот материал был частью продвинутого курса по pytest, но был вынесен отдельно как специализированная тема. Если вам нужно тестировать async код с задержками — этот гайд для вас.

Для кого этот материал?

✅ Вы работаете с async/await

✅ Ваш код использует

Проблема: медленные тесты

# slow_service.py
import asyncio
 
async def fetch_with_retry(url: str, retries: int = 3):
    """Retry с exponential backoff"""
    for attempt in range(retries):
        try:
            result = await fetch(url)
            return result
        except Exception:
            if attempt < retries - 1:
                await asyncio.sleep(2 ** attempt)  # 1s, 2s, 4s
                continue
            raise
 
# Тест ждёт 7 секунд! (1 + 2 + 4)

Проблема: Тесты медленные, CI тормозит.

Решение: Мокируем время.

Решение #1: freezegun (простой)

Установка

pip install freezegun

Базовый пример

import pytest
from freezegun import freeze_time
from datetime import datetime, timedelta
 
@pytest.mark.asyncio
@freeze_time("2024-01-01 12:00:00")
async def test_timestamp():
    """Время заморожено"""
    now = datetime.now()
    assert now.year == 2024
    assert now.month == 1
    assert now.day == 1

Проблема с asyncio.sleep

@pytest.mark.asyncio
@freeze_time("2024-01-01 12:00:00")
async def test_sleep():
    """❌ НЕ РАБОТАЕТ с asyncio.sleep!"""
    await asyncio.sleep(60)  # Всё равно ждёт 60 секунд!
    # freezegun не влияет на asyncio.sleep

Вывод: freezegun работает с time.sleep, но НЕ работает с asyncio.sleep.

Решение #2: pytest-asyncio + mock

Мокирование asyncio.sleep

# test_retry.py
import pytest
from unittest.mock import patch, AsyncMock
import asyncio
 
async def retry_operation(operation, retries=3):
    """Retry с задержками"""
    for attempt in range(retries):
        try:
            return await operation()
        except Exception:
            if attempt < retries - 1:
                await asyncio.sleep(2 ** attempt)
                continue
            raise
 
@pytest.mark.asyncio
@patch('asyncio.sleep', new_callable=AsyncMock)
async def test_retry_fast(mock_sleep):
    """Тест без реального ожидания"""
    call_count = 0
 
    async def failing_operation():
        nonlocal call_count
        call_count += 1
        if call_count < 3:
            raise ValueError("Failed")
        return "Success"
 
    result = await retry_operation(failing_operation, retries=3)
 
    assert result == "Success"
    assert call_count == 3
    # Проверяем что sleep вызывался с правильными значениями
    assert mock_sleep.call_count == 2
    mock_sleep.assert_any_call(1)  # 2^0
    mock_sleep.assert_any_call(2)  # 2^1

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

Решение #3: Manual event loop control

Для более сложных сценариев можно вручную контролировать event loop.

Пример: таймаут операции

import pytest
import asyncio
 
async def slow_operation(delay: float):
    """Операция с задержкой"""
    await asyncio.sleep(delay)
    return "Done"
 
@pytest.mark.asyncio
async def test_timeout_fast():
    """Тестируем таймаут без ожидания"""
    with pytest.raises(asyncio.TimeoutError):
        # Используем очень маленький таймаут
        await asyncio.wait_for(slow_operation(10), timeout=0.001)

Фокус: Мы НЕ ждём 10 секунд, таймаут срабатывает через 0.001с.

Практика: тестирование retry-логики

Реальный пример: API клиент

# api_client.py
import asyncio
from typing import Optional
 
class APIClient:
    def __init__(self, base_url: str):
        self.base_url = base_url
 
    async def fetch(self, endpoint: str, retries: int = 3) -> dict:
        """Fetch с exponential backoff"""
        for attempt in range(retries):
            try:
                # Имитация HTTP запроса
                result = await self._http_get(endpoint)
                return result
            except Exception as e:
                if attempt < retries - 1:
                    delay = 2 ** attempt
                    await asyncio.sleep(delay)
                    continue
                raise
 
    async def _http_get(self, endpoint: str) -> dict:
        """Stub для реального HTTP запроса"""
        raise NotImplementedError

Тест с мокированием

# tests/test_api_client.py
import pytest
from unittest.mock import patch, AsyncMock
import asyncio
 
@pytest.mark.asyncio
@patch('asyncio.sleep', new_callable=AsyncMock)
async def test_api_retry_success(mock_sleep):
    """Успешный retry после 2 неудач"""
    client = APIClient("https://api.example.com")
 
    call_count = 0
 
    async def mock_http_get(endpoint):
        nonlocal call_count
        call_count += 1
        if call_count < 3:
            raise ConnectionError("Network error")
        return {"status": "ok"}
 
    client._http_get = mock_http_get
 
    result = await client.fetch("/users")
 
    assert result == {"status": "ok"}
    assert call_count == 3
    # Проверяем задержки
    assert mock_sleep.call_count == 2
    mock_sleep.assert_any_call(1)  # Первый retry
    mock_sleep.assert_any_call(2)  # Второй retry
 
@pytest.mark.asyncio
@patch('asyncio.sleep', new_callable=AsyncMock)
async def test_api_retry_failure(mock_sleep):
    """Все retry исчерпаны"""
    client = APIClient("https://api.example.com")
 
    async def mock_http_get(endpoint):
        raise ConnectionError("Network error")
 
    client._http_get = mock_http_get
 
    with pytest.raises(ConnectionError):
        await client.fetch("/users", retries=3)
 
    # Проверяем что было 2 retry (3 попытки = 2 задержки)
    assert mock_sleep.call_count == 2

Мокирование datetime

Проблема: тесты зависят от текущего времени

from datetime import datetime, timedelta
 
def is_expired(timestamp: datetime, ttl_minutes: int = 5) -> bool:
    """Проверка истечения TTL"""
    now = datetime.now()
    return now > timestamp + timedelta(minutes=ttl_minutes)
 
# Тест нестабильный — зависит от реального времени!

Решение: freezegun

import pytest
from freezegun import freeze_time
from datetime import datetime, timedelta
 
@freeze_time("2024-01-01 12:00:00")
def test_not_expired():
    """Не истекло"""
    timestamp = datetime.now() - timedelta(minutes=3)
    assert not is_expired(timestamp, ttl_minutes=5)
 
@freeze_time("2024-01-01 12:00:00")
def test_expired():
    """Истекло"""
    timestamp = datetime.now() - timedelta(minutes=7)
    assert is_expired(timestamp, ttl_minutes=5)

Проблема: freezegun + asyncio

@pytest.mark.asyncio
@freeze_time("2024-01-01 12:00:00")
async def test_async_time():
    """freezegun работает, но не с asyncio.sleep"""
    now = datetime.now()
    assert now.year == 2024  # ✅ Работает
 
    await asyncio.sleep(60)  # ❌ Всё равно ждёт 60 секунд!

Решение: Комбинируйте freezegun + мокирование asyncio.sleep:

@pytest.mark.asyncio
@freeze_time("2024-01-01 12:00:00")
@patch('asyncio.sleep', new_callable=AsyncMock)
async def test_combined(mock_sleep):
    """Лучшее из двух миров"""
    now = datetime.now()
    assert now.year == 2024
 
    await asyncio.sleep(60)  # Мгновенно!
    mock_sleep.assert_called_once_with(60)

Продвинутые техники

Техника #1: Контроль времени через фикстуру

# conftest.py
import pytest
from unittest.mock import patch, AsyncMock
 
@pytest.fixture
def mock_async_sleep():
    """Фикстура для мокирования asyncio.sleep"""
    with patch('asyncio.sleep', new_callable=AsyncMock) as mock:
        yield mock
 
# Использование
@pytest.mark.asyncio
async def test_with_fixture(mock_async_sleep):
    await asyncio.sleep(10)
    mock_async_sleep.assert_called_once_with(10)

Техника #2: Частичное мокирование

Иногда нужно мокировать только НЕКОТОРЫЕ вызовы asyncio.sleep:

import pytest
from unittest.mock import patch
 
@pytest.mark.asyncio
async def test_partial_mock():
    """Мокируем только длинные задержки"""
    original_sleep = asyncio.sleep
 
    async def smart_sleep(delay):
        if delay > 1:
            # Длинные задержки — пропускаем
            return
        else:
            # Короткие задержки — реальные
            await original_sleep(delay)
 
    with patch('asyncio.sleep', side_effect=smart_sleep):
        await asyncio.sleep(0.001)  # Реально ждёт 1ms
        await asyncio.sleep(60)     # Мгновенно!

Техника #3: Симуляция времени в фикстуре Redis

import pytest
from unittest.mock import AsyncMock
from freezegun import freeze_time
 
@pytest.fixture
async def redis_with_time():
    """Redis с контролем времени"""
    client = AsyncMock()
    storage = {}
 
    async def setex(key, ttl, value):
        from datetime import datetime, timedelta
        expires_at = datetime.now() + timedelta(seconds=ttl)
        storage[key] = {"value": value, "expires_at": expires_at}
 
    async def get(key):
        from datetime import datetime
        if key not in storage:
            return None
        item = storage[key]
        if datetime.now() > item["expires_at"]:
            del storage[key]
            return None
        return item["value"]
 
    client.setex = setex
    client.get = get
 
    return client
 
@pytest.mark.asyncio
@freeze_time("2024-01-01 12:00:00")
async def test_redis_ttl(redis_with_time):
    """Тестируем TTL без ожидания"""
    await redis_with_time.setex("key", 300, "value")
 
    # Проверяем что ключ есть
    result = await redis_with_time.get("key")
    assert result == "value"
 
    # Перематываем время на 6 минут
    with freeze_time("2024-01-01 12:06:00"):
        result = await redis_with_time.get("key")
        assert result is None  # ✅ Истекло!

Типичные ошибки

Ошибка #1: Забыли AsyncMock

@patch('asyncio.sleep')  # ❌ Не AsyncMock!
async def test_bad(mock_sleep):
    await asyncio.sleep(1)
    # TypeError: object MagicMock can't be used in 'await' expression

Исправление:

@patch('asyncio.sleep', new_callable=AsyncMock)  # ✅
async def test_good(mock_sleep):
    await asyncio.sleep(1)

Ошибка #2: Неправильный путь для patch

# your_module.py
import asyncio
 
async def func():
    await asyncio.sleep(1)
 
# tests/test_module.py
@patch('asyncio.sleep', new_callable=AsyncMock)  # ❌ Не сработает!
async def test():
    from your_module import func
    await func()

Проблема: Нужно патчить в модуле, где используется.

Исправление:

@patch('your_module.asyncio.sleep', new_callable=AsyncMock)  # ✅
async def test(mock_sleep):
    from your_module import func
    await func()
    mock_sleep.assert_called_once()

Ошибка #3: Реальные задержки в CI

# ❌ В CI тест ждёт 10 секунд!
async def test_slow():
    await asyncio.sleep(10)
    assert True

Исправление: Всегда мокируйте asyncio.sleep в тестах.

Альтернативы

pytest-freezegun (интеграция с pytest)

pip install pytest-freezegun

@pytest.mark.freeze_time("2024-01-01 12:00:00")
def test_frozen():
    now = datetime.now()
    assert now.year == 2024

time-machine (быстрее freezegun)

pip install time-machine

import time_machine
 
@time_machine.travel("2024-01-01 12:00:00")
def test_time():
    now = datetime.now()
    assert now.year == 2024

Когда использовать

✅ Retry-логика с задержками

✅ Таймауты и deadline

✅ TTL и expiration

✅ Rate limiting

✅ Периодические задачи

❌ Реальное тестирование производительности

❌ Стресс-тесты

❌ Бенчмарки

✅ Мокирование asyncio.sleep с AsyncMock ✅ freezegun для datetime (но не asyncio.sleep!) ✅ Тестирование retry-логики без ожидания ✅ Комбинирование техник для сложных сценариев ✅ Типичные ошибки и как их избежать

Дополнительные материалы

Следующий шаг

Если вы хотите глубже изучить async-тестирование:

Pytest: Async-тестирование и race conditions

Оглавление