1

我一直在尝试通过 Python 3.7 中的 aiobotocore 测试上下文管理的异步协程。我一直在使用 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 来处理上下文管理器，但随后我需要我制作的自定义模拟来返回协程。我知道有一个带有 asynctest 的 CoroutineMock 但我无法让它在这种情况下工作，我将如何解决这个问题？

4

1 回答 1

1

因此，从其他答案看来，我需要使用 CoroutineMock 模拟特定的 delete_object 方法，而我让它工作的问题是 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))
于 2020-04-09T14:58:19.613 回答