Пошаговый гайд по тестированию асинхронных функций на Python с использованием pytest и моков.

Часть 1: Основы тестирования и pytest

* Что такое тестирование? *

Тестирование представляет собой проверку того, работает ли ваш код, как вы предполагаете. Если все ваши тесты проходят, ваш код в порядке. Если ваши тесты не прошли, ошибка в вашем коде.

* Что такое pytest? *

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

Установка pytest и плагинов

Для начала нужно установить pytest и плагин для работы с асинхронным кодом:

```

pip install pytest pytest-asyncio

```

Базовая структура теста

Каждый тест — это обычная функция, имя которой начинается с test_. Например:

```

def test_sum():

assert 1 + 1 == 2

```

Запуск всех тестов в проекте:

```

pytest -v

```

Часть 2: Подготовка к тестированию асинхронных функций

Что нужно для тестирования асинхронного кода?

1. Асинхронный тестовый фреймворк. Мы будем использовать pytest-asyncio.
2. Умение "мокать" зависимости. Это позволит изолировать функции от внешних факторов, например базы данных.

Часть 3: Написание первого теста

Пример простой функции

Создадим небольшую асинхронную функцию:

```

async def simple_function(x: int, y: int) -> int:

return x + y

```

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

Тестируем её с помощью pytest:

```

import pytest '

@pytest.mark.asyncio

async def test_simple_function():

result = await simple_function(2, 3)

assert result == 5

```

Часть 4: Моки и фикстуры

Что такое мок?

Мок (mock) — это "поддельный" объект, который заменяет реальный объект во время теста. Например, вы можете заменить реальную базу данных мок-объектом, чтобы ускорить тестирование и избежать изменений в базе.

Установка библиотеки для моков

Python уже включает встроенный модуль unittest.mock. Вы можете использовать его для создания моков.

Часть 5: Тестирование find_all

Теперь мы начнём тестировать метод find_all. Для этого мы создадим мок для async_session_maker и проверим, вызывается ли нужный запрос.

вот сам метод:

```

from datetime import datetime

from fastapi import HTTPException,status

from sqlalchemy import delete, insert, select

from sqlalchemy.exc import SQLAlchemyError

from app.database import async_session_maker

class BaseDAO:

model = None

@classmethod

async def find_all(cls, offset: int = 0, limit: int = 10, **filter_by):

async with async_session_maker() as session:

query = (

select(cls.model)

.filter_by(**filter_by) # Используем фильтры

.offset(offset) # Добавляем пагинацию

.limit(limit)

)

result = await session.execute(query)

return result.scalars().all()

```

Теперь подробно разберем код теста:

```

@pytest.mark.asyncio

class TestBaseDAO:

@pytest.fixture(autouse=True)

def setup(self):

self.valid_uuid = str(UUID(int=1))

self.valid_uuid2 = str(UUID(int=2))

self.invalid_uuid = "invalid_uuid"

self.walletrequest_list = [

WalletRequest(

id=self.valid_uuid,

wallet_address="TE2RzoSV3wFK99w6J9UnnZ4vLfXYoxvRwP",

bandwidth = 200,

energy = 150,

trx_balance = Decimal(300.000)),

WalletRequest(

id=self.valid_uuid2,

wallet_address="TNMcQVGPzqH9ZfMCSY4PNrukevtDgp24dK",

bandwidth = 400,

energy = 550,

trx_balance = Decimal(800.000))

]

BaseDAO.model = WalletRequest

@pytest.mark.asyncio

@patch('app.dao.base.async_session_maker', new_callable=MagicMock)

async def test_find_all(self, mock_session_maker):

# Настройка моков

mock_session = AsyncMock()

mock_session_maker.return_value.__aenter__.return_value = mock_session

# Создаем моки для результата execute

mock_result = MagicMock()

mock_scalars = MagicMock()

mock_scalars.all.return_value = self.walletrequest_list

mock_result.scalars.return_value = mock_scalars

# Настраиваем execute возвращать mock_result

mock_session.execute.return_value = mock_result

# Вызываем тестируемый метод

result = await BaseDAO.find_all()

# Проверяем, что execute был вызван

mock_session.execute.assert_awaited_once()

# Проверяем результат

assert result == self.walletrequest_list

```

