如何在Python3.7中测试基于aiobotocore的上下文管理器异步协程？

我来帮你搞定这个异步测试的问题！你碰到的TypeError，本质是因为session.create_client()返回的mock没被正确配置成异步上下文管理器，导致async with块里的s3对象没法被正常await。下面一步步给你修复：

问题根源拆解 #

你的测试里，botomock（也就是get_session的mock）返回的session对象，它的create_client方法默认返回普通MagicMock。这个对象既没有实现异步上下文管理器要求的__aenter__和__aexit__协程方法，返回的s3客户端的delete_object也不是可await的协程，所以才会抛出错误。

修复步骤&完整代码 #

我们只需要用asynctest的原生mock工具，正确配置异步上下文管理器和协程方法即可，不需要自己写自定义mock类：

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

class TestMain(TestCase):
    @mock.patch('src.main.aiobotocore.get_session', new_callable=AsyncMagicMock)
    @mock.patch('src.main.Logger', new_callable=AsyncMagicMock)
    def test_delete_file(self, alogger, botomock):
        # 1. 配置Logger的info方法为可await的协程mock
        alogger.info = CoroutineMock()

        # 2. 创建s3客户端mock，把delete_object设为协程mock
        s3_client_mock = AsyncMagicMock()
        s3_client_mock.delete_object = CoroutineMock()

        # 3. 配置异步上下文管理器：__aenter__返回s3客户端，__aexit__是协程
        client_context_mock = AsyncMagicMock()
        client_context_mock.__aenter__ = CoroutineMock(return_value=s3_client_mock)
        client_context_mock.__aexit__ = CoroutineMock()

        # 4. 让session的create_client返回这个上下文管理器
        session_mock = AsyncMagicMock()
        session_mock.create_client.return_value = client_context_mock
        botomock.return_value = session_mock

        # 5. 运行测试协程
        loop = asyncio.get_event_loop()
        loop.run_until_complete(main.delete_file('test_bucket', 'test_key.csv', alogger))

        # 可选：验证所有调用是否符合预期
        alogger.info.assert_awaited_once_with('deleting file test_key.csv')
        session_mock.create_client.assert_called_once_with('s3', use_ssl=False)
        s3_client_mock.delete_object.assert_awaited_once_with(Bucket='test_bucket', Key='test_key.csv')

关键细节说明 #

• 移除自定义AsyncMockCall：asynctest.AsyncMagicMock已经原生支持异步方法调用和实例化，完全没必要自己实现__call__方法。
• 异步上下文管理器必须有__aenter__/__aexit__：这两个方法必须是CoroutineMock类型，因为它们是异步方法；__aenter__的返回值就是async with块中使用的s3客户端对象。
• 所有await的方法都要用CoroutineMock：比如alogger.info、s3.delete_object，显式设置能避免隐式转换的坑。

这样修改后，你的测试就能正常运行，不会再抛出TypeError了。

内容的提问来源于stack exchange，提问作者JSwordy