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)
Pytest-rich - Pytest + rich integration (proof of concept)

pytest-rich Leverage rich for richer test session output. This plugin is not pub

Bruno Oliveira 170 Dec 02, 2022
pytest plugin to test mypy static type analysis

pytest-mypy-testing — Plugin to test mypy output with pytest pytest-mypy-testing provides a pytest plugin to test that mypy produces a given output. A

David Fritzsche 21 Dec 21, 2022
Kent - Fake Sentry server for local development, debugging, and integration testing

Kent is a service for debugging and integration testing Sentry.

Will Kahn-Greene 100 Dec 15, 2022
Redis fixtures and fixture factories for Pytest.

Redis fixtures and fixture factories for Pytest.This is a pytest plugin, that enables you to test your code that relies on a running Redis database. It allows you to specify additional fixtures for R

Clearcode 86 Dec 23, 2022
A configurable set of panels that display various debug information about the current request/response.

Django Debug Toolbar The Django Debug Toolbar is a configurable set of panels that display various debug information about the current request/respons

Jazzband 7.3k Jan 02, 2023
Spam the buzzer and upgrade automatically - Selenium

CookieClicker Usage: Let's check your chrome navigator version : Consequently, you have to : download the right chromedriver in the follow link : http

Iliam Amara 1 Nov 22, 2021
A test fixtures replacement for Python

factory_boy factory_boy is a fixtures replacement based on thoughtbot's factory_bot. As a fixtures replacement tool, it aims to replace static, hard t

FactoryBoy project 3k Jan 05, 2023
pytest plugin for manipulating test data directories and files

pytest-datadir pytest plugin for manipulating test data directories and files. Usage pytest-datadir will look up for a directory with the name of your

Gabriel Reis 191 Dec 21, 2022
bulk upload files to libgen.lc (Selenium script)

LibgenBulkUpload bulk upload files to http://libgen.lc/librarian.php (Selenium script) Usage ./upload.py to_upload uploaded rejects So title and autho

8 Jul 07, 2022
Parameterized testing with any Python test framework

Parameterized testing with any Python test framework Parameterized testing in Python sucks. parameterized fixes that. For everything. Parameterized te

David Wolever 714 Dec 21, 2022
Headless chrome/chromium automation library (unofficial port of puppeteer)

Pyppeteer Pyppeteer has moved to pyppeteer/pyppeteer Unofficial Python port of puppeteer JavaScript (headless) chrome/chromium browser automation libr

miyakogi 3.5k Dec 30, 2022
Voip Open Linear Testing Suite

VOLTS Voip Open Linear Tester Suite Functional tests for VoIP systems based on voip_patrol and docker 10'000 ft. view System is designed to run simple

Igor Olhovskiy 17 Dec 30, 2022
Um scraper feito em python que gera arquivos de excel baseados nas tier lists do site LoLalytics.

LoLalytics-scraper Um scraper feito em python que gera arquivos de excel baseados nas tier lists do site LoLalytics. Começando por um único script com

Kevin Souza 1 Feb 19, 2022
A simple tool to test internet stability.

pingtest Description A personal project for testing internet stability, intended for use in Linux and Windows.

chris 0 Oct 17, 2021
A small automated test structure using python to test *.cpp codes

Get Started Insert C++ Codes Add Test Code Run Test Samples Check Coverages Insert C++ Codes you can easily add c++ files in /inputs directory there i

Alireza Zahiri 2 Aug 03, 2022
Test utility for validating OpenAPI documentation

DRF OpenAPI Tester This is a test utility to validate DRF Test Responses against OpenAPI 2 and 3 schema. It has built-in support for: OpenAPI 2/3 yaml

snok 103 Dec 21, 2022
A framework-agnostic library for testing ASGI web applications

async-asgi-testclient Async ASGI TestClient is a library for testing web applications that implements ASGI specification (version 2 and 3). The motiva

122 Nov 22, 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
RAT-el is an open source penetration test tool that allows you to take control of a windows machine.

To prevent RATel from being detected by antivirus, please do not upload the payload to TOTAL VIRUS. Each month I will test myself if the payload gets detected by antivirus. So you’ll have a photo eve

218 Dec 16, 2022
Getting the most out of your hobby servo

ServoProject by Adam Bäckström Getting the most out of your hobby servo Theory The control system of a regular hobby servo looks something like this:

209 Dec 20, 2022