Тестирование асинхронных сопрограмм с помощью диспетчера контекста

Question

Я пытался протестировать сопрограмму Async, управляемую контекстом, через aiobotocore в Python 3.7 . Я использовал пакет asynctest, чтобы получить включенный MagicMock, который имеет волшебные методы__aenter__ а также __aexit__и настраиваемая фабрика макетов, которая возвращает объект MagicMock в результате ожидаемой сопрограммы, но у меня проблемы с сопрограммой внутри диспетчера контекста. Функция, которую я пытаюсь высмеять:

from aiologger import Logger
import aiobotocore


async def delete_file(bucket, key, alogger):
    await alogger.info(f'deleting file {key}')
    session = aiobotocore.get_session()
    async with session.create_client('s3', use_ssl=False) as s3:
        await s3.delete_object(
            Bucket=bucket,
            Key=key)

Это вызывается с входными параметрами позже в коде, мой тестовый код:

import asyncio
from src import main
from unittest import TestCase, mock
from asynctest.mock import CoroutineMock, MagicMock as AsyncMagicMock

 class AsyncMockCall(mock.MagicMock):
    async def __call__(self, *args, **kwargs):
        return super().__call__(*args, **kwargs)

class TestMain(TestCase):

    @mock.patch('src.main.aiobotocore.get_session', new_callable=AsyncMagicMock)
    @mock.patch('src.main.Logger', new_callable=AsyncMockCall)
    def test_delete_file(self, alogger, botomock):
        loop = asyncio.get_event_loop()
        loop.run_until_complete(main.delete_file('test_bucket',
                                                'test_key.csv',
                                                alogger))

Но когда я запускаю его, я получаю сообщение об ошибке:

____________________________________________________________________________ TestMain.test_delete_file ____________________________________________________________________________

self = <tests.test_main.TestMain testMethod=test_delete_file>, alogger = <AsyncMockCall name='Logger' id='4480486312'>, botomock = <MagicMock name='get_session' id='4480486144'>

    @mock.patch('src.main.aiobotocore.get_session', new_callable=AsyncMagicMock)
    @mock.patch('src.main.Logger', new_callable=AsyncMockCall)
    def test_delete_file(self, alogger, botomock):
        loop = asyncio.get_event_loop()
        loop.run_until_complete(main.delete_file('test_bucket',
                                                'test_key.csv',
>                                               alogger))

tests/test_main.py:21: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/base_events.py:584: in run_until_complete
    return future.result()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

bucket = 'test_bucket', key = 'test_key.csv', alogger = <AsyncMockCall name='Logger' id='4480486312'>

    async def delete_file(bucket, key, alogger):
        await alogger.info(f'deleting file {key}')
        session = aiobotocore.get_session()
        async with session.create_client('s3', use_ssl=False) as s3:
            await s3.delete_object(
                Bucket=bucket,
>               Key=key)
E           TypeError: object MagicMock can't be used in 'await' expression

src/main.py:20: TypeError
============================================================================= short test summary info =============================================================================
FAILED tests/test_main.py::TestMain::test_delete_file - TypeError: object MagicMock can't be used in 'await' expression

Мне кажется, что мне нужен asynctest magicmock для обработки диспетчера контекста, но тогда мне нужен собственный макет, который я сделал, чтобы вернуть сопрограмму. Я знаю, что есть CoroutineMock с asynctest, но я не могу заставить его работать в этом контексте, как мне это решить?

Answer 1

Итак, из других ответов похоже, что мне нужно издеваться над конкретным методом delete_object с помощью CoroutineMock, и моя проблема с его работой заключалась в том, что aiobotocore использует AioSession в качестве класса для сеанса получения, приведенный ниже код работает:

import asyncio
from src import main
from unittest import TestCase, mock
from asynctest.mock import CoroutineMock, MagicMock as AsyncMagicMock

class AsyncMockCall(mock.MagicMock):
    async def __call__(self, *args, **kwargs):
        return super().__call__(*args, **kwargs)

class TestMain(TestCase):

    @mock.patch('src.main.aiobotocore.AioSession.create_client', new_callable=AsyncMagicMock)
    @mock.patch('src.main.Logger', new_callable=AsyncMockCall)
    def test_delete_file(self, alogger, botomock):
        loop = asyncio.get_event_loop()
        botomock.return_value.__aenter__.return_value.delete_object = CoroutineMock(return_value=[])
        loop.run_until_complete(main.delete_file('test_bucket',
                                                'test_key.csv',
                                                alogger))