3

我想从 aiohttp.ClientSession.get 方法模拟 json() 协程。它看起来返回一个异步生成器对象，这是我对如何在我的示例中模拟感到困惑的地方。这是我的代码：

async def get_access_token():

async with aiohttp.ClientSession(auth=auth_credentials) as client:
    async with client.get(auth_path, params={'grant_type': 'client_credentials'}) as auth_response:
        assert auth_response.status == 200
        auth_json = await auth_response.json()
        return auth_json['access_token']

这是我模拟 get 方法的测试用例：

json_data = [{
'access_token': 'HSG9hsf328bJSWO82sl',
'expires_in': 86399, 
'token_type': 'bearer'
}]

class AsyncMock:
    async def __aenter__(self):
        return self

    async def __aexit__(self, *error_info):
        return self

@pytest.mark.asyncio
async def test_wow_api_invalid_credentials(monkeypatch, mocker):
    def mock_client_get(self, auth_path, params):
        mock_response = AsyncMock()
        mock_response.status = 200
        mock_response.json = mocker.MagicMock(return_value=json_data)
        return mock_response

    monkeypatch.setattr('wow.aiohttp.ClientSession.get', mock_client_get)
    result = await wow.get_access_token()

assert result == 'HSG9hsf328bJSWO82sl'

我认为问题可能是 mock_response.json() 不可等待。在我的示例中，我无法从非异步函数调用 await，所以我对如何做到这一点感到困惑。我想将测试库保持在最低限度，即 pytest 和 pytest-asyncio 用于学习体验，并减少对 3rd 方库的依赖。

4

2 回答 2

1

我让它变得比它需要的更复杂。我只是将 json 定义为 AsyncMock 的可等待属性，它返回 json_data。完整的代码如下所示：

json_data = {
'access_token': 'HSG9hsf328bJSWO82sl',
'expires_in': 86399, 
'token_type': 'bearer'
}

 class AsyncMock:
    async def __aenter__(self):
        return self

    async def __aexit__(self, *error_info):
        return self

    async def json(self):
        return json_data

@pytest.mark.asyncio
async def test_wow_api_invalid_credentials(monkeypatch):
    def mock_client_get(self, auth_path, params):
        mock_response = AsyncMock()
        mock_response.status = 200
        return mock_response

    monkeypatch.setattr('wow.aiohttp.ClientSession.get', mock_client_get)
    result = await wow.get_access_token()

    assert result == 'HSG9hsf328bJSWO82sl'
于 2018-12-29T17:37:02.423 回答
0

这是第1部分，但我建议你看第2部分。

我不确定我是否完全理解您的问题，因为使用async defor@asyncio.coroutine可以帮助您做到这一点。实际上，我想将其写为评论，但是有很多不同之处，我无法将其放入评论中。

import asyncio

json_ = [{
'access_token': 'HSG9hsf328bJSWO82sl',
'expires_in': 86399, 
'token_type': 'bearer'
}]


async def response_from_sun():
    return json_


class AsyncMock:

    async def specify(self):
        return self.json[0].get("access_token")

    async def __aenter__(self):
        return self

    async def __aexit__(self, *error_info):
        return self

async def mock_client_get():
    mock_response = AsyncMock()
    mock_response.status = 200
    mock_response.json = await response_from_sun()
    return mock_response

async def go():
    resp = await mock_client_get()
    result = await resp.specify()
    assert result == 'HSG9hsf328bJSWO82sl'

asyncio.get_event_loop().run_until_complete(go())

第2部分

添加我的答案后，我发现您的mock_response内容存在问题。因为mock_response不包含ClientResponse具有的变量和函数。

编辑：我尝试了很多次并观看了 ClientSession 的代码，然后我发现你可以通过它的参数指定一个新的响应类。注：connector=aiohttp.TCPConnector(verify_ssl=False)是不必要的

import asyncio
import aiohttp

class Mock(aiohttp.ClientResponse):
    print("Mock")

    async def specify(self):
        json_ = (await self.json()).get("hello")
        return json_

async def go():
    async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(verify_ssl=False),response_class=Mock) as session:
        resp = await session.get("https://www.mocky.io/v2/5185415ba171ea3a00704eed")
        result = await resp.specify()
    print(result)
    assert result == 'world'

asyncio.get_event_loop().run_until_complete(go())
于 2018-12-29T10:51:26.533 回答