Как работать с асинхронными функциями при тестировании?

Тестирование асинхронных функций

Асинхронный код в Python (asyncio, aiohttp) требует специального подхода к тестированию. Нельзя просто вызвать async функцию как обычную — нужен event loop.

Основные способы тестирования async кода

1. pytest-asyncio для pytest

Самый популярный способ — использовать pytest-asyncio:

pip install pytest-asyncio

import pytest
import asyncio

# Основной способ: @pytest.mark.asyncio
@pytest.mark.asyncio
async def test_async_function():
    result = await some_async_function()
    assert result == expected_value

2. Простые async функции

import asyncio
from typing import Coroutine

# Функция которую нужно тестировать
async def fetch_user(user_id: int) -> dict:
    await asyncio.sleep(0.1)  # Имитация запроса
    return {"id": user_id, "name": "Alice"}

# Способ 1: с @pytest.mark.asyncio
@pytest.mark.asyncio
async def test_fetch_user():
    user = await fetch_user(1)
    assert user["name"] == "Alice"
    assert user["id"] == 1

# Способ 2: ручное создание loop (старый способ)
def test_fetch_user_manual():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        user = loop.run_until_complete(fetch_user(1))
        assert user["name"] == "Alice"
    finally:
        loop.close()

# Способ 3: asyncio.run() (Python 3.7+)
def test_fetch_user_run():
    user = asyncio.run(fetch_user(1))
    assert user["name"] == "Alice"

Тестирование HTTP запросов с aiohttp

Использование aioresponses (mock)

pip install aioresponses pytest-asyncio aiohttp

import aiohttp
import pytest
from aioresponses import aioresponses

# Функция которую тестируем
async def get_user_from_api(user_id: int) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.example.com/users/{user_id}") as resp:
            return await resp.json()

# Тест с мокированием
@pytest.mark.asyncio
async def test_get_user_from_api():
    # Мокируем HTTP ответ
    with aioresponses() as mocked:
        mocked.get(
            "https://api.example.com/users/1",
            payload={"id": 1, "name": "Alice", "email": "alice@example.com"},
            status=200
        )
        
        result = await get_user_from_api(1)
        assert result["name"] == "Alice"
        assert result["email"] == "alice@example.com"

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

@pytest.mark.asyncio
async def test_api_error_handling():
    with aioresponses() as mocked:
        # Мокируем ошибку 404
        mocked.get(
            "https://api.example.com/users/999",
            status=404,
            payload={"error": "Not found"}
        )
        
        # Ожидаем исключение
        with pytest.raises(Exception):
            await get_user_from_api(999)

Параллельное выполнение (asyncio.gather)

Тестирование параллельных задач

# Функция с параллельными запросами
async def fetch_multiple_users(user_ids: list[int]) -> list[dict]:
    tasks = [fetch_user(uid) for uid in user_ids]
    return await asyncio.gather(*tasks)

# Тест
@pytest.mark.asyncio
async def test_fetch_multiple_users():
    with aioresponses() as mocked:
        for i in range(1, 4):
            mocked.get(
                f"https://api.example.com/users/{i}",
                payload={"id": i, "name": f"User{i}"}
            )
        
        users = await fetch_multiple_users([1, 2, 3])
        assert len(users) == 3
        assert users[0]["name"] == "User1"

Таймауты и отмена задач

asyncio.wait_for

@pytest.mark.asyncio
async def test_timeout():
    # Функция может зависнуть
    async def slow_function():
        await asyncio.sleep(10)
        return "done"
    
    # Ограничиваем время выполнения
    with pytest.raises(asyncio.TimeoutError):
        await asyncio.wait_for(slow_function(), timeout=1)

@pytest.mark.asyncio
async def test_cancellation():
    async def long_task():
        try:
            await asyncio.sleep(10)
        except asyncio.CancelledError:
            return "cancelled"
    
    task = asyncio.create_task(long_task())
    await asyncio.sleep(0.1)
    task.cancel()
    
    result = await task
    assert result == "cancelled"

Fixture для async тестов

# conftest.py
import pytest
import asyncio
from typing import AsyncGenerator

