Как писать тесты для асинхронных функций с использованием мок-объектов в `pytest`?

Для тестирования асинхронных функций с моками в pytest:

  1. Используйте pytest-asyncio для поддержки асинхронных тестов.
  2. Применяйте unittest.mock или pytest-mock для создания мок-объектов.
  3. Обязательно сделайте тестовую функцию асинхронной (async def test_...).
  4. Мокайте асинхронные функции/методы используя async_return_value для возврата значений и async_side_effect для вызова исключений или других асинхронных функций.
  5. Используйте 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")
  

Пояснения:

  1. @pytest.mark.asyncio: Эта декоратор сообщает pytest-asyncio, что функция является асинхронной и должна быть запущена в асинхронном контексте.
  2. 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").
  3. mock_fetch_data.return_value = {"result": "mocked_data"}: Устанавливает значение, которое должен возвращать мок-объект при вызове.
  4. result = await process_data("dummy_url"): Вызываем тестируемую функцию process_data, которая теперь будет использовать замокированную версию fetch_data_from_api.
  5. assert result == "mocked_data": Проверяем, что функция process_data вернула ожидаемое значение, основанное на данных, возвращенных моком.
  6. 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 вызовы асинхронных функций в тестах.

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

0