Тестируем асинхронный код: asyncio и pytest-asyncio в деле

Тестируем асинхронный код: asyncio и pytest-asyncio в деле

Асинхронное программирование в Python — это мощный инструмент для работы с задачами, которые требуют ожидания, такими как сетевые запросы, операции с файлами и другие долгие процессы. Однако когда дело доходит до тестирования асинхронного кода, многие сталкиваются с трудностями. Как правильно тестировать функции с async def? Как работать с event loop? И как мокать асинхронные вызовы?

В этой статье мы разберемся, как тестировать асинхронный код с помощью библиотеки pytest-asyncio, а также коснемся тонкостей работы с таймаутами, ошибками и мокированием асинхронных вызовов.

Зачем нам асинхронное программирование?

Для начала давайте вспомним, что такое асинхронный код. Когда программа выполняет длительные операции, такие как запросы к базе данных или HTTP-запросы, асинхронный подход позволяет не блокировать основной поток программы. Вместо этого выполнение продолжается, пока операция не завершится.

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

import asyncio

async def fetch_data(url):
print(f"Fetching data from {url}...")
await asyncio.sleep(2) # Имитируем задержку в 2 секунды
return f"Data from {url}"

Здесь fetch_data — асинхронная функция, которая имитирует задержку с помощью await asyncio.sleep().

Теперь, когда мы понимаем, что такое асинхронный код, давайте перейдем к тому, как его тестировать!

Подготовка окружения

Для того чтобы начать тестировать асинхронный код, нам понадобится несколько вещей:

1. pytest — популярный фреймворк для тестирования в Python.
2. pytest-asyncio — плагин для pytest, который позволяет работать с асинхронными функциями и event loop.

Чтобы установить необходимые библиотеки, используйте pip:

pip install pytest pytest-asyncio

1. Тестируем асинхронные функции

Основная проблема, с которой мы сталкиваемся при тестировании асинхронных функций, — это необходимость работать с event loop. Но с помощью pytest-asyncio все становится гораздо проще. Давайте разберемся на примере.

Пример 1: Тестирование асинхронной функции без моков

Предположим, у нас есть асинхронная функция, которая выполняет задержку, а затем возвращает данные:

# async_func.py
import asyncio

async def fetch_data(url):
await asyncio.sleep(2) # Имитируем сетевой запрос
return f"Data from {url}"

Теперь напишем тест, который проверяет эту функцию:

# test_async_func.py
import pytest
from async_func import fetch_data

@pytest.mark.asyncio # Декоратор, который сообщает pytest, что это асинхронный тест
async def test_fetch_data():
result = await fetch_data("http://example.com")
assert result == "Data from http://example.com"

Комментарий к коду:

• Мы используем декоратор @pytest.mark.asyncio, который сообщает pytest, что этот тест асинхронный.
• В тесте мы вызываем await fetch_data(), чтобы дождаться завершения асинхронной функции.
• Проверяем, что результат соответствует ожидаемому значению.

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

2. Мокирование асинхронных вызовов

Допустим, что вместо того, чтобы реальный запрос уходит на сервер, нам нужно замокать этот вызов. Мокирование асинхронных функций можно делать с помощью unittest.mock и pytest-asyncio.

Пример 2: Мокируем асинхронные вызовы

Предположим, что функция fetch_data делает HTTP-запрос. Мы хотим замокать этот запрос, чтобы тесты не зависели от реального сервера.

import pytest
from unittest.mock import AsyncMock
from async_func import fetch_data

@pytest.mark.asyncio
async def test_fetch_data_with_mock():
mock_fetch = AsyncMock(return_value="Mocked data")

# Подставляем наш мок в функцию
result = await mock_fetch("http://example.com")

assert result == "Mocked data"
mock_fetch.assert_awaited_once_with("http://example.com")

Комментарий к коду:

