Как имитировать асинхронный контекстный менеджер aiohttp.client.ClientSession.get ⇐ Python

[Anonymous]

У меня возникли некоторые проблемы с издевательством над контекстным менеджером aiohttp.client.ClientSession.get. Я нашел несколько статей, и вот один пример, который, кажется, работает: статья 1

Итак, мой код, который я хочу протестировать:

async_app.py

Код: Выделить всё

import random
from aiohttp.client import ClientSession

async def get_random_photo_url():
while True:
async with ClientSession() as session:
async with session.get('random.photos') as resp:
json = await resp.json()
photos = json['photos']
if not photos:
continue
return random.choice(photos)['img_src']

И проверьте:

test_async_app.py

Код: Выделить всё

from asynctest import CoroutineMock, MagicMock, patch

from asynctest import TestCase as TestCaseAsync

from async_app import get_random_photo_url

class AsyncContextManagerMock(MagicMock):
async def __aenter__(self):
return self.aenter

async def __aexit__(self, *args):
pass

class TestAsyncExample(TestCaseAsync):
@patch('aiohttp.client.ClientSession.get', new_callable=AsyncContextManagerMock)
async def test_call_api_again_if_photos_not_found(self, mock_get):
mock_get.return_value.aenter.json = CoroutineMock(side_effect=[{'photos': []},
{'photos': [{'img_src': 'a.jpg'}]}])

image_url = await get_random_photo_url()

assert mock_get.call_count == 2
assert mock_get.return_value.aenter.json.call_count == 2
assert image_url == 'a.jpg'

Когда я запускаю тест, я получаю сообщение об ошибке:

Код: Выделить всё

(test-0zFWLpVX) ➜  test python -m unittest test_async_app.py -v
test_call_api_again_if_photos_not_found (test_async_app.TestAsyncExample) ...  ERROR

======================================================================
ERROR: test_call_api_again_if_photos_not_found (test_async_app.TestAsyncExample)
----------------------------------------------------------------------
Traceback (most recent call last):
File "/home/kamyanskiy/.local/share/virtualenvs/test-0zFWLpVX/lib/python3.6/site-packages/asynctest/case.py", line 294, in run
self._run_test_method(testMethod)
File "/home/kamyanskiy/.local/share/virtualenvs/test-0zFWLpVX/lib/python3.6/site-packages/asynctest/case.py", line 351, in _run_test_method
self.loop.run_until_complete(result)
File "/home/kamyanskiy/.local/share/virtualenvs/test-0zFWLpVX/lib/python3.6/site-packages/asynctest/case.py", line 221, in wrapper
return method(*args, **kwargs)
File "/usr/lib/python3.6/asyncio/base_events.py", line 467, in run_until_complete
return future.result()
File "/home/kamyanskiy/.local/share/virtualenvs/test-0zFWLpVX/lib/python3.6/site-packages/asynctest/_awaitable.py", line 21, in wrapper
return await coroutine(*args, **kwargs)
File "/home/kamyanskiy/.local/share/virtualenvs/test-0zFWLpVX/lib/python3.6/site-packages/asynctest/mock.py", line 588, in __next__
return self.gen.send(None)
File "/home/kamyanskiy/work/test/test_async_app.py", line 23, in test_call_api_again_if_photos_not_found
image_url = await get_random_photo_url()
File "/home/kamyanskiy/work/test/async_app.py", line 9, in get_random_photo_url
json = await resp.json()
TypeError: object MagicMock can't be used in 'await' expression

----------------------------------------------------------------------
Ran 1 test in 0.003s

FAILED (errors=1)

Итак, я попытался отладить — вот что я вижу:

Код: Выделить всё

> /home/kamyanskiy/work/test/async_app.py(10)get_random_photo_url()
9                 import ipdb; ipdb.set_trace()
---> 10                 json = await resp.json()
11         photos = json['photos']

ipdb> resp.__aenter__()

ipdb> resp.aenter

ipdb> resp.__aenter__().json()
*** AttributeError: 'generator' object has no attribute 'json'
ipdb> resp.__aenter__()

ipdb> resp.json()

ipdb> session

ipdb> next(resp.__aenter__())
TypeError: object MagicMock can't be used in 'await' expression

Итак, как правильно имитировать асинхронный контекстный менеджер?

Подробнее здесь: https://stackoverflow.com/questions/487 ... xt-manager