Юнит-тестирование aiomysql

Рейтинг: 0Ответов: 0Опубликовано: 04.04.2023

Для основы написания теста взял ответ по связанному вопросу на Stack Overflow, там описан код для тестирования синхронного PyMySQL (на котором базируется aiomysql). Приём с AwaitableMock взял с этого ответа.

Вот что получилось:

class TestMySQL(unittest.IsolatedAsyncioTestCase):
    class AwaitableMock(unittest.mock.AsyncMock):
        def __await__(self) -> t.Iterator[t.Any]:
            self.await_count += 1
            return iter([])

    @unittest.mock.patch("<...>.aiomysql")
    async def test_x(self, mock_db):
        connection = self.AwaitableMock()
        mock_db.connect.return_value = connection

        cursor = unittest.mock.MagicMock()
        result = unittest.mock.MagicMock()

        cursor.__aenter__.return_value = result
        cursor.__aexit___ = unittest.mock.MagicMock()

        connection.cursor.return_value = cursor

У меня есть класс-враппер, в котором хранится результат aiomysql.connect(<...>) и через такую конструкцию выполняются SQL-запросы в методах:

# self.connection = await aiomysql.connect(<...>)
async with self.connection.cursor() as cursor:
    await cursor.execute(<...>)

Инициализация класса-враппера (obj = await MySQLWrapper.async_init(<...>)):

class MySQLWrapper:
    @staticmethod
    async def async_init(host, port, user, password, db):
        connection = await aiomysql.connect(
            host=host,
            port=port,
            user=user,
            password=password,
            db=db
        )

            return MySQLWrapper(connection)

    def __init__(self, connection):
        self.connection = connection

Ошибка:

Traceback (most recent call last):
  File "C:\Users\u\AppData\Local\Programs\Python\Python311\Lib\asyncio\runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\u\AppData\Local\Programs\Python\Python311\Lib\asyncio\base_events.py", line 653, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "C:\Users\u\Desktop\...\test\test_mysql_wrapper.py", line 50, in test_select
    await self.db.select("member", "id", 12345)
  File "C:\Users\u\Desktop\...\mysql_wrapper.py", line 51, in wrapper
    return await function(self, *args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\u\Desktop\...\mysql_wrapper.py", line 94, in select
    async with self.connection.cursor() as cursor:
               ^^^^^^^^^^^^^^^^^^^^^^
AttributeError: 'NoneType' object has no attribute 'cursor'

Не могу понять почему self.connection принимает None (assert_called_with который проверяет, что aiomysql.connect() запускался с нужными аргументами, отрабатывает без ошибок).

Ответы

Ответов пока нет.