Для тестирования асинхронных функций с моками в pytest:
pytest-asyncio для поддержки асинхронных тестов.unittest.mock или pytest-mock для создания мок-объектов.async def test_...).async_return_value для возврата значений и async_side_effect для вызова исключений или других асинхронных функций.await для вызова мокированных асинхронных функций внутри теста.Пример:
import asyncio
import pytest
from unittest.mock import AsyncMock
@pytest.mark.asyncio
async def test_my_async_function(mocker):
mock_func = AsyncMock(return_value=123)
mocker.patch("your_module.your_async_func", new_callable=lambda: mock_func)
result = await your_module.your_async_function()
assert result == 123
mock_func.assert_called_once()
Написание тестов для асинхронных функций с использованием мок-объектов в pytest требует комбинации нескольких инструментов и техник. Основная сложность заключается в правильной обработке корутин и обеспечении их выполнения в контексте теста.
Основные инструменты и библиотеки:
pytest: Основной фреймворк для запуска и организации тестов.pytest-asyncio: Плагин для pytest, который обеспечивает поддержку асинхронных тестов. Он позволяет использовать async def для определения тестовых функций и фикстур.asyncio: Стандартная библиотека Python для работы с асинхронным кодом.unittest.mock или pytest-mock: Для создания мок-объектов и управления их поведением. pytest-mock предоставляет удобный fixture mocker, который упрощает создание и управление моками.Пример:
Предположим, у нас есть асинхронная функция, которая взаимодействует с внешним сервисом (например, делает HTTP-запрос):
import asyncio
import aiohttp
async def fetch_data_from_api(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.json()
async def process_data(url):
data = await fetch_data_from_api(url)
# Производим какую-то обработку данных
return data['result']
Теперь напишем тест для функции process_data, замокировав fetch_data_from_api:
import pytest
import asyncio
from unittest.mock import AsyncMock
from your_module import process_data, fetch_data_from_api # Замените your_module на имя вашего модуля
@pytest.mark.asyncio
async def test_process_data(mocker):
# 1. Создаем мок-объект для fetch_data_from_api
mock_fetch_data = mocker.patch("your_module.fetch_data_from_api", new_callable=AsyncMock) # Замените your_module на имя вашего модуля
# 2. Определяем, какое значение должен возвращать мок
mock_fetch_data.return_value = {"result": "mocked_data"}
# 3. Вызываем тестируемую функцию
result = await process_data("dummy_url")
# 4. Проверяем результат
assert result == "mocked_data"
# 5. (Опционально) Проверяем, что мок был вызван с ожидаемыми аргументами
mock_fetch_data.assert_called_once_with("dummy_url")
Пояснения:
@pytest.mark.asyncio: Эта декоратор сообщает pytest-asyncio, что функция является асинхронной и должна быть запущена в асинхронном контексте.mocker.patch("your_module.fetch_data_from_api", new_callable=AsyncMock): Этот код заменяет реальную функцию fetch_data_from_api в модуле your_module на мок-объект mock_fetch_data. new_callable=AsyncMock указывает, что создаваемый мок должен быть асинхронным. Без этого указания, pytest может не правильно интерпретировать результат вызова мока. Важно указать правильный путь к функции ("your_module.fetch_data_from_api").mock_fetch_data.return_value = {"result": "mocked_data"}: Устанавливает значение, которое должен возвращать мок-объект при вызове.result = await process_data("dummy_url"): Вызываем тестируемую функцию process_data, которая теперь будет использовать замокированную версию fetch_data_from_api.assert result == "mocked_data": Проверяем, что функция process_data вернула ожидаемое значение, основанное на данных, возвращенных моком.mock_fetch_data.assert_called_once_with("dummy_url"): (Опционально) Проверяем, что мок-объект был вызван с ожидаемыми аргументами. Это полезно для уверенности в том, что функция вызывала внешний сервис с правильными параметрами.Альтернативные способы создания AsyncMock:
Начиная с Python 3.8, можно использовать `unittest.mock.AsyncMock` напрямую:
from unittest.mock import AsyncMock
async def my_async_function():
pass
mock = AsyncMock(side_effect=my_async_function)
В более ранних версиях Python или при использовании pytest-mock, можно использовать `mocker.patch("path.to.function", new_callable=AsyncMock)`.
Ключевые моменты:
pytest-asyncio.@pytest.mark.asyncio для обозначения асинхронных тестов.AsyncMock или new_callable=AsyncMock в mocker.patch.await вызовы асинхронных функций в тестах.Этот подход позволяет изолировать тестируемую функцию от внешних зависимостей, упростить тестирование и сделать его более надежным.