Юнит-тестирование aiomysql
Для основы написания теста взял ответ по связанному вопросу на 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()
запускался с нужными аргументами, отрабатывает без ошибок).
Источник: Stack Overflow на русском