How can I mock out responses made by aiohttp.ClientSession?

Question

I am using aiohttp to make asynchronous requests and I want to test my code. I want to mock out requests sent by aiohttp.ClientSession. I am looking for something similar to the way responses handles mocking for the requests lib.

How can I mock out responses made by aiohttp.ClientSession?

# sample method
async def get_resource(self, session):
    async with aiohttp.ClientSession() as session:
        response = await self.session.get("some-external-api.com/resource")
        if response.status == 200:
            result = await response.json()
            return result

        return {...}

# I want to do something like ...
aiohttp_responses.add(
    method='GET', 
    url="some-external-api.com/resource", 
    status=200, 
    json={"message": "this worked"}
)

async def test_get_resource(self):
    result = await get_resource()
    assert result == {"message": "this worked"}

• I have read through the aiohttp testing docs. It seems they cover mocking out incoming requests to your web server, but I'm not sure this helps me mock out responses to outgoing requests

---

Edit

I've used https://github.com/pnuckowski/aioresponses on a few projects and it has worked well for my needs.

Answer 1

Since I posted this question, I have used this library for mocking out aiohttp requests: https://github.com/pnuckowski/aioresponses and it has worked well for my needs.

Answer 2

1. create mockresponse

class MockResponse:
    def __init__(self, text, status):
        self._text = text
        self.status = status

    async def text(self):
        return self._text

    async def __aexit__(self, exc_type, exc, tb):
        pass

    async def __aenter__(self):
        return self

2. use pytest mocker mock the request

@pytest.mark.asyncio
async def test_exchange_access_token(self, mocker):
    data = {}

    resp = MockResponse(json.dumps(data), 200)

    mocker.patch('aiohttp.ClientSession.post', return_value=resp)

    resp_dict = await account_api.exchange_access_token('111')

Answer 3

You can use aioresponse, pytest and pytest-asyncio library.

    from aioresponses import aioresponse
    Import pytest

    # mock object here
    @pytest_asyncio.fixture
    async def mock_response():
        with aioresponses() as mocker:
            yield mocker

    # your client async here
    @pytest_asyncio.fixture
    @async def async_client():
        async with aiohttp.ClientSession as session:
            yield session

Then apply the mock response method

    @pytest.mark.fixture
    async def my_mock_client(async_client, mock_response):
        url = ("https://get-items.com/")
        response = {"key":"value"}
        mock_response.get(url, status=200, payload=response)
        response = await async_client.request("GET", url, **kwargs)
        assert response is not None

Also, aioresponse has no json parameter unlike responses.

You can use the .add in responses, but not in aioresponses.

I hope this helps.

Answer 4

1. Create a function which will return your mocked response:

async def create_resp(status_code=200, resp_data=None):
    resp = mock.AsyncMock(status_code=status_code)
    resp.json.return_value = resp_data
    return resp

2. Then use it in your test:

@pytest.mark.asyncio
@mock.patch('ms_printers.clients.menu.aiohttp.ClientSession.get')
async def test_ok(self, mock_get):
    mock_get.return_value = create_resp(resp_data={'a': 1})