DOCSEARCHbyAlgolia
Submit Docs

aioresponses

repository·master·Indexed 20 days ago

https://github.com/pnuckowski/aioresponses

A helper library for mocking and faking asynchronous HTTP requests made using the aiohttp package in Python. It can be used as a method decorator or context manager to intercept requests and mock components such as status codes, bodies, JSON payloads, headers, and exceptions. Supports regular expression URL matching, redirect mocking, dynamic responses via callbacks, and configurable passthrough behavior for unmatched requests.

Tokens
3.2K
Snippets
12
Records
12
Agent score
19%

What's inside aioresponses

  1. Repeat responses using the repeat argument

    master

    To test retry mechanisms, you can control how many times a mock response is used:

    • repeat=False (default) or repeat=1: The response is used once.
    • repeat=n: The response is repeated n times.
    • repeat=True: The response is repeated indefinitely.
    import asyncio
import aiohttp
from aioresponses import aioresponses

@aioresponses()
def test_multiple_responses(m):
    loop = asyncio.get_event_loop()
    session = aiohttp.ClientSession()
    m.get('http://example.com', status=500, repeat=2)
    m.get('http://example.com', status=200)  # will take effect after two preceding calls

    resp1 = loop.run_until_complete(session.get('http://example.com'))
    resp2 = loop.run_until_complete(session.get('http://example.com'))
    resp3 = loop.run_until_complete(session.get('http://example.com'))

    assert resp1.status == 500
    assert resp2.status == 500
    assert resp3.status == 200

  3. Mock HTTP requests with aioresponses

    master

    You can use aioresponses as a method decorator or as a context manager to intercept and mock aiohttp requests.

    Supported HTTP methods include: GET, POST, PUT, PATCH, DELETE, and OPTIONS.

    You can mock the following response components:

    • status: The HTTP status code.
    • body: The response body.
    • payload: A dictionary used to automatically mock a JSON response.
    • headers: A dictionary of HTTP headers.
    • exception: An exception to be raised when the request is made.
    import aiohttp
import asyncio
from aioresponses import aioresponses

@aioresponses()
def test_request(mocked):
    loop = asyncio.get_event_loop()
    mocked.get('http://example.com', status=200, body='test')
    session = aiohttp.ClientSession()
    resp = loop.run_until_complete(session.get('http://example.com'))

    assert resp.status == 200
    mocked.assert_called_once_with('http://example.com')

  4. Configure passthrough behavior

    master

    You can control whether unmatched requests are allowed to proceed to real servers:

    1. Specific servers: Use passthrough=['http://server'] to allow requests to a specific list of servers to bypass the mock.
    2. Unmatched requests: Use passthrough_unmatched=True to allow all requests that haven't been explicitly mocked to perform real network calls.
    # Passthrough specific servers
@aioresponses(passthrough=['http://backend'])
def test_passthrough(m):
    ...

# Passthrough all unmatched requests
@aioresponses(passthrough_unmatched=True)
def test_passthrough_unmatched(m):
    ...

  5. Register multiple responses for the same URL

    master

    If you register multiple mocks for the same URL, aioresponses will return them in the order they were registered.

    import asyncio
import aiohttp
from aioresponses import aioresponses

@aioresponses()
def test_multiple_responses(m):
    loop = asyncio.get_event_loop()
    session = aiohttp.ClientSession()
    m.get('http://example.com', status=500)
    m.get('http://example.com', status=200)

    resp1 = loop.run_until_complete(session.get('http://example.com'))
    resp2 = loop.run_until_complete(session.get('http://example.com'))

    assert resp1.status == 500
    assert resp2.status == 200

  6. Use aioresponses in a pytest fixture

    master

    You can wrap aioresponses in a pytest fixture to provide a mocked session to your tests automatically.

    import pytest
from aioresponses import aioresponses

