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

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

0 голосов
/ 29 декабря 2018

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

Ответы [ 2 ]

0 голосов
/ 29 декабря 2018

Я делал это сложнее, чем нужно.Я просто определил 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'
0 голосов
/ 29 декабря 2018

Это часть 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())

PART2

После добавлениямой ответ, я обнаружил, что есть проблема с вашим 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())
...