Skip to main content

Mock out requests made by ClientSession from aiohttp package

Project description

https://travis-ci.org/pnuckowski/aioresponses.svg?branch=master https://coveralls.io/repos/github/pnuckowski/aioresponses/badge.svg?branch=master Code Health Updates https://img.shields.io/pypi/v/aioresponses.svg Documentation Status

Aioresponses is a helper to mock/fake web requests in python aiohttp package.

For requests module there are a lot of packages that help us with testing (eg. httpretty, responses, requests-mock).

When it comes to testing asynchronous HTTP requests it is a bit harder (at least at the beginning). The purpose of this package is to provide an easy way to test asynchronous HTTP requests.

Installing

$ pip install aioresponses

Supported versions

  • Python 3.7+

  • aiohttp>=3.3.0,<4.0.0

Usage

To mock out HTTP request use aioresponses as a method decorator or as a context manager.

Response status code, body, payload (for json response) and headers can be mocked.

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

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')

for convenience use payload argument to mock out json response. Example below.

as a context manager

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')

aioresponses allows to mock out any HTTP headers

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'))

    # note that we pass 'connection' but get 'Connection' (capitalized)
    # under the neath `multidict` is used to work with HTTP headers
    assert resp.headers['Connection'] == 'keep-alive'
    m.assert_called_once_with('http://example.com', method='POST')

allows to register different responses for the same url

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

Repeat response for the same url

E.g. for cases where you want to test retrying mechanisms.

  • By default, repeat=False means the response is not repeated (repeat=1 does the same).

  • Use repeat=n to repeat a response n times.

  • Use repeat=True to repeat a response 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

match URLs with regular expressions

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

allows to make redirects responses

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'

allows to passthrough to a specified list of servers

import asyncio
import aiohttp
from aioresponses import aioresponses

@aioresponses(passthrough=['http://backend'])
def test_passthrough(m, test_client):
    session = aiohttp.ClientSession()
    # this will actually perform a request
    resp = loop.run_until_complete(session.get('http://backend/api'))

also you can passthrough all requests except specified by mocking object

import asyncio
import aiohttp
from aioresponses import aioresponses

@aioresponses(passthrough_unmatched=True)
def test_passthrough_unmatched(m, test_client):
    url = 'https://httpbin.org/get'
    m.get(url, status=200)
    session = aiohttp.ClientSession()
    # this will actually perform a request
    resp = loop.run_until_complete(session.get('http://backend/api'))
    # this will not perform a request and resp2.status will return 200
    resp2 = loop.run_until_complete(session.get(url))

aioresponses allows to throw an 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, test_client):
    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.

aioresponses allows to use callbacks to provide dynamic responses

import asyncio
import aiohttp
from aioresponses import CallbackResult, aioresponses

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

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

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

    assert resp.status == 418

aioresponses can be used in a pytest fixture

import pytest
from aioresponses import aioresponses

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

Features

  • Easy to mock out HTTP requests made by aiohttp.ClientSession

License

  • Free software: MIT license

Credits

This package was created with Cookiecutter and the audreyr/cookiecutter-pypackage project template.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

aioresponses-0.7.8.tar.gz (40.3 kB view details)

Uploaded Source

Built Distribution

aioresponses-0.7.8-py2.py3-none-any.whl (12.5 kB view details)

Uploaded Python 2Python 3

File details

Details for the file aioresponses-0.7.8.tar.gz.

File metadata

  • Download URL: aioresponses-0.7.8.tar.gz
  • Upload date:
  • Size: 40.3 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/5.1.1 CPython/3.9.20

File hashes

Hashes for aioresponses-0.7.8.tar.gz
Algorithm Hash digest
SHA256 b861cdfe5dc58f3b8afac7b0a6973d5d7b2cb608dd0f6253d16b8ee8eaf6df11
MD5 05bd377afda3a4bf5e3dc297345672f6
BLAKE2b-256 de03532bbc645bdebcf3b6af3b25d46655259d66ce69abba7720b71ebfabbade

See more details on using hashes here.

File details

Details for the file aioresponses-0.7.8-py2.py3-none-any.whl.

File metadata

  • Download URL: aioresponses-0.7.8-py2.py3-none-any.whl
  • Upload date:
  • Size: 12.5 kB
  • Tags: Python 2, Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/5.1.1 CPython/3.9.20

File hashes

Hashes for aioresponses-0.7.8-py2.py3-none-any.whl
Algorithm Hash digest
SHA256 b73bd4400d978855e55004b23a3a84cb0f018183bcf066a85ad392800b5b9a94
MD5 ef4c72be9d24e1ca7c4513986211f62a
BLAKE2b-256 12b7584157e43c98aa89810bc2f7099e7e01c728ecf905a66cf705106009228f

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page