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)
Mypy static type checker plugin for Pytest

pytest-mypy Mypy static type checker plugin for pytest Features Runs the mypy static type checker on your source files as part of your pytest test run

Dan Bader 218 Jan 03, 2023
1st Solution to QQ Browser 2021 AIAC Track 2

1st Solution to QQ Browser 2021 AIAC Track 2 This repository is the winning solution to QQ Browser 2021 AI Algorithm Competition Track 2 Automated Hyp

DAIR Lab 24 Sep 10, 2022
Python wrapper of Android uiautomator test tool.

uiautomator This module is a Python wrapper of Android uiautomator testing framework. It works on Android 4.1+ (API Level 16~30) simply with Android d

xiaocong 1.9k Dec 30, 2022
Just for testing video streaming using pytgcalls.

tgvc-video-tests Just for testing video streaming using pytgcalls. Note: The features used in this repository is highly experimental and you might not

wrench 34 Dec 27, 2022
Mixer -- Is a fixtures replacement. Supported Django, Flask, SqlAlchemy and custom python objects.

The Mixer is a helper to generate instances of Django or SQLAlchemy models. It's useful for testing and fixture replacement. Fast and convenient test-

Kirill Klenov 871 Dec 25, 2022
This is a web test framework based on python+selenium

Basic thoughts for this framework There should have a BasePage.py to be the parent page and all the page object should inherit this class BasePage.py

Cactus 2 Mar 09, 2022
Pytest plugin for testing the idempotency of a function.

pytest-idempotent Pytest plugin for testing the idempotency of a function. Usage pip install pytest-idempotent Documentation Suppose we had the follo

Tyler Yep 3 Dec 14, 2022
Network automation lab using nornir, scrapli, and containerlab with Arista EOS

nornir-scrapli-eos-lab Network automation lab using nornir, scrapli, and containerlab with Arista EOS. Objectives Deploy base configs to 4xArista devi

Vireak Ouk 13 Jul 07, 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
Python script to automatically download from Zippyshare

Zippyshare downloader and Links Extractor Python script to automatically download from Zippyshare using Selenium package and Internet Download Manager

Daksh Khurana 2 Oct 31, 2022
A Modular Penetration Testing Framework

fsociety A Modular Penetration Testing Framework Install pip install fsociety Update pip install --upgrade fsociety Usage usage: fsociety [-h] [-i] [-

fsociety-team 802 Dec 31, 2022
Statistical tests for the sequential locality of graphs

Statistical tests for the sequential locality of graphs You can assess the statistical significance of the sequential locality of an adjacency matrix

2 Nov 23, 2021
Fully functioning price detector built with selenium and python

Fully functioning price detector built with selenium and python

mark sikaundi 4 Mar 30, 2022
Pytest-typechecker - Pytest plugin to test how type checkers respond to code

pytest-typechecker this is a plugin for pytest that allows you to create tests t

vivax 2 Aug 20, 2022
WIP SAT benchmarking tooling, written with only my personal use in mind.

SAT Benchmarking Some early work in progress tooling for running benchmarks and keeping track of the results when working on SAT solvers and related t

Jannis Harder 1 Dec 26, 2021
A grab-bag of nifty pytest plugins

A goody-bag of nifty plugins for pytest OS Build Coverage Plugin Description Supported OS pytest-server-fixtures Extensible server-running framework w

Man Group 492 Jan 03, 2023
Make Selenium work on Github Actions

Make Selenium work on Github Actions Scraping with BeautifulSoup on GitHub Actions is easy-peasy. But what about Selenium?? After you jump through som

Jonathan Soma 33 Dec 27, 2022
🎓 Stepik Academy Автоматизация тестирования на Python

🎓 Stepik Academy Автоматизация тестирования на Python Запуск тестов выполняется в командной строке: pytest -v --tb=line --language=en --alluredir=all

Sergey 1 Dec 03, 2021
Automated Penetration Testing Framework

Automated Penetration Testing Framework

OWASP 2.1k Jan 01, 2023
🐍 Material for PyData Global 2021 Presentation: Effective Testing for Machine Learning Projects

Effective Testing for Machine Learning Projects Code for PyData Global 2021 Presentation by @edublancas. Slides available here. The project is develop

Eduardo Blancas 73 Nov 06, 2022