• Мы создаем мок AsyncMock, который является асинхронной версией обычного Mock.
• AsyncMock(return_value="Mocked data") указывает, что при вызове мок-функции она вернет строку "Mocked data".
• Используем assert_awaited_once_with(), чтобы убедиться, что мок был вызван с нужным параметром.

В этом примере мы замещаем реальную асинхронную функцию мок-версией, чтобы тестировать логику без сетевых запросов.

3. Тестирование таймаутов и ошибок в асинхронном коде

Асинхронные функции часто связаны с тайм-аутами, ошибками и исключениями, поэтому важно тестировать, как они обрабатывают такие ситуации.

Пример 3: Тестируем таймауты

Предположим, у нас есть асинхронная функция, которая делает запрос с таймаутом:

import asyncio

async def fetch_data_with_timeout(url):
try:
await asyncio.wait_for(asyncio.sleep(3), timeout=2) # Тайм-аут 2 секунды
return f"Data from {url}"
except asyncio.TimeoutError:
return "Request timed out"

Теперь протестируем тайм-аут:

# test_async_timeout.py
import pytest
from async_func import fetch_data_with_timeout

@pytest.mark.asyncio
async def test_fetch_data_with_timeout():
result = await fetch_data_with_timeout("http://example.com")
assert result == "Request timed out"

Комментарий к коду:

• Мы тестируем, что если время ожидания превышает тайм-аут (2 секунды), то возникает ошибка и возвращается строка "Request timed out".
• Важно убедиться, что наш код правильно обрабатывает исключение и возвращает нужный результат.

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

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

import asyncio

async def fetch_data_with_error(url):
if url == "http://example.com":
raise ValueError("Invalid URL")
return f"Data from {url}"

Тестируем это поведение:

# test_async_error.py
import pytest
from async_func import fetch_data_with_error

@pytest.mark.asyncio
async def test_fetch_data_with_error():
with pytest.raises(ValueError):
await fetch_data_with_error("http://example.com")

Комментарий к коду:

• Мы используем pytest.raises для того, чтобы проверить, что при попытке выполнить функцию с неправильным URL будет выброшено исключение ValueError.

4. Интеграционное тестирование с асинхронным кодом

Когда мы тестируем асинхронные функции, важно понимать, как они взаимодействуют с другими компонентами системы, такими как базы данных, файлы или веб-сервисы.

Пример 5: Интеграционное тестирование с мокированием асинхронных вызовов

Предположим, что у нас есть функция, которая получает данные из базы данных и затем выполняет их обработку:

import asyncio

async def fetch_from_db(db_connection, query):
result = await db_connection.execute(query)
return result

Мы можем протестировать её, замокав асинхронный запрос:

# test_async_integration.py
import pytest
from unittest.mock import AsyncMock
from async_func import fetch_from_db

@pytest.mark.asyncio
async def test_fetch_from_db():
mock_db_connection = AsyncMock()
mock_db_connection.execute.return_value = "Mocked data"

result = await fetch_from_db(mock_db_connection, "SELECT * FROM table")

assert result == "Mocked data"
mock_db_connection.execute.assert_awaited_once_with("SELECT * FROM table")

Комментарий к коду:

• Мы создаем мок для соединения с базой данных и мокаем метод execute, чтобы вернуть данные.
• Тест проверяет, что функция вернет корректные данные и правильно взаимодействует с мокированным объектом.

Заключение

Тестирование асинхронного кода может быть непростой задачей, но с библиотеками, такими как pytest-asyncio, этот процесс становится гораздо проще. Мы разобрали основные подходы, включая:

• Как тестировать асинхронные функции с помощью pytest-asyncio.
• Как мокировать асинхронные вызовы.
• Как тестировать ошибки, тайм-ауты и асинхронные исключения.
• Как выполнять интеграционные тесты с асинхронными вызовами.

С правильными инструментами и подходами тестирование асинхронного кода станет гораздо удобнее и эффективнее.