@pytest.fixture
def mock_aioresponse():
    with aioresponses() as m:
        yield m

  7. Use callbacks for dynamic responses

    master

    For complex logic, you can provide a callback function. The callback must return a CallbackResult object. The callback receives the url and other keyword arguments.

    import asyncio
import aiohttp
from aioresponses import aioresponses, CallbackResult

def callback(url, **kwargs):
    return CallbackResult(status=418)

@aioresponses()
def test_callback(m):
    loop = asyncio.get_event_loop()
    session = aiohttp.ClientSession()
    m.get('http://example.com', callback=callback)

    resp = loop.run_until_complete(session.get('http://example.com'))

    assert resp.status == 418

  8. Mock HTTP headers

    master

    You can mock HTTP headers by passing a headers dictionary to the mock method. Note that while you might pass lowercase keys (e.g., connection), aiohttp may return them capitalized (e.g., Connection) because headers are handled via multidict.

    import asyncio
import aiohttp
from aioresponses import aioresponses

@aioresponses()
def test_http_headers(m):
    loop = asyncio.get_event_loop()
    session = aiohttp.ClientSession()
    m.post(
        'http://example.com',
        payload=dict(),
        headers=dict(connection='keep-alive'),
    )

    resp = loop.run_until_complete(session.post('http://example.com'))

    assert resp.headers['Connection'] == 'keep-alive'
    m.assert_called_once_with('http://example.com', method='POST')

  9. Use aioresponses as a context manager

    master

    Use aioresponses as a context manager to define mocks within a specific block of code.

    import asyncio
import aiohttp
from aioresponses import aioresponses

def test_ctx():
    loop = asyncio.get_event_loop()
    session = aiohttp.ClientSession()
    with aioresponses() as m:
        m.get('http://test.example.com', payload=dict(foo='bar'))

        resp = loop.run_until_complete(session.get('http://test.example.com'))
        data = loop.run_until_complete(resp.json())

        assert dict(foo='bar') == data
        m.assert_called_once_with('http://test.example.com')

  10. Mock redirect responses

    master

    You can mock redirects by setting the Location header and a 3xx status code. aioresponses supports both absolute and relative URLs in the Location header.

    import asyncio
import aiohttp
from aioresponses import aioresponses

@aioresponses()
def test_redirect_example(m):
    loop = asyncio.get_event_loop()
    session = aiohttp.ClientSession()

    # absolute urls are supported
    m.get(
        'http://example.com/',
        headers={'Location': 'http://another.com/'},
        status=307
    )

    resp = loop.run_until_complete(
        session.get('http://example.com/', allow_redirects=True)
    )
    assert resp.url == 'http://another.com/'

    # and also relative
    m.get(
        'http://example.com/',
        headers={'Location': '/test'},
        status=307
    )
    resp = loop.run_until_complete(
        session.get('http://example.com/', allow_redirects=True)
    )
    assert resp.url == 'http://example.com/test'

  11. Mock exceptions

    master

    Use the exception argument to simulate network or processing errors. When the mocked URL is called, aiohttp will raise the provided exception.

    import asyncio
from aiohttp import ClientSession
from aiohttp.http_exceptions import HttpProcessingError
from aioresponses import aioresponses

@aioresponses()
def test_how_to_throw_an_exception(m):
    loop = asyncio.get_event_loop()
    session = ClientSession()
    m.get('http://example.com/api', exception=HttpProcessingError('test'))

    # calling loop.run_until_complete(session.get('http://example.com/api'))
    # will throw an exception.

  12. Match URLs with regular expressions

    master

    Instead of a string, you can pass a compiled regular expression to the mock method to match a pattern of URLs.

    import asyncio
import aiohttp
import re
from aioresponses import aioresponses

@aioresponses()
def test_regexp_example(m):
    loop = asyncio.get_event_loop()
    session = aiohttp.ClientSession()
    pattern = re.compile(r'^http://example\.com/api\?foo=.*$')
    m.get(pattern, status=200)

    resp = loop.run_until_complete(session.get('http://example.com/api?foo=bar'))

    assert resp.status == 200