How to mock aiohttp.client.ClientSession.get async context manager
In your link, there is an edit:
EDIT: A GitHub issue mentioned in this post has been resolved and as of version 0.11.1 asynctest supports asynchronous context managers out of the box.
Since asynctest==0.11.1
, it was changed, a working example is:
import random
from aiohttp import ClientSession
from asynctest import CoroutineMock, patch
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']
@patch('aiohttp.ClientSession.get')
async def test_call_api_again_if_photos_not_found(mock_get):
mock_get.return_value.__aenter__.return_value.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__.return_value.json.call_count == 2
assert image_url == 'a.jpg'
The critical problem is that you need to correctly mock function json
as by default it is a MagicMock
instance. To get access to this function, you need mock_get.return_value.__aenter__.return_value.json
.
The asynctest
hasn't received any update since 2020 and one keeps getting the following deprecation notice:
python3.9/site-packages/asynctest/mock.py:434
python3.9/site-packages/asynctest/mock.py:434: DeprecationWarning: "@coroutine" decorator is deprecated since Python 3.8, use "async def" instead
def wait(self, skip=0):
Instead MagicMock can be used for mocking the coroutine as mentioned in the documentation:
Setting the spec of a Mock or MagicMock to an async function will result in a coroutine object being returned after calling.
So you could easily use the following:
from unittest.mock import MagicMock
@pytest.mark.asyncio
async def test_download():
mock = aiohttp.ClientSession
mock.get = MagicMock()
mock.get.return_value.__aenter__.return_value.status = 200
mock.get.return_value.__aenter__.return_value.text.return_value = 'test content'
async with aiohttp.ClientSession() as session:
async with session.get('http://test.com') as response:
assert response.text() == 'test content'
Building on @Sraw 's answer:
@pytest.mark.gen_test
@patch('application.adapters.http_retriever.aiohttp.ClientSession.get')
async def test_get_files(mock_get):
with open('tests/unit/sample_csv_files/Katapult_mock_response.json','r') as f:
json_dict = json.load(f)
mock_get.return_value.__aenter__.return_value.json = CoroutineMock()
mock_get.return_value.__aenter__.return_value.status = 200
mock_get.return_value.__aenter__.return_value.json.return_value = json_dict
This worked for me