# Fixture для базы данных
@pytest.fixture
async def async_db() -> AsyncGenerator:
    """Инициализация БД для теста"""
    db = await connect_to_db()
    await db.execute("CREATE TABLE users (id INT, name TEXT)")
    
    yield db
    
    await db.execute("DROP TABLE users")
    await db.close()

# Использование fixture
@pytest.mark.asyncio
async def test_with_db(async_db):
    await async_db.execute("INSERT INTO users VALUES (1, 'Alice')")
    result = await async_db.fetch("SELECT * FROM users")
    assert len(result) == 1

Тестирование WebSocket

import websockets
import pytest
from unittest.mock import AsyncMock, patch

# Функция которую тестируем
async def send_ws_message(uri: str, message: str):
    async with websockets.connect(uri) as websocket:
        await websocket.send(message)
        response = await websocket.recv()
        return response

# Тест с мокированием WebSocket
@pytest.mark.asyncio
async def test_websocket():
    mock_ws = AsyncMock()
    mock_ws.__aenter__.return_value = mock_ws
    mock_ws.__aexit__.return_value = None
    mock_ws.recv.return_value = "pong"
    
    with patch('websockets.connect', return_value=mock_ws):
        result = await send_ws_message("ws://localhost:8000", "ping")
        assert result == "pong"
        mock_ws.send.assert_called_with("ping")

Использование unittest.mock для async

from unittest.mock import AsyncMock, patch
import asyncio

# Функция с зависимостями
async def process_user(user_id: int, user_service):
    user = await user_service.get_user(user_id)
    return {"processed": user["name"].upper()}

# Мокируем async метод
@pytest.mark.asyncio
async def test_with_mock():
    mock_service = AsyncMock()
    mock_service.get_user = AsyncMock(return_value={"id": 1, "name": "alice"})
    
    result = await process_user(1, mock_service)
    assert result["processed"] == "ALICE"
    mock_service.get_user.assert_called_once_with(1)

Тестирование контекстных менеджеров

class AsyncDatabaseConnection:
    async def __aenter__(self):
        await asyncio.sleep(0.1)
        self.connected = True
        return self
    
    async def __aexit__(self, exc_type, exc, tb):
        self.connected = False
    
    async def query(self, sql):
        if not self.connected:
            raise RuntimeError("Not connected")
        return [{"result": "data"}]

# Тест
@pytest.mark.asyncio
async def test_async_context_manager():
    async with AsyncDatabaseConnection() as db:
        result = await db.query("SELECT * FROM users")
        assert result[0]["result"] == "data"

Конфигурация pytest.ini

[pytest]
# Автоматически добавляем asyncio loop для всех async тестов
asyncio_mode = auto

# Вместо @pytest.mark.asyncio можно просто написать async def
# но с автоматическим loop

Лучшие практики

✅ Делай так:

# 1. Используй pytest-asyncio
@pytest.mark.asyncio
async def test_async():
    ...

# 2. Мокируй внешние API
with aioresponses() as mocked:
    mocked.get(...)
    result = await function()

# 3. Используй AsyncMock для внутренних функций
mock = AsyncMock(return_value=data)

# 4. Тестируй таймауты
async with pytest.raises(asyncio.TimeoutError):
    await asyncio.wait_for(func(), timeout=1)

# 5. Используй fixture для инициализации
@pytest.fixture
async def async_client():
    ...

❌ Не делай так:

# 1. Не используй asyncio.run() в async тестах
def test_async():  # синхронный тест
    result = asyncio.run(async_func())
    # Плохо! Используй @pytest.mark.asyncio

# 2. Не забывай await
async def test():
    task = fetch_user(1)  # ОШИБКА! нет await

# 3. Не мешай sync и async код
# Если используешь async, всё должно быть async

# 4. Не блокируй event loop
async def test():
    time.sleep(1)  # ПЛОХО! используй await asyncio.sleep(1)

Заключение

Тестирование async кода — это специализированный навык. Ключевые инструменты:

• pytest-asyncio — для запуска async тестов
• aioresponses — для мокирования HTTP
• AsyncMock — для мокирования внутренних функций
• asyncio.wait_for — для таймаутов
• pytest fixture — для инициализации

Помни: async код требует особого внимания к таймаутам, отмене задач и управлению ошибками!