Code Monkey home page Code Monkey logo

aresponses's Introduction

aresponses

image image build status Code style: black

an asyncio testing server for mocking external services

Features

  • Fast mocks using actual network connections
  • allows mocking some types of network issues
  • use regular expression matching for domain, path, method, or body
  • works with https requests as well (by switching them to http requests)
  • works with callables

Usage

Add routes and responses via the aresponses.add method:

def add(
    host_pattern=ANY, 
    path_pattern=ANY, 
    method_pattern=ANY, 
    response="", 
    *, 
    route=None, 
    body_pattern=ANY, m
    match_querystring=False, 
    repeat=1
    )

When a request is received the first matching response will be returned and removed from the routing table. The response argument can be either a string, Response, dict, or list. Use aresponses.Response when you need to do something more complex.

Note that version >=2.0 requires explicit assertions!

@pytest.mark.asyncio
async def test_simple(aresponses):
    aresponses.add("google.com", "/api/v1/", "GET", response="OK")
    aresponses.add('foo.com', '/', 'get', aresponses.Response(text='error', status=500))

    async with aiohttp.ClientSession() as session:
        async with session.get("http://google.com/api/v1/") as response:
            text = await response.text()
            assert text == "OK"
        
        async with session.get("https://foo.com") as response:
            text = await response.text()
            assert text == "error"

    aresponses.assert_plan_strictly_followed()

Assertions

In aresponses 1.x requests that didn't match a route stopped the event loop and thus forced an exception. In aresponses >2.x it's required to make assertions at the end of the test.

There are three assertions functions provided:

  • aresponses.assert_no_unused_routes Raises UnusedRouteError if all the routes defined were not used up.
  • aresponses.assert_called_in_order - Raises UnorderedRouteCallError if the routes weren't called in the order they were defined.
  • aresponses.assert_all_requests_matched - Raises NoRouteFoundError if any requests were made that didn't match to a route. It's likely but not guaranteed that your code will throw an exception in this situation before the assertion is reached.

Instead of calling these individually, it's recommended to call aresponses.assert_plan_strictly_followed() at the end of each test as it runs all three of the above assertions.

Regex and Repeat

host_pattern, path_pattern, method_pattern and body_pattern may be either strings (exact match) or regular expressions.

The repeat argument permits a route to be used multiple times.

If you want to just blanket mock a service, without concern for how many times its called you could set repeat to a large number and not call aresponses.assert_plan_strictly_followed or arespones.assert_no_unused_routes.

@pytest.mark.asyncio
async def test_regex_repetition(aresponses):
    aresponses.add(re.compile(r".*\.?google\.com"), response="OK", repeat=2)

    async with aiohttp.ClientSession() as session:
        async with session.get("http://google.com") as response:
            text = await response.text()
            assert text == "OK"

        async with session.get("http://api.google.com") as response:
            text = await response.text()
            assert text == "OK"

    aresponses.assert_plan_strictly_followed()

Json Responses

As a convenience, if a dict or list is passed to response then it will create a json response. A aiohttp.web_response.json_response object can be used for more complex situations.

@pytest.mark.asyncio
async def test_json(aresponses):
    aresponses.add("google.com", "/api/v1/", "GET", response={"status": "OK"})

    async with aiohttp.ClientSession() as session:
        async with session.get("http://google.com/api/v1/") as response:
            assert {"status": "OK"} == await response.json()

    aresponses.assert_plan_strictly_followed()

Custom Handler

Custom functions can be used for whatever other complex logic is desired. In example below the handler is set to repeat infinitely and always return 500.

import math

@pytest.mark.asyncio
async def test_handler(aresponses):
    def break_everything(request):
        return aresponses.Response(status=500, text=str(request.url))

    aresponses.add(response=break_everything, repeat=math.inf)

    async with aiohttp.ClientSession() as session:
        async with session.get("http://google.com/api/v1/") as response:
            assert response.status == 500

Passthrough

Pass aresponses.passthrough into the response argument to allow a request to bypass mocking.

    aresponses.add('httpstat.us', '/200', 'get', aresponses.passthrough)

