Add request matching

[paulschmeida]

Correct me if I'm wrong but I don't see any way to test request body, so it's not really possible to test POST or PATCH calls with it.... Is there a workaround that I'm missing? How would I test if my functions send correct payloads?

[hydrargyrum]

I have the same problem, it does not seem possible, matching is only based on method+URL: https://github.com/pnuckowski/aioresponses/blob/master/aioresponses/core.py#L118

The sync counterpart libs can do it though, maybe it could be an inspiration for aioresponses?

• requests_mock has this: https://requests-mock.readthedocs.io/en/latest/matching.html#additional-matchers
• responses has this: https://pypi.org/project/responses/#matching-request-body-contents

[Kylmakalle]

I somehow reproduced the responses match logic with a callback. You'll need a single callback for all of the similar URLs you can't distinguish with basic match

import responses

with responses.RequestsMock() as rsps:
    rsps.add(
        responses.POST,
        'https://url/',
        json={"status": 1, "error": None},
        status=200,
        match=[responses.urlencoded_params_matcher({"verbose": "1", "data": '{"foo":"bar"}'})],
    )

aioresponses

from aioresponses import aioresponses, CallbackResult

with aioresponses() as rsps:
    def callback(url, **kwargs):
        if kwargs['data'] == {"verbose": 1, "data": '{"foo":"bar"}'}:
            pass
        elif:
            raise Exception("Unable to match the request")

    rsps.post(
        'https://url/',
        body=json.dumps({"status": 1, "error": None}),
        status=200,
        callback=callback
    )

Returning CallbackResult in the callback, you can provide your own CallbackResult implementation to add a resolved json field based on the request provided.

from aioresponses import aioresponses, CallbackResult

with aioresponses() as rsps:

    def callback(url, **kwargs):
        if kwargs['data'] == {"verbose": 1, "data": '{"foo":"bar"}'}:
            return CallbackResult(json={"status": 1, "error": None})
        elif:
            raise Exception("Unable to match the request")

    rsps.post(
        'https://url/',
        status=200,
        callback=callback
    )

[CrypticGuru]

I accomplished this by writing a custom (reusable) function which matches any kwargs given to the request. It is contained in the following script, which you can run yourself (just make sure your filename has test_ prepended, for pytest).

import aiohttp
import aioresponses
import pytest
import typing
import yarl

def _aio_match_any(
    requests: typing.Dict[
        typing.Tuple['str', yarl.URL],
        typing.List[aioresponses.core.RequestCall]
    ],
    method: typing.Literal['GET', 'HEAD', 'POST', 'PUT', 'DELETE', 'CONNECT', 'OPTIONS', 'TRACE', 'PATCH'],
    url: str,
    **kwargs
) -> None:
    """
    Match any arbitrary keyword parameters given to an `aiohttp.request`.

    Meant to be tacked-on to the `aio` fixture, but can be imported and used directly if needed.

    Args:
        requests (dict): The aioresponses.aioresponses.requests dict.
        method (str): The request HTTP method.
        url (str): The URL of the request to match.
        kwargs: Any key/value pair to match against

    Raises:
        (KeyError): If the combination of (method, url) does not exist in the provided requests.
        (AssertionError): If the provided **kwargs do not match.

    """
    matches = requests[(method.upper(), yarl.URL(url))]  # can intentionally raise a KeyError
    matched = matches[0]  # there might be others, this only works with the first request currently
    for key, value in kwargs.items():
        if (actual := matched.kwargs.get(key)) != value:
            raise AssertionError(f'Expected "{key}" to be {value}, got {actual} instead.')

@pytest.fixture
def aio():
    """Fixture for mocking aiohttp.request calls."""
    with aioresponses.aioresponses() as m:
        m.match_any = _aio_match_any  # `_aio_match_any` attached for ease-of-use within tests!
        yield m

@pytest.mark.asyncio
async def test_param(aio):
    async with aiohttp.ClientSession() as session:
        aio.post('http://test.example.com')
        await session.post('http://test.example.com', json={'expected': 'json'})
        aio.match_any(aio.requests, 'POST', 'http://test.example.com', json={'expected': 'json'})

pytest.main([__file__])

[lVlayhem]

I was able to assert request payload and headers (!) by using assert_called_with

m.assert_called_with(url, method, data={'grant_type': 'client_credentials'}, auth=BasicAuth(login='client id', password='client secret', encoding='latin1'))