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

Overview

aioresponses

https://travis-ci.org/pnuckowski/aioresponses.svg?branch=master https://coveralls.io/repos/github/pnuckowski/aioresponses/badge.svg?branch=master Code Health Updates 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.5.3+
  • aiohttp>=2.0.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

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

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'

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

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

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.

Comments
  • TypeError: can't pickle async_generator objects

    TypeError: can't pickle async_generator objects

    My tests started fall down after I pull the new version

    Traceback (most recent call last):
      File "/Users/project/app/views.py", line 659, in _post
        responses = await asyncio.gather(*tasks)
      File "/Users/project/utils/ncapi/client.py", line 314, in upload_verification_document
        return await self._action("uploadVerificationDocument", "post", data, files=True, single_file=True)
      File "/Users/project/utils/ncapi/client.py", line 181, in _action
        res = await request(self.url(action_name), data=result)
      File "/Users/env-2fy-KEPT/lib/python3.7/site-packages/aioresponses/core.py", line 338, in _request_mock
        self.requests[key].append(RequestCall(args, copy.deepcopy(kwargs)))
      File "/Users/env-2fy-KEPT/lib/python3.7/copy.py", line 150, in deepcopy
        y = copier(x, memo)
      File "/Users/env-2fy-KEPT/lib/python3.7/copy.py", line 240, in _deepcopy_dict
        y[deepcopy(key, memo)] = deepcopy(value, memo)
      File "/Users/env-2fy-KEPT/lib/python3.7/copy.py", line 180, in deepcopy
        y = _reconstruct(x, memo, *rv)
      File "/Users/env-2fy-KEPT/lib/python3.7/copy.py", line 280, in _reconstruct
        state = deepcopy(state, memo)
      File "/Users/env-2fy-KEPT/lib/python3.7/copy.py", line 150, in deepcopy
        y = copier(x, memo)
      File "/Users/env-2fy-KEPT/lib/python3.7/copy.py", line 240, in _deepcopy_dict
        y[deepcopy(key, memo)] = deepcopy(value, memo)
      File "/Users/env-2fy-KEPT/lib/python3.7/copy.py", line 150, in deepcopy
        y = copier(x, memo)
      File "/Users/env-2fy-KEPT/lib/python3.7/copy.py", line 215, in _deepcopy_list
        append(deepcopy(a, memo))
      File "/Users/env-2fy-KEPT/lib/python3.7/copy.py", line 150, in deepcopy
        y = copier(x, memo)
      File "/Users/env-2fy-KEPT/lib/python3.7/copy.py", line 220, in _deepcopy_tuple
        y = [deepcopy(a, memo) for a in x]
      File "/Users/env-2fy-KEPT/lib/python3.7/copy.py", line 220, in <listcomp>
        y = [deepcopy(a, memo) for a in x]
      File "/Users/env-2fy-KEPT/lib/python3.7/copy.py", line 169, in deepcopy
        rv = reductor(4)
    TypeError: can't pickle async_generator objects
    Traceback (most recent call last):
      File "/Users/project/app/views.py", line 659, in _post
        responses = await asyncio.gather(*tasks)
    TypeError: can't pickle async_generator objects
    
    opened by darland 9
  • Add callbacks to provide dynamic responses

    Add callbacks to provide dynamic responses

    Callback's arguments are URL object and aiohttp.ClientRequest **kwargs.

    Callback should return CallbackResult object. If callback returns None then default response will be built automatically (useful for just checking aiohttp.ClientRequest **kwargs).

    Closes #15

    feature 
    opened by decaz 9
  • Upgrade to aiohttp v3 series and drop support for v1

    Upgrade to aiohttp v3 series and drop support for v1

    • This PR is for #88.

    • aiohttp v3 uses native async/await syntax and drops support for Python 3.5.2 or lower.

    • I think it would be reasonable enough to support latest two major versions of aiohttp: v2 and v3 series. (Of course the decision is up to the main author of this project!)

    opened by achimnol 6
  • Mock timeout?

    Mock timeout?

    Thanks for the awesome library. Installed, used it, loved it. Combined with asynctest it's very pleasant.

    Just in case you're interested with feature ideas, I think mocking request timeouts would fall in the scope. In order to test stuff like that:

    # Code example
    async with session.get(url, headers=headers, timeout=timeout) as response:
        return await response.json()
    

    or

    with async_timeout.timeout(timeout):
        async with session.get(url, headers=headers, timeout=None) as response:
            return await response.json()
    

    It could be by specifying a delay parameter to aioresponses.add(...), that would then rely on await asyncio.sleep() for example.

    What do you think?

    feature 
    opened by leplatrem 6
  • Don't print aiohttp version

    Don't print aiohttp version

    I'd suggest that we don't print the aiohttp version (or perhaps move it behind an env variable flag or something). I feel that it is best practice to not pollute stdout when running tests.

    opened by alukach 5
  • Fix compatibility with aiohttp==3.7.0

    Fix compatibility with aiohttp==3.7.0

    The new version of aiohttp has made limit a required argument for the StreamReader class. This change adds an explicit limit of 2 ** 16 which is the same as what aiohttp uses internally.

    Fixes #173

    opened by davidwtbuxton 4
  • Prevent saved requests to be modified

    Prevent saved requests to be modified

    As kwargs are stored as is, they store references that might be updated, so the stored request might contains something else than the request that should have been sent.

    Storing a deep copy of kwargs solves this.

    I updated an already existing test case but if you'd rather want me to write a new one I can as well.

    Also, would it be possible to plan a release on pypi?

    Thanks again

    opened by Colin-b 4
  • Support coroutine callbacks

    Support coroutine callbacks

    Great testing library! :+1:

    I have a need for coroutine response callbacks, so that I can control their execution by waiting on other events. As far as I could tell, this is a fairly simple change, so here's a PR implementing (and testing) it.

    opened by alanbriolat 4
  • ResponseHandler - missing 1 required positional argument: 'loop'

    ResponseHandler - missing 1 required positional argument: 'loop'

    Hi there. After an upgrade of aiohttp to version 3.5.0, I'm getting the following exception when running tests using aioresponses 0.5.0:

    /usr/local/lib/python3.6/asyncio/coroutines.py:110: in __next__
        return self.gen.send(None)
    /usr/local/lib/python3.6/site-packages/aioresponses/core.py:256: in _request_mock
        response = await self.match(method, url)
    /usr/local/lib/python3.6/asyncio/coroutines.py:110: in __next__
        return self.gen.send(None)
    /usr/local/lib/python3.6/site-packages/aioresponses/core.py:232: in match
        response = await matcher.build_response(url)
    /usr/local/lib/python3.6/asyncio/coroutines.py:110: in __next__
        return self.gen.send(None)
    /usr/local/lib/python3.6/site-packages/aioresponses/core.py:105: in build_response
        resp.content = stream_reader_factory()
    _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
    
        def stream_reader_factory():
    >       protocol = ResponseHandler()
    E       TypeError: __init__() missing 1 required positional argument: 'loop'
    
    /usr/local/lib/python3.6/site-packages/aioresponses/compat.py:23: TypeError
    

    Looks like in aiottp 3.5.0, with this change https://github.com/aio-libs/aiohttp/pull/3372 , the loop parameter is now required.

    Thanks

    bug 
    opened by thanosexcite 4
  • Support repeated executions of mocked requests

    Support repeated executions of mocked requests

    Currently, aioresponses does not support repeated execution of the same mocked request which means you have to dupe the mocked request in your code:

    with aioresponses() as mocked:
        mocked.get('http://auth')
        # some code that hits the above url
        mocked.get('http://auth')
        # some code that hits the above url for a second time
    

    Fix is to add a repeat flag which allows the mock to be repeated:

    with aioresponses() as mocked:
        mocked.get('http://auth', repeat=True)
        # some code that hits the above url
        # some code that hits the above url for a second time
    

    I believe requests-mock does this by default so i'm not sure if we should make this the default in aioresponses too?

    opened by leetreveil 4
  • Allow unregistered url to hit actual server 2

    Allow unregistered url to hit actual server 2

    Hi guys!

    passthrough doesn't work for me

    class SerpTopTestCase(AioHTTPTestCase):
    
        async def get_application(self):
            app = create_app(loop=self.loop)
            return app
    
        @unittest_run_loop
        async def test_serp_top(self):
            with aioresponses(passthrough=['https://api.vertifire.com']) as mocked:
                mocked.get(
                    VERTIFIRE_SERP_TOP_API_URL, status=200, payload=dict(data=[]))
                request = await self.client.request("GET", "/serp_top/")
                assert request.status == 200
                response_data = await request.json()
                assert "data" in response_data
    
    Traceback (most recent call last):
      File "/usr/local/lib/python3.5/dist-packages/aiohttp/test_utils.py", line 415, in new_func
        return self.loop.run_until_complete(func(self))
      File "/usr/lib/python3.5/asyncio/base_events.py", line 387, in run_until_complete
        return future.result()
      File "/usr/lib/python3.5/asyncio/futures.py", line 274, in result
        raise self._exception
      File "/usr/lib/python3.5/asyncio/tasks.py", line 239, in _step
        result = coro.send(None)
      File "/usr/share/marketmuse/marketmuse/tests/functional/serp_top.py", line 165, in test_serp_top
        request = await self.client.request("GET", "/serp_top/")
      File "/usr/local/lib/python3.5/dist-packages/aiohttp/test_utils.py", line 253, in request
        method, self.make_url(path), *args, **kwargs
      File "/usr/local/lib/python3.5/dist-packages/aiohttp/client.py", line 616, in __iter__
        resp = yield from self._coro
      File "/usr/lib/python3.5/asyncio/coroutines.py", line 206, in coro
        res = func(*args, **kw)
      File "/usr/local/lib/python3.5/dist-packages/aioresponses/core.py", line 170, in _request_mock
        if url.startswith(prefix):
    AttributeError: 'URL' object has no attribute 'startswith'
    

    what am I doing wrong?

    ubuntu 14.04 aiohttp==2.0.7 aioresponses==0.1.4

    opened by polosatyi 4
  • mocking ignores base_url

    mocking ignores base_url

    aiohttp.ClientSession takes a base_url (https://docs.aiohttp.org/en/stable/client_reference.html?highlight=base_url#aiohttp.ClientSession) that is then used to build the URL with .request() and other methods.

    Sample usage:

    In [3]: session = aiohttp.ClientSession("http://httpbin.org/")
    
    In [4]: await session.get("/get")
    Out[4]: 
    <ClientResponse(http://httpbin.org/get) [200 OK]>
    […]
    

    Unfortunately, aioresponses ignores that, so it's not possible to mock the full URL, because aioresponses would only pass /get to its matchers.

    opened by hydrargyrum 0
  • Assert called #133 incomplete

    Assert called #133 incomplete

    Assert called #133 (core.py line 389 method = method.upper()) assumes that all methods are upper case, but that is not the case as it works just as well to input lowercase methods as well.

    opened by Kane610 0
  • 0.7.4 pypi sdist includes what looks like py.typed from a local venv

    0.7.4 pypi sdist includes what looks like py.typed from a local venv

    $ tar -tf /tmp/dist/aioresponses-0.7.4.tar.gz | grep '\.env'
    aioresponses-0.7.4/.env/
    aioresponses-0.7.4/.env/lib/
    aioresponses-0.7.4/.env/lib/python3.9/
    aioresponses-0.7.4/.env/lib/python3.9/site-packages/
    aioresponses-0.7.4/.env/lib/python3.9/site-packages/_pytest/
    aioresponses-0.7.4/.env/lib/python3.9/site-packages/_pytest/py.typed
    aioresponses-0.7.4/.env/lib/python3.9/site-packages/aiohttp/
    aioresponses-0.7.4/.env/lib/python3.9/site-packages/aiohttp/py.typed
    aioresponses-0.7.4/.env/lib/python3.9/site-packages/aiosignal/
    aioresponses-0.7.4/.env/lib/python3.9/site-packages/aiosignal/py.typed
    [...]
    

    Seems to be accidental.

    opened by mgorny 0
  • ModuleNotFoundError: No module named 'pkg_resources' (Python 3.11)

    ModuleNotFoundError: No module named 'pkg_resources' (Python 3.11)

    Starting in Python 3.8, importlib is recommended over pkg_resources: https://docs.python.org/3/library/importlib.metadata.html?highlight=pkg_resources

    pkg_resources is no longer included in some distributions of Python 3.10+ including CircleCI Python images.

    opened by jacebrowning 0
  • Support python 3.10 and aiohttp 3.8

    Support python 3.10 and aiohttp 3.8

    Remove remnants of aiohttp 2.x and Python 3.5 support

    Commit adfb65038caebce3d62 removed support for aiohttp 2.x and Python 3.5 but there are still some remnants of these left. We get rid of them. This also removes one mypy warning:

    compat.py:31: error: All conditional function variants must have identical signatures [misc]

    Add type annotation for stream_reader_factory

    Remove aiohttp 2.x from tox.ini

    We no longer run tests against aiohttp 2.x so we don't need it in tox.ini.

    Add Python 3.10 support

    aioresponses works with Python 3.10 with no issues so we update the package metadata to say that, and we update the list of test environments to include Python 3.10.

    Add tests against aiohttp 3.8

    aioresponses works with aiohttp 3.8 so we extend the list of test environments to include this version.

    Add Python 3.10 tests to GitHub workflows

    Add tests against aiohttp 3.8 to GitHub workflows

    Drop Python 3.6 support

    We cannot run tests with Python 3.10 using Pytest 6.x due to [1]. This is fixed in Pytest 7.x but Pytest 7.x no longer supports Python 3.6 which aioresponses tries to support. However because Python 3.6 is already past the EOL date, the simplest solution is to no longer test with Python 3.6 and to stop declaring support for that version.

    [1] https://github.com/pytest-dev/pytest/issues/8546

    Update Pytest to the latest version to fix tests on Python 3.10

    Our tests are filing on Python 3.10 due to [1] so we update Pytest in requirements-dev.txt to the latest version.

    [1] https://github.com/pytest-dev/pytest/issues/8546

    Don't test with aiohttp 3.6 and older on Python 3.10

    aiohttp 3.6 and older use Python features which were removed in Python 3.10:

                assert port is not None
    >           hosts = await asyncio.shield(self._resolve_host(
                    host,
                    port,
                    traces=traces), loop=self._loop)
    E               TypeError: shield() got an unexpected keyword argument 'loop'
    
    .tox/py3.10-aiohttp35/lib/python3.10/site-packages/aiohttp/connector.py:952: TypeError
    

    To avoid test failures, we remove unsupported combinations from GitHub workflows.

    Drop support for aiohttp 3.2.x and older

    aiohttp 3.2.x and older don't work on Python 3.7 and because we now require Python 3.7+, we can drop support for aiohttp 3.0, 3.1, and 3.2.

    opened by marcinsulikowski 6
  • Mocking a Slow API

    Mocking a Slow API

    I'm wondering if there's a way of mocking an API which takes a long time to respond. I currently have issues with slow APIs and want to be able to test that my requests wait long enough for them. I also want to make sure they retry if the slow API fails.

    Is there any way of doing this with the aioresponses library?

    opened by Enprogames 1
Releases(0.7.4)
Python program that uses pynput to simulate key presses. Probably only works on Windows.

AutoKey Python program that uses pynput to simulate key presses. Probably only works on Windows. Can be used for pretty much whatever you want except

2 Oct 28, 2022
Code for "SUGAR: Subgraph Neural Network with Reinforcement Pooling and Self-Supervised Mutual Information Mechanism"

SUGAR Code for "SUGAR: Subgraph Neural Network with Reinforcement Pooling and Self-Supervised Mutual Information Mechanism" Overview train.py: the cor

41 Nov 08, 2022
A pure Python script to easily get a reverse shell

easy-shell A pure Python script to easily get a reverse shell. How it works? After sending a request, it generates a payload with different commands a

Cristian Souza 48 Dec 12, 2022
A cross-platform GUI automation Python module for human beings. Used to programmatically control the mouse & keyboard.

PyAutoGUI PyAutoGUI is a cross-platform GUI automation Python module for human beings. Used to programmatically control the mouse & keyboard. pip inst

Al Sweigart 7.5k Dec 31, 2022
Test python asyncio-based code with ease.

aiounittest Info The aiounittest is a helper library to ease of your pain (and boilerplate), when writing a test of the asynchronous code (asyncio). Y

Krzysztof Warunek 55 Oct 30, 2022
πŸƒπŸ’¨ For when you need to fill out feedback in the last minute.

BMSCE Auto Feedback For when you need to fill out feedback in the last minute. πŸƒ πŸ’¨ Setup Clone the repository Run pip install selenium Set the RATIN

Shaan Subbaiah 10 May 23, 2022
Aioresponses is a helper for mock/fake web requests in python aiohttp package.

aioresponses Aioresponses is a helper to mock/fake web requests in python aiohttp package. For requests module there are a lot of packages that help u

402 Jan 06, 2023
Python scripts for a generic performance testing infrastructure using Locust.

TODOs Reference to published paper or online version of it loadtest_plotter.py: Cleanup and reading data from files ARS_simulation.py: Cleanup, docume

Juri Tomak 3 Dec 15, 2022
A pytest plugin to skip `@pytest.mark.slow` tests by default.

pytest-skip-slow A pytest plugin to skip @pytest.mark.slow tests by default. Include the slow tests with --slow. Installation $ pip install pytest-ski

Brian Okken 19 Jan 04, 2023
Descriptor Vector Exchange

Descriptor Vector Exchange This repo provides code for learning dense landmarks without supervision. Our approach is described in the ICCV 2019 paper

James Thewlis 74 Nov 29, 2022
BDD library for the py.test runner

BDD library for the py.test runner pytest-bdd implements a subset of the Gherkin language to enable automating project requirements testing and to fac

pytest-dev 1.1k Jan 09, 2023
Ab testing - The using AB test to test of difference of conversion rate

Facebook recently introduced a new type of offer that is an alternative to the current type of bidding called maximum bidding he introduced average bidding.

5 Nov 21, 2022
Minimal example of getting Django + PyTest running on GitHub Actions

Minimal Django + Pytest + GitHub Actions example This minimal example shows you how you can runs pytest on your Django app on every commit using GitHu

Matt Segal 5 Sep 19, 2022
A pytest plugin, that enables you to test your code that relies on a running PostgreSQL Database

This is a pytest plugin, that enables you to test your code that relies on a running PostgreSQL Database. It allows you to specify fixtures for PostgreSQL process and client.

Clearcode 252 Dec 21, 2022
MongoDB panel for the Flask Debug Toolbar

Flask Debug Toolbar MongoDB Panel Info: An extension panel for Rob Hudson's Django Debug Toolbar that adds MongoDB debugging information Author: Harry

Cenk AltΔ± 4 Dec 11, 2019
Yet another python home automation project. Because a smart light is more than just on or off

Automate home Yet another home automation project because a smart light is more than just on or off. Overview When talking about home automation there

Maja Massarini 62 Oct 10, 2022
A simple python script that uses selenium(chrome web driver),pyautogui,time and schedule modules to enter google meets automatically

A simple python script that uses selenium(chrome web driver),pyautogui,time and schedule modules to enter google meets automatically

3 Feb 07, 2022
Fills out the container extension form automatically. (Specific to IIT Ropar)

automated_container_extension Fills out the container extension form automatically. (Specific to IIT Ropar) Download the chrome driver from the websit

Abhishek Singh Sambyal 1 Dec 24, 2021
Mock smart contracts for writing Ethereum test suites

Mock smart contracts for writing Ethereum test suites This package contains comm

Trading Strategy 222 Jan 04, 2023
Automated Penetration Testing Framework

Automated Penetration Testing Framework

OWASP 2.1k Jan 01, 2023