Как работать с сетью?

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

Асинхронное программирование стало стандартом в Python, поэтому навык правильного тестирования async функций критичен. Я использовал несколько подходов в своей работе.

1. Основы: pytest-asyncio

pytest-asyncio — это расширение pytest для работы с async функциями:

# pip install pytest-asyncio

import pytest
import asyncio
from typing import List

# Сервис для примеров
class UserService:
    async def get_user(self, user_id: int) -> dict:
        await asyncio.sleep(0.1)  # Имитируем I/O
        return {"id": user_id, "name": f"User {user_id}"}
    
    async def get_users(self, user_ids: List[int]) -> List[dict]:
        tasks = [self.get_user(uid) for uid in user_ids]
        return await asyncio.gather(*tasks)
    
    async def create_user(self, name: str) -> dict:
        await asyncio.sleep(0.05)
        return {"id": 1, "name": name}

# Тест с @pytest.mark.asyncio
@pytest.mark.asyncio
async def test_get_user():
    service = UserService()
    user = await service.get_user(123)
    
    assert user["id"] == 123
    assert user["name"] == "User 123"

# Тест для функции возвращающей несколько значений
@pytest.mark.asyncio
async def test_get_users():
    service = UserService()
    users = await service.get_users([1, 2, 3])
    
    assert len(users) == 3
    assert users[0]["id"] == 1
    assert users[2]["name"] == "User 3"

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

[pytest]
addopts = -v
asyncio_mode = auto

2. Мокирование async функций

Основная проблема — мокирование async функций требует создания async мока:

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

class PaymentService:
    async def process_payment(self, user_id: int, amount: float) -> dict:
        # В реальности это вызывает внешний API
        pass

class OrderService:
    def __init__(self, payment_service: PaymentService):
        self.payment_service = payment_service
    
    async def create_order(self, user_id: int, amount: float) -> dict:
        # Вызываем async функцию payment service
        payment = await self.payment_service.process_payment(user_id, amount)
        return {
            "order_id": 1,
            "user_id": user_id,
            "amount": amount,
            "payment_id": payment["id"]
        }

# Правильно: используем AsyncMock
@pytest.mark.asyncio
async def test_create_order_with_async_mock():
    # Создаём async мок
    payment_service = MagicMock(spec=PaymentService)
    payment_service.process_payment = AsyncMock(return_value={"id": 999})
    
    order_service = OrderService(payment_service)
    order = await order_service.create_order(user_id=1, amount=100.0)
    
    # Проверяем результат
    assert order["order_id"] == 1
    assert order["payment_id"] == 999
    
    # Проверяем, что мок был вызван
    payment_service.process_payment.assert_called_once_with(1, 100.0)

# ❌ НЕПРАВИЛЬНО — будет ошибка!
# payment_service.process_payment = MagicMock(return_value={"id": 999})  # Это не сработает!

3. Тестирование с pytest-mock (более удобно)

import pytest
from unittest.mock import AsyncMock

@pytest.mark.asyncio
async def test_with_pytest_mock(mocker):
    """Используем mocker fixture для удобства."""
    payment_service = MagicMock(spec=PaymentService)
    
    # mocker автоматически создаёт AsyncMock
    mocker.patch.object(
        payment_service,
        'process_payment',
        new_callable=AsyncMock,
        return_value={"id": 999}
    )
    
    order_service = OrderService(payment_service)
    order = await order_service.create_order(1, 100.0)
    
    assert order["payment_id"] == 999

4. Тестирование с timeout

Для гарантии, что тесты не зависают:

@pytest.mark.asyncio
@pytest.mark.timeout(5)  # Таймаут 5 секунд
async def test_long_operation():
    service = UserService()
    result = await service.get_user(1)
    assert result is not None

# Или через asyncio
@pytest.mark.asyncio
async def test_with_asyncio_timeout():
    service = UserService()
    try:
        result = await asyncio.wait_for(
            service.get_user(1),
            timeout=1.0
        )
        assert result is not None
    except asyncio.TimeoutError:
        pytest.fail("Operation took too long")

5. Тестирование исключений в async функциях

class PaymentService:
    async def process_payment(self, amount: float) -> dict:
        if amount <= 0:
            raise ValueError("Invalid amount")
        return {"id": 1, "status": "success"}

@pytest.mark.asyncio
async def test_payment_error():
    service = PaymentService()
    
    # Способ 1: pytest.raises с async функциями
    with pytest.raises(ValueError, match="Invalid amount"):
        await service.process_payment(-10.0)

