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)
Hypothesis is a powerful, flexible, and easy to use library for property-based testing.

Hypothesis Hypothesis is a family of testing libraries which let you write tests parametrized by a source of examples. A Hypothesis implementation the

Hypothesis 6.4k Jan 05, 2023
Divide full port scan results and use it for targeted Nmap runs

Divide Et Impera And Scan (and also merge the scan results) DivideAndScan is used to efficiently automate port scanning routine by splitting it into 3

snovvcrash 226 Dec 30, 2022
Codeforces Test Parser for C/C++ & Python on Windows

Codeforces Test Parser for C/C++ & Python on Windows Installation Run pip instal

Minh Vu 2 Jan 05, 2022
How to Create a YouTube Bot that Increases Views using Python Programming Language

YouTube-Bot-in-Python-Selenium How to Create a YouTube Bot that Increases Views using Python Programming Language. The app is for educational purpose

Edna 14 Jan 03, 2023
A rewrite of Python's builtin doctest module (with pytest plugin integration) but without all the weirdness

The xdoctest package is a re-write of Python's builtin doctest module. It replaces the old regex-based parser with a new abstract-syntax-tree based pa

Jon Crall 174 Dec 16, 2022
Based on the selenium automatic test framework of python, the program crawls the score information of the educational administration system of a unive

whpu_spider 该程序基于python的selenium自动化测试框架,对某高校的教务系统的成绩信息实时爬取,在检测到成绩更新之后,会通过电子邮件的方式,将更新的成绩以文本的方式发送给用户,可以使得用户在不必手动登录教务系统网站时,实时获取成绩更新的信息。 该程序仅供学习交流,不可用于恶意攻

1 Dec 30, 2021
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
Simple frontend TypeScript testing utility

TSFTest Simple frontend TypeScript testing utility. Installation Install webpack in your project directory: npm install --save-dev webpack webpack-cli

2 Nov 09, 2021
d4rk Ghost is all in one hacking framework For red team Pentesting

d4rk ghost is all in one Hacking framework For red team Pentesting it contains all modules , information_gathering exploitation + vulnerability scanning + ddos attacks with 12 methods + proxy scraper

d4rk sh4d0w 15 Dec 15, 2022
Using openpyxl in Python, performed following task

Python-Automation-with-openpyxl Using openpyxl in Python, performed following tasks on an Excel Sheet containing Product Suppliers along with their pr

1 Apr 06, 2022
API mocking with Python.

apyr apyr (all lowercase) is a simple & easy to use mock API server. It's great for front-end development when your API is not ready, or when you are

Umut Seven 55 Nov 25, 2022
A set of pytest fixtures to test Flask applications

pytest-flask An extension of pytest test runner which provides a set of useful tools to simplify testing and development of the Flask extensions and a

pytest-dev 433 Dec 23, 2022
No longer maintained, please migrate to model_bakery

Model Mommy: Smart fixtures for better tests IMPORTANT: Model Mommy is no longer maintained and was replaced by Model Bakery. Please, consider migrati

Bernardo Fontes 917 Oct 04, 2022
A Simple Unit Test Matcher Library for Python 3

pychoir - Python Test Matchers for humans Super duper low cognitive overhead matching for Python developers reading or writing tests. Implemented in p

Antti Kajander 15 Sep 14, 2022
Browser reload with uvicorn

uvicorn-browser This project is inspired by autoreload. Installation pip install uvicorn-browser Usage Run uvicorn-browser --help to see all options.

Marcelo Trylesinski 64 Dec 17, 2022
Travel through time in your tests.

time-machine Travel through time in your tests. A quick example: import datetime as dt

Adam Johnson 373 Dec 27, 2022
PacketPy is an open-source solution for stress testing network devices using different testing methods

PacketPy About PacketPy is an open-source solution for stress testing network devices using different testing methods. Currently, there are only two c

4 Sep 22, 2022
Test django schema and data migrations, including migrations' order and best practices.

django-test-migrations Features Allows to test django schema and data migrations Allows to test both forward and rollback migrations Allows to test th

wemake.services 382 Dec 27, 2022