A simple asyncio
compatible httpretty
mock using aiohttp
. Tells aiohttp
to return bespoke payloads from certain urls. If aiohttp
is used to access a url that isn't mocked, throws an error to prevent tests from making live requests.
import myproject
import pytest
import aiohttpretty
@pytest.mark.asyncio
async def test_get_keys_foo():
good_response = {'dog': 'woof', 'duck': 'quack'}
good_url = 'http://example.com/dict'
aiohttpretty.register_json_uri('GET', good_url, body=good_response)
bad_response = ['dog', 'duck']
bad_url = 'http://example.com/list'
aiohttpretty.register_json_uri('GET', bad_url, body=bad_response)
aiohttpretty.activate()
# .get_keys_from_url() calls .keys() on response
keys_from_good = await myproject.get_keys_from_url(good_url)
assert keys_from_good == ['dog', 'duck']
with pytest.raises(exceptions.AttributeError) as exc:
await myproject.get_keys_from_url(bad_url)
# aiohttpretty will die screaming that this url hasn't been mocked
# await myproject.get_keys_from_url('http://example.com/unhandled')
aiohttpretty.deactivate()
.register_uri(method, uri, **options)
Register the specified request with aiohttpretty. When aiohttp.request
is called, aiohttpretty
will look for a request that matches in its registry. If it finds one, it returns a response defined by the parameters given in options
. If no matching request is found, aiohttpretty
will throw an error. The HTTP method, uri, and query parameters must all match to be found.
method
: HTTP method to be issued against the uri
.
uri
: The uri to be mocked.
options
: modifiers to the expected request and the mock response
params
: Affects the request. These will be added to the registered uri.
responses
: Affects the response. A list of dicts containing one or more of the following parameters. Each call to the uri will return the next response in the sequence.
status
: Affects the response. The HTTP status code of the response. Defaults to 200.
reason
: Affects the response. The HTTP reason phrase of the response. Defaults to ''
(an empty string).
auto_length
: Affects the response. Generate the Content-Length
header for the response based on the size of the body
. If the same header is provided in the headers
option, the value will be overwritten by the auto-generated value.
headers
: Affects the response. A dict of headers to be included with the response. Default is no headers.
body
: Affects the response. The content to be returned when .read()
is called on the response object. Must be either bytes
, str
, or instanceOf(asyncio.StreamReader)
.
.register_json_uri(method, uri, **options)
Same as .register_uri
but automatically adds a Content-Type: application/json
header to the response (though this can be overwritten if an explicit Content-Type
is passed in the headers
kwarg). Will also json encode the data structure given in the body
kwarg.
.fake_request(method, uri, **kwargs)
aiohttpretty
's fake implementation of aiohttp's request method. Takes the same parameters, but only uses method
, uri
, and params
to lookup the mocked response in the registry. If the data
kwarg is set and is an instance of asyncio.StreamReader
, .fake_request()
will read the stream to exhaustion to mimic its consumption.
.activate()
Replaces aiohttp.ClientSession._request
with .fake_request()
. The original implementation is saved.
.deactivate()
Restores aiohttp.ClientSession._request
to the saved implementation.
.clear()
Purge the registry and the list of intercepted calls.
.has_call(uri, check_params=True)
Checks to see if the given uri was called during the test. By default, will verify that the query params match up. Setting check_params
to False
will strip params from the registered uri, not the passed-in uri.
To simplify usage of aiohttpretty
in tests, you can make a pytest marker that will automatically activate/deactivate aiohttpretty
for the scope of a test. To do so, add the following to your conftest.py
:
import aiohttpretty
def pytest_configure(config):
config.addinivalue_line(
'markers',
'aiohttpretty: mark tests to activate aiohttpretty'
)
def pytest_runtest_setup(item):
marker = item.get_marker('aiohttpretty')
if marker is not None:
aiohttpretty.clear()
aiohttpretty.activate()
def pytest_runtest_teardown(item, nextitem):
marker = item.get_marker('aiohttpretty')
if marker is not None:
aiohttpretty.deactivate()
Then add @pytest.mark.aiohttpretty
before async def test_foo
.