I have some troubles with mocking aiohttp.client.ClientSession.get context manager. I found some articles and here is one example that seems was working: article 1
So my code that I want to test:
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']
And test:
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'
When I'm running test, I'm getting an error:
(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)
So I've tried to debug - here is what I can see:
> /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__()
<generator object CoroutineMock._mock_call.<locals>.<lambda> at 0x7effad980048>
ipdb> resp.aenter
<MagicMock name='get().__aenter__().aenter' id='139636643357584'>
ipdb> resp.__aenter__().json()
*** AttributeError: 'generator' object has no attribute 'json'
ipdb> resp.__aenter__()
<generator object CoroutineMock._mock_call.<locals>.<lambda> at 0x7effad912468>
ipdb> resp.json()
<MagicMock name='get().__aenter__().json()' id='139636593767928'>
ipdb> session
<aiohttp.client.ClientSession object at 0x7effb15548d0>
ipdb> next(resp.__aenter__())
TypeError: object MagicMock can't be used in 'await' expression
So what is proper way to mock async context manager ?
See Question&Answers more detail:
os