Inspecting history

History of calls can be inspected via aresponses.history which returns the namedTuple RoutingLog(request, route, response)

@pytest.mark.asyncio
async def test_history(aresponses):
    aresponses.add(response=aresponses.Response(text="hi"), repeat=2)

    async with aiohttp.ClientSession() as session:
        async with session.get("http://foo.com/b") as response:
            await response.text()
        async with session.get("http://bar.com/a") as response:
            await response.text()

    assert len(aresponses.history) == 2
    assert aresponses.history[0].request.host == "foo.com"
    assert aresponses.history[1].request.host == "bar.com"
    assert "Route(" in repr(aresponses.history[0].route)
    aresponses.assert_plan_strictly_followed()

Context manager usage

import aiohttp
import pytest
import aresponses


@pytest.mark.asyncio
async def test_foo(event_loop):
    async with aresponses.ResponsesMockServer(loop=event_loop) as arsps:
        arsps.add('foo.com', '/', 'get', 'hi there!!')
        arsps.add(arsps.ANY, '/', 'get', arsps.Response(text='hey!'))
        
        async with aiohttp.ClientSession(loop=event_loop) as session:
            async with session.get('http://foo.com') as response:
                text = await response.text()
                assert text == 'hi'
            
            async with session.get('https://google.com') as response:
                text = await response.text()
                assert text == 'hey!'
        

working with pytest-aiohttp

If you need to use aresponses together with pytest-aiohttp, you should re-initialize main aresponses fixture with loop fixture

from aresponses import ResponsesMockServer

@pytest.fixture
async def aresponses(loop):
    async with ResponsesMockServer(loop=loop) as server:
        yield server

If you're trying to use the aiohttp_client test fixture then you'll need to mock out the aiohttp loop fixture instead:

@pytest.fixture
def loop(event_loop):
    """replace aiohttp loop fixture with pytest-asyncio fixture"""
    return event_loop

Contributing

Dev environment setup

  • install pyenv and pyenv-virtualenv - Makes it easy to install specific versions of python and switch between them. Make sure you install the virtualenv bash hook
  • git clone the repo and cd into it.
  • make init - installs proper version of python, creates the virtual environment, activates it and installs all the requirements

Submitting a feature request

  • git checkout -b my-feature-branch
  • make some cool changes
  • make autoformat
  • make test
  • make lint
  • create pull request

Updating package on pypi

  • make deploy

Changelog

3.0.0

  • fix: start using asyncio.get_running_loop() instead of event_loop per the error:
    PytestDeprecationWarning: aresponses is asynchronous and explicitly requests the "event_loop" fixture. 
    Asynchronous fixtures and test functions should use "asyncio.get_running_loop()" instead.
    
  • drop support for python 3.6
  • add comprehensive matrix testing of all supported python and aiohttp versions
  • tighten up the setup.py requirements

