使用上下文管理器测试异步协程

Question

我一直在尝试通过 Python 3.7 中的 aiobotocore 测试上下文管理的异步协程。我一直在使用 asynctest 包来获取包含的 MagicMock，它具有 __aenter__ 和 __aexit__ 的魔法方法，以及一个自定义模拟工厂，return 是一个 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 来处理上下文管理器，但是我需要我为 return 协程制作的自定义模拟。我知道有一个带有 asynctest 的 CoroutineMock 但我无法让它在这种情况下工作，我该如何解决这个问题？

Answer 1

所以从其他答案看来我需要用 CoroutineMock 模拟特定的 delete_object 方法，而我让它工作的问题是 aiobotocore 使用 AioSession 作为 class 获取会话，以下代码有效：

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))

使用上下文管理器测试异步协程

Testing Async Coroutines With Context Manager

mocking

python-unittest

python-asyncio

python-3.7

asynctest