如何模拟 aiohttp.ClientSession 做出的响应？

Question

我正在使用 aiohttp 发出异步请求，我想测试我的代码。我想模拟 aiohttp.ClientSession 发送的请求。我正在寻找类似于响应处理requestslib模拟的方式。

我如何模拟aiohttp.ClientSession?

# sample method
async def get_resource(self, session):
    async with aiohttp.ClientSession() as session:
        response = await self.session.get("some-external-api.com/resource")
        if response.status == 200:
            result = await response.json()
            return result

        return {...}

# I want to do something like ...
aiohttp_responses.add(
    method='GET', 
    url="some-external-api.com/resource", 
    status=200, 
    json={"message": "this worked"}
)

async def test_get_resource(self):
    result = await get_resource()
    assert result == {"message": "this worked"}

• 我已经通读了aiohttp 测试文档。似乎他们涵盖了模拟对您的 Web 服务器的传入请求，但我不确定这是否有助于我模拟对传出请求的响应

---

编辑

我在几个项目中使用了https://github.com/pnuckowski/aioresponses，它非常适合我的需求。

Answer 1

1. 创建模拟响应

class MockResponse:
    def __init__(self, text, status):
        self._text = text
        self.status = status

    async def text(self):
        return self._text

    async def __aexit__(self, exc_type, exc, tb):
        pass

    async def __aenter__(self):
        return self

2. 使用 pytest mocker 模拟请求

@pytest.mark.asyncio
async def test_exchange_access_token(self, mocker):
    data = {}

    resp = MockResponse(json.dumps(data), 200)

    mocker.patch('aiohttp.ClientSession.post', return_value=resp)

    resp_dict = await account_api.exchange_access_token('111')

Answer 2

自从我发布这个问题以来，我已经使用这个库来模拟 aiohttp 请求：https : //github.com/pnuckowski/aioresponses，它非常适合我的需求。