如何Mock aiohttp.ClientSession.get异步上下文管理器返回多个不同响应

问题说明 #

已通过MagicMock实现单个aiohttp API请求的Mock，但需要让多次请求返回不同响应（如第一次返回200状态码与对应内容，第二次返回400状态码与对应内容）。直接覆盖Mock属性的方式会导致两次请求均取最后设置的值，尝试aioresponses也未成功。

单个请求Mock的有效代码 #

from unittest.mock import MagicMock
import pytest
import aiohttp

@pytest.mark.asyncio
async def test_single_request():
    mock = aiohttp.ClientSession
    mock.get = MagicMock()
    mock.get.return_value.__aenter__.return_value.status = 200
    mock.get.return_value.__aenter__.return_value.text.return_value = 'test content'

    async with aiohttp.ClientSession() as session:
        async with session.get('http://test.com') as response:
            # 注意：text()是异步方法，需要await
            assert await response.text() == 'test content'

无效的尝试代码 #

from unittest.mock import MagicMock
import pytest
import aiohttp

@pytest.mark.asyncio
async def test_failed_multiple_requests():
    mock = aiohttp.ClientSession
    mock.get = MagicMock()
    # 第一次请求设置
    mock.get.return_value.__aenter__.return_value.status = 200
    mock.get.return_value.__aenter__.return_value.text.return_value = 'test content'
    # 第二次请求设置（会覆盖第一次的配置）
    mock.get.return_value.__aenter__.return_value.status = 400
    mock.get.return_value.__aenter__.return_value.text.return_value = 'bad test content'
    
    async with aiohttp.ClientSession() as session:
        async with session.get('http://test.com') as response:
            # 这里实际会得到'bad test content'，断言失败
            assert await response.text() == 'test content'

有效解决方案 #

利用MagicMock的side_effect属性，为每次get调用生成独立的响应Mock对象，确保每次请求返回不同的结果：

from unittest.mock import MagicMock
import pytest
import aiohttp

@pytest.mark.asyncio
async def test_multiple_requests():
    # 为两次请求创建独立的响应Mock
    response1 = MagicMock()
    response1.status = 200
    response1.text.return_value = 'test content'

    response2 = MagicMock()
    response2.status = 400
    response2.text.return_value = 'bad test content'

    # 配置get方法的side_effect，按顺序返回不同的上下文管理器
    mock_get = MagicMock()
    mock_get.side_effect = [
        # 第一次调用get时，上下文返回response1
        MagicMock(__aenter__=MagicMock(return_value=response1)),
        # 第二次调用get时，上下文返回response2
        MagicMock(__aenter__=MagicMock(return_value=response2))
    ]

    # 临时替换ClientSession的get方法
    original_get = aiohttp.ClientSession.get
    aiohttp.ClientSession.get = mock_get

    try:
        async with aiohttp.ClientSession() as session:
            # 验证第一次请求
            async with session.get('http://test.com') as resp:
                assert resp.status == 200
                assert await resp.text() == 'test content'
            
            # 验证第二次请求
            async with session.get('http://test.com') as resp:
                assert resp.status == 400
                assert await resp.text() == 'bad test content'
    finally:
        # 还原原始方法，避免影响其他测试用例
        aiohttp.ClientSession.get = original_get

关键说明 #

• 独立响应Mock：为每个请求创建单独的MagicMock实例，各自设置status和异步方法text的返回值。
• side_effect控制顺序：将get的side_effect设为包含两个上下文Mock的列表，每次调用get会按顺序返回列表中的对象。
• 还原原始方法：测试结束后恢复ClientSession.get的原始实现，防止污染其他测试。
• 异步方法调用：response.text()是异步方法，必须使用await获取返回值，否则会得到协程对象导致断言失败。

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