Как издеваться над сопрограммой json() при использовании aiohttp.ClientSession.get

Question

Я хочу издеваться над сопрограммой json() из метода aiohttp.ClientSession.get. Похоже, он возвращает объект асинхронного генератора, поэтому я не совсем понимаю, как в моем примере издеваться. Вот мой код:

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 для обучения, и меньше полагаться на сторонние библиотеки.

Answer 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'

Answer 2

Это часть 1, но я предлагаю вам посмотреть часть2.

Я не уверен, что полностью понимаю ваш вопрос, потому что async def или же @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())