我将添加此信息，因为我很难弄清楚如何模拟异步 api 调用。
这是我为模拟异步调用所做的。
这是我想测试的功能
async def get_user_info(headers, payload):
return await httpx.AsyncClient().post(URI, json=payload, headers=headers)
你仍然需要 MockResponse 类
class MockResponse:
def __init__(self, json_data, status_code):
self.json_data = json_data
self.status_code = status_code
def json(self):
return self.json_data
您添加 MockResponseAsync 类
class MockResponseAsync:
def __init__(self, json_data, status_code):
self.response = MockResponse(json_data, status_code)
async def getResponse(self):
return self.response
这是测试。这里重要的是我之前创建了响应，因为init函数不能是异步的，并且对 getResponse 的调用是异步的，所以它全部被检查出来。
@pytest.mark.asyncio
@patch('httpx.AsyncClient')
async def test_get_user_info_valid(self, mock_post):
"""test_get_user_info_valid"""
# Given
token_bd = "abc"
username = "bob"
payload = {
'USERNAME': username,
'DBNAME': 'TEST'
}
headers = {
'Authorization': 'Bearer ' + token_bd,
'Content-Type': 'application/json'
}
async_response = MockResponseAsync("", 200)
mock_post.return_value.post.return_value = async_response.getResponse()
# When
await api_bd.get_user_info(headers, payload)
# Then
mock_post.return_value.post.assert_called_once_with(
URI, json=payload, headers=headers)
如果你有更好的方法告诉我，但我认为这样很干净。