• Фикстура setup:Зачем нужно: В setup мы подготавливаем данные, которые будут использоваться в тестах. Это необходимо, чтобы не повторять одно и то же создание объектов в каждом тесте.
  Что мы здесь делаем: Мы создаем несколько тестовых данных, например, два объекта WalletRequest, которые будут имитировать записи из базы данных. Эти данные будут использоваться в тестах, чтобы проверить, как метод find_all обрабатывает их.
  Где используется: Эти данные (переменная walletrequest_list) используются в самом тесте, когда мы проверяем, что результат работы метода find_all совпадает с этими данными.
  
• Патчинг сессии с помощью @patch('app.dao.base.async_session_maker', new_callable=MagicMock):Зачем нужно: В реальной ситуации BaseDAO.find_all использует функцию async_session_maker для получения сессии и выполнения запроса к базе данных. Патчинг (замена) этой функции позволяет избежать реальных запросов к базе данных и заменить их на моки, с помощью которых можно контролировать поведение функции.
  Что мы здесь делаем: Патчим (замещаем) вызов async_session_maker, чтобы вместо настоящей сессии работы с базой данных возвращался мок-объект. Мы указываем, что мок-сессия будет возвращаться каждый раз, когда будет вызываться эта функция.
  Где используется: Патчинг используется в тесте, когда мы вызываем await BaseDAO.find_all(). Мок-сессия будет использоваться вместо реальной, и все вызовы внутри find_all будут выполняться с этим мок-объектом.
  
• Мокирование mock_session и других моков:Зачем нужно: Моки позволяют имитировать поведение реальных объектов. В данном случае нам нужно замокировать сессию работы с базой данных и результат выполнения запроса, чтобы проверить, правильно ли работает метод find_all без обращения к настоящей базе данных.
  Что мы здесь делаем: Мы создаем мок для сессии mock_session, который будет использоваться для имитации работы с базой данных. Затем создаем моки для результата выполнения запроса mock_result и для того, что возвращает метод scalars — это mock_scalars, который возвращает список объектов WalletRequest.
  Где используется: Эти моки используются в тесте, когда метод find_all выполняет запрос к базе данных. В результате вызова mock_session.execute возвращается мока mock_result, а через него мы получаем имитацию выполнения запроса и получения данных.
  
• Вызов тестируемого метода result = await BaseDAO.find_all():Зачем нужно: Это основной шаг теста, где мы проверяем логику работы метода find_all. Мы ожидаем, что метод выполнит запрос к базе данных (но вместо этого будет использоваться мок-сессия) и вернет правильные данные.
  Что мы здесь делаем: Здесь мы вызываем метод find_all, который должен выполнить запрос к базе данных, вернуть результаты, а затем передать их в возвращаемый список. Так как мы мокируем все взаимодействия с базой, этот вызов работает с мока-сессией.
  Где используется: В тесте мы вызываем find_all для того, чтобы проверить, правильно ли он извлекает данные из базы данных и правильно ли возвращает эти данные (например, возвращает walletrequest_list).
  
• Проверка вызова mock_session.execute.assert_awaited_once():Зачем нужно: Мы проверяем, что запрос к базе данных действительно был выполнен (в нашем случае через мок-сессию). Это необходимо для того, чтобы убедиться, что метод find_all действительно сделал попытку запросить данные из базы данных.
  Что мы здесь делаем: Мы проверяем, что метод execute был вызван ровно один раз в процессе работы теста. Это подтверждает, что запрос к базе данных был отправлен.
  Где используется: Проверка этого вызова используется внутри теста, чтобы удостовериться, что метод find_all корректно взаимодействует с мок-сессией.
  
• Проверка результата assert result == self.walletrequest_list:Зачем нужно: Мы проверяем, что метод find_all вернул правильный результат — список объектов, который мы предварительно определили как walletrequest_list.
  Что мы здесь делаем: Сравниваем результат выполнения функции с ожидаемым списком объектов WalletRequest, который был подготовлен в фикстуре.
  Где используется: Эта проверка используется для того, чтобы убедиться, что метод find_all работает правильно, и возвращает данные в ожидаемом формате.

И ещё немного про моки:

Разница между AsyncMock() и MagicMock()

