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)
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
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
Code coverage measurement for Python

Coverage.py Code coverage testing for Python. Coverage.py measures code coverage, typically during test execution. It uses the code analysis tools and

Ned Batchelder 2.3k 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
nose is nicer testing for python

On some platforms, brp-compress zips man pages without distutils knowing about it. This results in an error when building an rpm for nose. The rpm bui

1.4k Dec 12, 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
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
Show coverage stats online via coveralls.io

Coveralls for Python Test Status: Version Info: Compatibility: Misc: coveralls.io is a service for publishing your coverage stats online. This package

Kevin James 499 Dec 28, 2022
A automated browsing experience.

browser-automation This app is an automated browsing technique where one has to enter the required information, it's just like searching for Animals o

Ojas Barawal 3 Aug 04, 2021
Integration layer between Requests and Selenium for automation of web actions.

Requestium is a Python library that merges the power of Requests, Selenium, and Parsel into a single integrated tool for automatizing web actions. The

Tryolabs 1.7k Dec 27, 2022
模仿 USTC CAS 的程序,用于开发校内网站应用的本地调试。

ustc-cas-mock 模仿 USTC CAS 的程序,用于开发校内网站应用阶段调试。 请勿在生产环境部署! 只测试了最常用的三个 CAS route: /login /serviceValidate(验证 CAS ticket) /logout 没有测试过 proxy ticket。(因为我

taoky 4 Jan 27, 2022
Sixpack is a language-agnostic a/b-testing framework

Sixpack Sixpack is a framework to enable A/B testing across multiple programming languages. It does this by exposing a simple API for client libraries

1.7k Dec 24, 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
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
PyAutoEasy is a extension / wrapper around the famous PyAutoGUI, a cross-platform GUI automation tool to replace your boooring repetitive tasks.

PyAutoEasy PyAutoEasy is a extension / wrapper around the famous PyAutoGUI, a cross-platform GUI automation tool to replace your boooring repetitive t

Dingu Sagar 7 Oct 27, 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
An AWS Pentesting tool that lets you use one-liner commands to backdoor an AWS account's resources with a rogue AWS account - or share the resources with the entire internet 😈

An AWS Pentesting tool that lets you use one-liner commands to backdoor an AWS account's resources with a rogue AWS account - or share the resources with the entire internet 😈

Brandon Galbraith 276 Mar 03, 2021
Cornell record & replay mock server

Cornell: record & replay mock server Cornell makes it dead simple, via its record and replay features to perform end-to-end testing in a fast and isol

HiredScoreLabs 134 Sep 15, 2022
Scraping Bot for the Covid19 vaccination website of the Canton of Zurich, Switzerland.

Hi 👋 , I'm David A passionate developer from France. 🌱 I’m currently learning Kotlin, ReactJS and Kubernetes 👨‍💻 All of my projects are available

1 Nov 14, 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