How to mock aiohttp.client.ClientSession.get async context manager

Question

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 ?

score 35 · Accepted Answer · answered Feb 13 '18 at 09:04

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.

And how do you patch it to have access to the `resp.status`? Because inside of the context I do have `if 200 <= resp.status <= 300` and it complains because comparison operators are not support between `MagicMock` and `int`. — Hugo Sousa, Feb 22 '19 at 14:36

score 7 · Answer 2 · answered Dec 29 '21 at 13:49

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'

score 6 · Answer 3 · answered Apr 05 '22 at 14:35

You dont need to install any framework to test aiohttp.ClientSession

Check it out:

# Module A

import aiohttp


async def send_request():
    async with aiohttp.ClientSession() as session:
        async with session.post("https://example.com", json={"testing": True}) as response:
            if response.status != 200:
                print("Woops")
                return False

            print("YAY!")
            return True


# Module A test

import pytest
from types import SimpleNamespace
from unittest.mock import MagicMock, patch

# Do not forget to replace `path_to_module_a` with the module that imports aiohttp
import path_to_module_a.send_request

@patch("path_to_module_a.aiohttp.ClientSession")
@pytest.mark.asyncio
async def test_should_fail_to_send_request(mock: MagicMock):
    session = MagicMock()
    session.post.return_value.__aenter__.return_value = SimpleNamespace(status=500)

    mock.return_value.__aenter__.return_value = session

    response = await send_request()

    assert response == False
    assert session.post.call_args.kwargs["json"] == {"testing": True}

This is a nice solution, but how can I test the case when send_request() method calls 'request.raise_for_status()' ? — mehekek, Jun 12 '23 at 08:12

score 3 · Answer 4 · answered Jun 18 '20 at 15:47

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

Very useful and simple! Solved my issue. – Suyog Shimpi Aug 09 '21 at 15:11 — Suyog Shimpi, Aug 09 '21 at 15:11

How to mock aiohttp.client.ClientSession.get async context manager

4 Answers4

Linked