• MagicMock() — это стандартный мок, который используется для синхронных объектов и функций. Это классический мок, который позволяет вам мокировать методы и атрибуты обычных (синхронных) объектов. Он создает подделку любого объекта и позволяет контролировать его поведение и следить за вызовами.
• AsyncMock() — это специализированная версия MagicMock() для работы с асинхронными функциями и объектами. Когда вы мокируете асинхронный код (например, корутины или асинхронные функции, которые используют await), вам нужно использовать AsyncMock, чтобы мокировать эти объекты корректно. Он позволяет правильно эмулировать поведение асинхронных функций, таких как ожидание завершения задачи (await).

Почему в вашем коде используется AsyncMock() и MagicMock()?

1. mock_session = AsyncMock() — Мокирование сессии базы данных (асинхронный объект):В вашем тесте сессия базы данных, которая создается через async_session_maker, является асинхронным объектом. Сессии в SQLAlchemy (и других библиотеках для работы с асинхронными базами данных) возвращаются как асинхронные объекты. Это означает, что операции с такими объектами должны использовать await для получения результата.
   Когда мы создаем мок для сессии (например, mock_session), мы используем AsyncMock, чтобы правильно имитировать асинхронное поведение — например, выполнение запроса к базе данных через await session.execute(...).
   В частности, мы мокируем вызовы, которые должны быть асинхронными (например, await session.execute(query)), и AsyncMock() позволяет нам эмулировать асинхронный код.
   
2. mock_result = MagicMock() и mock_scalars = MagicMock() — Мокирование синхронных объектов:В отличие от сессии, результат выполнения запроса (например, result и scalars()) и другие вспомогательные объекты в коде, такие как mock_scalars, не являются асинхронными. В этих случаях не нужно использовать AsyncMock, так как данные просто возвращаются синхронно, без использования await.
   MagicMock() используется для мокирования этих синхронных объектов. Например, мы ожидаем, что mock_scalars.all() вернет список, а для этого достаточно использовать MagicMock(), потому что этот вызов является обычным синхронным методом.
   

Важные моменты:

• Сессия базы данных (например, mock_session), которая взаимодействует с базой данных, должна быть мокирована с помощью AsyncMock(), потому что запросы к базе данных выполняются асинхронно (с использованием await).
• Результаты запроса (например, mock_result и mock_scalars), которые не требуют асинхронной работы, могут быть мокированы через MagicMock(), так как они не используются с await и работают синхронно.

Пример использования:

• Когда мы выполняем запрос с использованием await session.execute(query) — это асинхронный вызов, и его нужно мокировать через AsyncMock(), чтобы правильно имитировать асинхронное поведение.
• Когда мы получаем результат запроса, например, с помощью result.scalars() или mock_scalars.all() — это синхронные вызовы, и их можно замокировать с помощью MagicMock().

Пример кода с объяснением:

```

mock_session = AsyncMock() # Это асинхронный объект, с помощью которого мы мокируем сессию базы данных

mock_result = MagicMock() # Это синхронный объект, который представляет результат выполнения запроса

mock_scalars = MagicMock() # Это также синхронный объект, представляющий возвращаемое значение .scalars()
# Настроим поведение моков:

mock_scalars.all.return_value = self.walletrequest_list # Здесь all() - синхронный метод, поэтому MagicMock
# Настройка mock_result для возвращения mock_scalars mock_result.scalars.return_value = mock_scalars # scalars() возвращает mock_scalars (синхронный объект)

mock_session.execute.return_value = mock_result # execute() будет возвращать mock_result (синхронный объект)
# Когда мы вызовем метод find_all, он будет работать с асинхронными вызовами, такими как session.execute

result = await BaseDAO.find_all() # find_all вызывает execute(), который асинхронен, поэтому здесь нужен AsyncMock
# Проверка вызова execute() на мок-сессии mock_session.execute.assert_awaited_once() # Проверка, что асинхронный execute() был вызван ровно один раз

В этом примере:

• mock_session использует AsyncMock(), потому что session.execute — асинхронная операция.
• mock_result и mock_scalars используют MagicMock(), потому что их методы (scalars() и all()) синхронные.

Когда использовать AsyncMock(), а когда MagicMock()?

• Используйте AsyncMock() для объектов, методов и функций, которые работают асинхронно и требуют использования await.
• Используйте MagicMock() для объектов, методов и функций, которые работают синхронно и не используют await.