# Способ 2: в тесте OrderService
@pytest.mark.asyncio
async def test_order_creation_with_payment_error(mocker):
    payment_service = MagicMock(spec=PaymentService)
    payment_service.process_payment = AsyncMock(
        side_effect=ValueError("Payment failed")
    )
    
    order_service = OrderService(payment_service)
    
    with pytest.raises(ValueError, match="Payment failed"):
        await order_service.create_order(1, 100.0)

6. Fixtures для async

@pytest.fixture
async def payment_service():
    """Async fixture для сервиса платежей."""
    service = PaymentService()
    await service.initialize()  # Инициализация
    yield service
    await service.cleanup()  # Очистка

@pytest.mark.asyncio
async def test_with_async_fixture(payment_service):
    # payment_service уже инициализирован
    result = await payment_service.process_payment(100.0)
    assert result["status"] == "success"

7. Тестирование сложных async сценариев

class TaskQueue:
    async def process_tasks(self, tasks: List[str]) -> List[str]:
        """Обрабатывает задачи параллельно."""
        results = []
        async def process(task):
            await asyncio.sleep(0.1)
            return f"processed: {task}"
        
        results = await asyncio.gather(
            *[process(task) for task in tasks]
        )
        return results

@pytest.mark.asyncio
async def test_parallel_processing():
    queue = TaskQueue()
    tasks = ["task1", "task2", "task3"]
    
    results = await queue.process_tasks(tasks)
    
    assert len(results) == 3
    assert "processed: task1" in results

@pytest.mark.asyncio
async def test_partial_failure():
    """Тест когда одна из задач падает."""
    async def failing_task():
        await asyncio.sleep(0.01)
        raise RuntimeError("Task failed")
    
    async def normal_task():
        await asyncio.sleep(0.01)
        return "success"
    
    # gather с return_exceptions=True не бросает исключение
    results = await asyncio.gather(
        failing_task(),
        normal_task(),
        return_exceptions=True
    )
    
    assert isinstance(results[0], RuntimeError)
    assert results[1] == "success"

8. Тестирование с реальным event loop

@pytest.mark.asyncio
async def test_event_loop_behavior():
    """Проверяем поведение в одном event loop."""
    results = []
    
    async def task(n):
        results.append(f"start {n}")
        await asyncio.sleep(0.01)
        results.append(f"end {n}")
    
    # Все задачи выполняются в одном loop
    await asyncio.gather(task(1), task(2), task(3))
    
    # Проверяем порядок: start 1,2,3 перемешаны, но end-ы в том же порядке
    assert results.count("start") == 3
    assert results.count("end") == 3

9. Практический пример из моего опыта

# Реальный сценарий: тестирование микросервиса с async вызовами
from httpx import AsyncClient
from unittest.mock import AsyncMock, patch

class APIService:
    def __init__(self, client: AsyncClient):
        self.client = client
    
    async def fetch_data(self, url: str) -> dict:
        response = await self.client.get(url)
        return response.json()

@pytest.mark.asyncio
async def test_api_service_with_mock():
    """Мокируем AsyncClient для избежания реальных запросов."""
    mock_client = AsyncMock(spec=AsyncClient)
    mock_response = AsyncMock()
    mock_response.json = AsyncMock(return_value={"data": "test"})
    mock_client.get = AsyncMock(return_value=mock_response)
    
    service = APIService(mock_client)
    data = await service.fetch_data("http://api.example.com/data")
    
    assert data["data"] == "test"
    mock_client.get.assert_called_once_with("http://api.example.com/data")

Чеклист для async тестов

• Используешь @pytest.mark.asyncio для всех async тестов?
• Используешь AsyncMock а не MagicMock для async функций?
• Добавил таймауты, чтобы тесты не зависали?
• Проверил что asyncio.gather с return_exceptions=True где нужно?
• Используешь async fixtures для инициализации ресурсов?
• Тестируешь как успешные, так и ошибочные случаи?
• Проверяешь, что async функции вызываются с правильными параметрами?

Конфигурация для быстрых тестов

[pytest]
addopts = -v --tb=short
asyncio_mode = auto
python_files = test_*.py
python_classes = Test*
python_functions = test_*
markers =
    asyncio: marks tests as async

В своих проектах я стараюсь покрывать async функции на 90%+ и всегда использую мокирование для внешних сервисов.