2.1.6

  • fix: incorrect pytest plugin entrypoint name (#72)

2.1.5

  • support asyncio_mode = strict (#68)

2.1.4

  • fix: don't assume utf8 request contents

2.1.3

  • accidental no-op release

2.1.2

  • documentation: add pypi documentation

2.1.1

  • bugfix: RecursionError when aresponses is used in more than 1000 tests (#63)

2.1.0

  • feature: add convenience method add_local_passthrough
  • bugfix: fix https subrequest mocks. support aiohttp_client compatibility

2.0.2

  • bugfix: ensure request body is available in history

2.0.0

Warning! Breaking Changes!

  • breaking change: require explicit assertions for test failures
  • feature: autocomplete works in intellij/pycharm
  • feature: can match on body of request
  • feature: store calls made
  • feature: repeated responses
  • bugfix: no longer stops event loop
  • feature: if dict or list is passed into response, a json response will be generated

1.1.2

  • make passthrough feature work with binary data

1.1.1

  • regex fix for Python 3.7.0

1.1.0

  • Added passthrough option to permit live network calls
  • Added example of using a callable as a response

1.0.0

  • Added an optional match_querystring argument that lets you match on querystring as well

Contributors

aresponses's People

Contributors

brycedrennan avatar f1ashhimself avatar felixonmars avatar gjcarneiro avatar ikhlestov avatar jessevogt avatar kache avatar marco-c avatar mattblack85 avatar nolar avatar olegt0rr avatar p4l1ly avatar scop avatar vaultvulp avatar

Stargazers

 avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar

Watchers

 avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar

aresponses's Issues

response with headers isn't matched

Hi all,

Are there any thoughts of implementing this? Or any ideas how to do it?

I'm doing this

@pytest.mark.asyncio
async def test_listing_fetch_206(aresponses):

    async def get_handler_206(request):
        xmlfile_path = Path(__file__).resolve().parent.joinpath('fetch_payload.xml')
        LOGGER.debug(f'xml file path = {xmlfile_path}')
        LOGGER.debug(f'request headers = {request.headers}')

        rng = request.http_range
        with open(xmlfile_path, 'rb') as f:
            f.seek(rng.start)
            data = f.read(rng.stop - rng.start)
            hdr = {'Content-Range': f'bytes {rng.start}-{rng.stop - 1}/{xmlfile_path.stat().st_size}'}
            #LOGGER.debug(f'resp header = {hdr}')
            resp = aresponses.Response(status=206, reason='OK', text=str(data), headers=hdr)
        return resp

    aresponses.add('foo.com', '/feed/6715', 'get', response=get_handler_206)

    # aresponses.add(aresponses.ANY, aresponses.ANY, aresponses.ANY, response=get_handler_206)


    #with isolated_filesystem():
    l = Listing('http://foo.com/feed/6715')
    await l.fetch(range_size=100)
    assert l._path.joinpath(l._filename).is_file()

I need to set the Content-Range header to have the mock serve Partial Content but when I set that on my Response object the match fails...

[2018-10-08 20:02:01,001] DEBUG:asyncio:Using selector: KqueueSelector
[2018-10-08 20:02:01,003] DEBUG:pyskyq.listing:Listing initialised: <List: url='http://foo.com/feed/6715', path='.epg_data', filename='ea51e77b9fdede19528d599f50182d37edcdbc082b06358146041fe446f6a855.xml'>
[2018-10-08 20:02:01,004] DEBUG:pyskyq.listing:Fetch(<List: url='http://foo.com/feed/6715', path='.epg_data', filename='ea51e77b9fdede19528d599f50182d37edcdbc082b06358146041fe446f6a855.xml'>) called started.
[2018-10-08 20:02:01,004] DEBUG:pyskyq.listing:Client session created. About to fetch url=http://foo.com/feed/6715
[2018-10-08 20:02:01,008] INFO:aresponses.main:Looking for match for foo.com /feed/6715 GET
[2018-10-08 20:02:01,009] DEBUG:tests.test_listing:xml file path = /Users/brad/Code/pyskyq/tests/fetch_payload.xml
[2018-10-08 20:02:01,009] DEBUG:tests.test_listing:request headers = <CIMultiDictProxy('Range': 'bytes=0-99', 'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate', 'Host': 'foo.com', 'User-Agent': 'Python/3.7 aiohttp/3.4.4', 'AResponsesIsSSL': '')>
[2018-10-08 20:02:01,011] DEBUG:tests.test_listing:resp header = {'Content-Range': 'bytes 0-99/292'}
[2018-10-08 20:02:01,012] INFO:aiohttp.access:127.0.0.1 [08/Oct/2018:19:02:01 +0000] "GET /feed/6715 HTTP/1.1" 206 322 "-" "Python/3.7 aiohttp/3.4.4"
[2018-10-08 20:02:01,013] DEBUG:pyskyq.listing:Attempting byte range download. Range = 0-100
[2018-10-08 20:02:01,013] INFO:pyskyq.listing:Server responsed with status: 206.
[2018-10-08 20:02:01,015] DEBUG:pyskyq.listing:Headers: <CIMultiDictProxy('Content-Range': 'bytes 0-99/292', 'Content-Length': '139', 'Content-Type': 'text/plain; charset=utf-8', 'Date': 'Mon, 08 Oct 2018 19:02:01 GMT', 'Server': 'Python/3.7 aiohttp/3.4.4')>
[2018-10-08 20:02:01,042] INFO:aresponses.main:Looking for match for foo.com /feed/6715 GET
[2018-10-08 20:02:01,042] ERROR:aiohttp.server:Error handling request
Traceback (most recent call last):
  File "/Users/brad/.virtualenvs/pyskyq-4vSEKDfZ/lib/python3.7/site-packages/aiohttp/web_protocol.py", line 390, in start
    resp = await self._request_handler(request)
  File "/Users/brad/.virtualenvs/pyskyq-4vSEKDfZ/lib/python3.7/site-packages/aresponses/main.py", line 61, in _handler
    return await self._find_response(request)
  File "/Users/brad/.virtualenvs/pyskyq-4vSEKDfZ/lib/python3.7/site-packages/aresponses/main.py", line 106, in _find_response
    raise self._exception  # noqa
Exception: No Match found for foo.com /feed/6715 GET.  Host Match: False  Path Match: False
[2018-10-08 20:02:01,063] INFO:aiohttp.access:127.0.0.1 [08/Oct/2018:19:02:01 +0000] "GET /feed/6715 HTTP/1.1" 500 910 "-" "Python/3.7 aiohttp/3.4.4"

Any suggestions?

Thanks

Brad

Removing urls once reached

Hi! Thanks for writing this!

I was working on integrating this into one of my first aiohttp projects and I could not figure out why the url I'd added was going away after the very first called had reached it. After a little while, I went to read the source and found https://github.com/CircleUp/aresponses/blob/master/aresponses/main.py#L93.

I can definitely see an argument for having this but I think it would be helpful to craft an error message saying why it's been removed. Or, the call to delete the route could simply be removed.

This functionality is different than every other mock utility I've used. IME the urls are always available and it is left to the caller to establish counts or use some sort of history storage mechanism to verify which paths were hit and how frequently. Overall, I think this behavior is surprising to users and can be somewhat difficult to diagnose. Also, the workaround (at least in my case) has been to add the same url in a loop so that my test can make repeated requests to the same resource.

What do you think?

Thanks again :)

No license information in package metadata

The license type is not visible in the package metadata, using pip show. This is a problem for scripts that try to gather dependency information automatically.

Other packages usually show a license either in the top-level "License:" entry, or in a Classifiers: > License :: entry.

$ pip show --verbose aresponses
Name: aresponses
Version: 2.0.0
Summary: Asyncio testing server. Similar to the responses library used for 'requests'
Home-page: https://github.com/circleup/aresponses
Author: Bryce Drennan, CircleUp
Author-email: [email protected]   
License: UNKNOWN
Location: /Users/bestave/Development/git/devx/sdk/contentsdk-python/venv/lib/python3.8/site-packages
Requires: pytest-asyncio, aiohttp
Required-by: 
Metadata-Version: 2.1
Installer: pip
Classifiers:
  Programming Language :: Python :: 3.6
  Programming Language :: Python :: 3.7
  Programming Language :: Python :: 3.8
Entry-points:
  [pytest11]
  name_of_plugin = aresponses.main

Proposed fix:

Add License :: OSI Approved :: MIT License into the Classifiers in setup.py

Test failure with 3.8

______________________________ test_raw_response _______________________________
Traceback (most recent call last):
  File "/home/travis/build/scop/aresponses/tests/test_server.py", line 131, in test_raw_response
    async with session.get("http://foo.com") as response:
  File "/home/travis/virtualenv/python3.8.0/lib/python3.8/site-packages/aiohttp/client.py", line 1012, in __aenter__
    self._resp = await self._coro
  File "/home/travis/virtualenv/python3.8.0/lib/python3.8/site-packages/aiohttp/client.py", line 504, in _request
    await resp.start(conn)
  File "/home/travis/virtualenv/python3.8.0/lib/python3.8/site-packages/aiohttp/client_reqrep.py", line 847, in start
    message, payload = await self._protocol.read()  # type: ignore  # noqa
  File "/home/travis/virtualenv/python3.8.0/lib/python3.8/site-packages/aiohttp/streams.py", line 591, in read
    await self._waiter
aiohttp.client_exceptions.ServerDisconnectedError

Support `pytest_aiohttp`

At the present pytest_aiohttp doesn't work correct with pytest_asyncio (pytest_aiohttp issue and pytest_asyncio issue ).
On the other hand aresponses doesn't require pytest_asyncio directly and can be executed without it. But aresponses depends on event_loop fixture here. So the main questions - can we use existing fixture event_loop from asyncio or loop from aiohttp on the fly for the aresponses fixture initialization?

"""Test script to show aresponses work without pytest_asyncio
Steps:
- install all required packages as `pip install pytest aiohttp pytest_aiohttp aresponses`
- run script as `pytest test_example.py -p no:asyncio
"""
import pytest
import aiohttp


# Simple server definition
async def handle(request):
    return aiohttp.web.Response(text="test response")


def create_app():
    app = aiohttp.web.Application()
    app.add_routes([aiohttp.web.get('/', handle)])
    return app


# Tests
@pytest.fixture
async def client(aiohttp_client):
    app = create_app()
    return await aiohttp_client(app)


async def test_root_url(client):
    resp = await client.get('/')
    assert resp.status == 200
    assert await resp.text() == "test response"


# this additional fixture required by aresponses
# by default aresponses depends on `event_loop` fixture defined at the pytest_asyncio
@pytest.fixture
def event_loop(loop):
    return loop


async def test_aresponses_without_aiohttp(aresponses):
    aresponses.add('foo.com', '/', 'get', 'hi there!!')
    url = 'http://foo.com'
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            text = await response.text()
            assert text == 'hi there!!'

passthrough response assumes text

    aresponses.add(
        host=re.compile(r"127\.0\.0\.1:.*"), response=aresponses.passthrough
    )

But my content is binary (msgpack), therefore I get an exception:

web_protocol.py            355 ERROR    Error handling request
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/aiohttp/web_protocol.py", line 418, in start
    resp = await task
  File "/usr/local/lib/python3.7/site-packages/aresponses/main.py", line 60, in _handler
    return await self._find_response(request)
  File "/usr/local/lib/python3.7/site-packages/aresponses/main.py", line 98, in _find_response
    return await response(request)
  File "/usr/local/lib/python3.7/site-packages/aresponses/main.py", line 125, in passthrough
    text = await r.text()
  File "/usr/local/lib/python3.7/site-packages/aiohttp/client_reqrep.py", line 1010, in text
    return self._body.decode(encoding, errors=errors)  # type: ignore
  File "/usr/local/lib/python3.7/encodings/cp1254.py", line 15, in decode
    return codecs.charmap_decode(input,errors,decoding_table)
UnicodeDecodeError: 'charmap' codec can't decode byte 0x81 in position 2033: character maps to <undefined>
web_log.py                 233 INFO     127.0.0.1 [26/Nov/2019:19:20:08 +0000] "GET /u/2 HTTP/1.1" 500 1156 "-" "Python/3.7 aiohttp/3.5.4"

The problem lies in text = await r.text(), it should not assume the response is text.

Allow introspecting captured requests and responses

I want to inspect the headers of the requests. With the responses module I can do something like this:

        assert mocked_responses.calls[1].request.headers["Authorization"] == \
            "Token this-is-token"

But aresponses doesn't seem to store the requests/responses anywhere.

pytest-asyncio as requirement

I've tried to follow the pytest-asyncio topic evolution (#52)

I've seen the PR #55 was closed due to breaking compatibilities with aiohttp later on.

Is absolutely necessary to install either pytest-aiohttp or pytest-asyncio?
Is it considered in a near future to tackle this? (We would prefer not to install any test pkg if possible)

Thanks in advance

Don't close event loop passed as argument

Hi. I have a proposal to don't close the loop when there is no match.
Code snippet:

    async def _find_response(self, request):
        host, path, path_qs, method = request.host, request.path, request.path_qs, request.method
        logger.info(f"Looking for match for {host} {path} {method}")
        i = 0
        host_matched = False
        path_matched = False
        for host_pattern, path_pattern, method_pattern, response, match_querystring in self._responses:
            if _text_matches_pattern(host_pattern, host):
                host_matched = True
                if (not match_querystring and _text_matches_pattern(path_pattern, path)) or (
                    match_querystring and _text_matches_pattern(path_pattern, path_qs)
                ):
                    path_matched = True
                    if _text_matches_pattern(method_pattern, method.lower()):
                        del self._responses[i]

                        if callable(response):
                            if asyncio.iscoroutinefunction(response):
                                return await response(request)
                            return response(request)

                        if isinstance(response, str):
                            return self.Response(body=response)

                        return response
            i += 1
        self._exception = Exception(f"No Match found for {host} {path} {method}.  Host Match: {host_matched}  Path Match: {path_matched}")
        self._loop.stop()
        raise self._exception  # noqa

This can cause errors like RuntimeError: Event loop stopped before Future completed. if this method was called in coroutine.

Setup CI

I see there is a .travis.yml file, but Travis is not hooked up to the repo.

Why does aresponses stop the event loop?

As far as i can tell, aresponese does not create new event loops. It just uses loops that something else (pytest asyncio libraries, or the user themselves) created.

If you didn't start the loop, you shouldn't stop the loop. This is causing

RuntimeError: Event loop stopped before Future completed.

which shouldn't exist because aresponses can't possibly know what the loop it is using is used for elsewhere.

What purpose does this self._loop.stop() serve? Everything seems to run more easily without it.
https://github.com/CircleUp/aresponses/blob/b21796733d5c230e792384b4ce2055df886e987a/aresponses/main.py#L107
I think this is the reason you have issues with pytest-aiohttp as well (I'm not using that).

aresponses does not work with aiohttp v3.8.1

After upgrading from aiohttp v3.7.4.post0 to version 3.8.1, my aresponses test cases fail with the below:

Traceback (most recent call last):
  File "../lib/python3.6/site-packages/aiohttp/web_protocol.py", line 602, in finish_response
    prepare_meth = resp.prepare
AttributeError: 'NoneType' object has no attribute 'prepare'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "../lib/python3.6/site-packages/aiohttp/web_protocol.py", line 514, in start
    resp, reset = await task
  File "../lib/python3.6/site-packages/aiohttp/web_protocol.py", line 460, in _handle_request
    reset = await self.finish_response(request, resp, start_time)
  File "../lib/python3.6/site-packages/aiohttp/web_protocol.py", line 605, in finish_response
    raise RuntimeError("Missing return " "statement on request handler")
RuntimeError: Missing return statement on request handler

If I downgrade to 3.7.4.post0 I don't experience this issue.

Unable to obtain body of a post request.

So I have a simple function that makes a simple post request with some parameters:

    async with aiohttp.ClientSession(raise_for_status=True) as session:
        async with session.post(trigger.url, data=params) as response:
            return await response.json()

When I try to test the parameters passed:

        aresponses.add("test.url.com", "/trigger/path", "POST", aresponses.Response(body=json.dumps({"some": "response"}), status=201, headers={"Content-Type": "application/json"}))
        response = await utils.invoke_trigger(trigger, variables)
        print(await aresponses.history[0].request.post())

I get following error:

self = <StreamReader 75 bytes eof e=PayloadAccessError()>

    async def readany(self) -> bytes:
        if self._exception is not None:
>           raise self._exception
E           aiohttp.web_protocol.PayloadAccessError

.venv/lib/python3.8/site-packages/aiohttp/streams.py:386: PayloadAccessError

Dependency on pytest-asyncio breaks support for aiohttp

When trying to run a test with the workaround provided for compatibility with aiohttp, the test fail with an error:

Test session starts (platform: darwin, Python 3.7.3, pytest 5.4.1, pytest-sugar 0.9.3)
rootdir: /Users/joakim/src/hass/nibeuplink
plugins: timeout-1.3.4, aresponses-2.0.0, mock-3.1.0, requests-mock-1.7.0, sugar-0.9.3, cov-2.5.1, threadleak-0.2.0, aiohttp-0.3.0, asyncio-0.14.0
collecting ... 

――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――― test_get_system ――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――――

aresponses = <aresponses.main.ResponsesMockServer object at 0x10745a978>, uplink = <nibeuplink.uplink.Uplink object at 0x10745ac18>

    async def test_get_system(aresponses, uplink: Uplink):
>       result = await uplink.get_system(123)

tests/test_uplink.py:30: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
src/nibeuplink/uplink.py:175: in get_system
    return await self.get(f"systems/{system_id}")
src/nibeuplink/uplink.py:74: in get
    "GET", f"{self.base}/api/v1/{url}", *args, **kwargs
src/nibeuplink/session.py:175: in request
    response = await self.session.request(*args, auth=await self._get_auth(), **kw)
../home-assistant/venv/lib/python3.7/site-packages/aiohttp/client.py:426: in _request
    with timer:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <aiohttp.helpers.TimerContext object at 0x10745a0f0>

    def __enter__(self) -> BaseTimerContext:
        task = current_task(loop=self._loop)
    
        if task is None:
>           raise RuntimeError('Timeout context manager should be used '
                               'inside a task')
E           RuntimeError: Timeout context manager should be used inside a task

../home-assistant/venv/lib/python3.7/site-packages/aiohttp/helpers.py:579: RuntimeError

 tests/test_uplink.py ⨯                                                                                                                                                          100% ██████████
=================================================================================== short test summary info ====================================================================================
FAILED tests/test_uplink.py::test_get_system - RuntimeError: Timeout context manager should be used inside a task

After doing a pip uninstall pytest-asyncio, the test succeeds. Just having pytest-asyncio installed breaks behavior of aiohttp tests.

Would it be possible to have aresponses check which pytest aio handler is installed and use the loop accordingly, then drop the requirement from setup.py (or make it either or).

README json example incomplete

This example in the README doesn't work against aiohttp as client because it checks the content type header:

    # JSON response
    aresponses.add('foo.com', '/', 'get', aresponses.Response(body=b'{"status":"ok"}'))

It requires a content type header, correct version should be:

    # JSON response
    aresponses.add('foo.com', '/', 'get', aresponses.Response(body=b'{"status":"ok"}', headers={"Content-Type": "application/json"}))

[bug] setup.py pytest entrypoints bug collides with other pytest plugins

https://github.com/aresponses/aresponses/blob/4bb4ea08c2a71a59ece5bac3e0df1ada59d0274e/setup.py#L35=

should instead be something like:

    entry_points={"pytest11": ["aresponses = aresponses.main"]},

e.g. what pytest-mock and pytest-asyncio do

The output of:

pytest --trace-config

Will show what name a plugin as been registered with, and aresponses has been registered with name_of_plugin, which collides with other pytest plugins that have made the same mistake, e.g. ipwnponies/pytest-antilru#22

Is dependency pytest-asyncio required?

Hi there! Appreciate the fantastic library.

I'm curious about whether we actually need pytest.asyncio as a dependency. Looks like that aiohttp.pytest_plugin supports coroutine executions on its own.

To make packages tests run we need to just add loop as dependency and remove all @pytest.mark.asyncio decorators

@pytest.fixture()
async def aresponses(loop) -> ResponsesMockServer:
    async with ResponsesMockServer(loop=loop) as server:
        yield server

Usage with pytest-aiohttp

The documentation states that with pytest-aiohttp one should re-initialise the mock server like below

from aresponses import ResponsesMockServer

@pytest.fixture
async def aresponses(loop):
    async with ResponsesMockServer(loop=loop) as server:
        yield server

However following this documentation and passing the aresponses-fixture to my test function, I get the following error

tests/test_foo.py::test_foo[pyloop]
  /Users/mvalkonen/Library/Caches/pypoetry/virtualenvs/test-py3.7/lib/python3.7/site-packages/aresponses/main.py:53: DeprecationWarning: The object should be created from async function
    srv = Server(self._handler, loop=self._loop, debug=True, **kwargs)

I can reproduce this with the following example

@pytest.fixture
async def aresponses(loop):
    async with ResponsesMockServer(loop=loop) as server:
        yield server


async def test_foo(aresponses):
    pass

How should one use aresponses with pytest-aiohttp?

PytestDeprecationWarning: "event_loop" fixture

PytestDeprecationWarning: aresponses is asynchronous and explicitly requests the "event_loop" fixture. 
Asynchronous fixtures and test functions should use "asyncio.get_running_loop()" instead.

Support for range responses?

Hi,

Is there a way to make this mock range responses? I want to mock the serving of a large file which will be downloaded in chunks.

thanks

Mocking a HTTP 204 No Content response

Hi,
I'd like to mock a API which for some methods returns the status code 204 and None as the body.

I tried a few ways to implement it with this lib but failed.

# No Content response HTTP 204
aresponses.add("foo.com", "/", "get", aresponses.Response(body=None, text=None, status=204))
url = "http://foo.com"
async with aiohttp.ClientSession() as session:
    async with session.get(url) as response:
        text= await response.text()
        assert text is None

Can you point me in a direction to mock a `None` response body?

Combining aresponses with aiohttp_client

Hello,

Assume a code under test which is a simple aiohttp web server that does an external HTTP call using aiohttp's ClientSession:
https://repl.it/@zopieux/aiohttpAresponses#app.py

To test my server works, I can write a small test:
https://repl.it/@zopieux/aiohttpAresponses#test.py

and this works. Sadly, this is non-hermetic, it makes real requests on the internet to httpbin.org.

I know, I'll use aresponses! So I write https://repl.it/@zopieux/aiohttpAresponses#testaresponses.py

If you run this REPL, you'll see the first "internet enabled" test passes, but the aresponses test hangs indefinitely on the client.get() line in the test.

I tried debugging it but couldn't find much. I can see the two aiohttp test webservers spawned on different ports on 127.0.0.1.

My understanding is that aresponses intercepts all the ClientSession requests, even the ones that are supposed to go to the other test server created by aiohttp_client. I thought I could add:

aresponses.add("127.0.0.1", response=aresponses.passthrough)
aresponses.add("127.0.0.1:4241", response=aresponses.passthrough)  # 4241 is aiohttp test port

to make it explicit, but aresponses doesn't care, the calls are not forwarded to the aiohttp_client server and this test hangs for minutes (TCP timeout I suppose).

How can I make aresponses play nice with aiohttp_client?

How to test timeouts?

I'm attempting to test an aiohttp call with a timeout; is there a mechanism in aresponses that would allow me to do that?

Recommend Projects

  • React photo React

    A declarative, efficient, and flexible JavaScript library for building user interfaces.

  • Vue.js photo Vue.js

    🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.

  • Typescript photo Typescript

    TypeScript is a superset of JavaScript that compiles to clean JavaScript output.

  • TensorFlow photo TensorFlow

    An Open Source Machine Learning Framework for Everyone

  • Django photo Django

    The Web framework for perfectionists with deadlines.

  • D3 photo D3

    Bring data to life with SVG, Canvas and HTML. 📊📈🎉

Recommend Topics

  • javascript

    JavaScript (JS) is a lightweight interpreted programming language with first-class functions.

  • web

    Some thing interesting about web. New door for the world.

  • server

    A server is a program made to process requests and deliver data to clients.

  • Machine learning

    Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.

  • Game

    Some thing interesting about game, make everyone happy.

Recommend Org

  • Facebook photo Facebook

    We are working to build community through open source technology. NB: members must have two-factor auth.

  • Microsoft photo Microsoft

    Open source projects and samples from Microsoft.

  • Google photo Google

    Google ❤️ Open Source for everyone.

  • D3 photo D3

    Data-Driven Documents codes.