Skip to content

Instantly share code, notes, and snippets.

Created February 8, 2019 19:59
Show Gist options
  • Save Droogans/6e284ad009138a01e34db288e5568097 to your computer and use it in GitHub Desktop.
Save Droogans/6e284ad009138a01e34db288e5568097 to your computer and use it in GitHub Desktop.
This is as close to abusing the fixture pattern as I would ever like to get, but it was needed in a hurry and it got rid of a lot of duplicate code.
from itertools import chain, starmap
import pydash
import pytest
def response(test_request):
Use after calling the `test_request` fixture. Provides the resulting api request response object(s).
return test_request.get('response')
def json(test_request):
Use after calling the `test_request` fixture. Provides the resulting api request response payloads(s).
return test_request.get('json')
async def test_request(request, api_client):
Quickly make an http request, and assert against common aspects of the resulting json and response objects
before testing it further. Reduces boilerplate. Use with `json` and `response` fixtures to examine results further.
If you only need the response object and/or json body from a GET request, without any validation, only a string
endpoint is needed. See more options below to validate responses, or to make POST, PUT, PATCH, and DELETE calls.
You must create and supply your own aiohttp fixture as `api_client` to use this.
### Example usage without validation:
async def test_get_user_by_id(response, json):
assert response.status is not None
assert isinstance(json, dict) is True
### Options for automatically validating api responses before running your tests:
`endpoint`: The api endpoint to make a request against.
`method`: The method to use to make the request. Defaults to "GET".
`request_kwargs`: Keyword arguments to pass to the api_client.
`expected_status_code`: If any options are specified, defaults to 200. Otherwise, no status checks occur.
`is_dict`: Check if the top-level json payload is a dict. Defaults to True.
`expected_json_keys`: Compare the top-level keys of the json body to the values provided here.
`assert_deep_json_keys`: If True, traverse into all objects and collect a unique set of keys representing each
unique key address in the payload. List of objects will produce the union of all key
values found while iterating over all list members. Format is "parent/child".
Default is True.
`type_checks`: For each key/value pair in `type_checks`, assert that the json body's value found at the key address
`key` matches the expected type `value`. Supports lists of types, but at least one must match.
A "key address" is a direct call to
### Example usage with validation:
'endpoint': '/users/1',
'expected_json_keys': [
'type_checks': {
'details': list,
'': str,
'details.0.custer.weight': [float, int],
async def test_get_user_by_id(json):
assert json['userId'] == '1'
### Testing multiple requests together:
In pytest, duplicate fixture calls are forbidden. If multiple test requests are required for a single test,
pass them as additional dicts. They will be called in the order they are listed.
Since there is more than one resulting json payload and more than one resulting request object, the fixtures
`json` and `response` will express their return values differently. These dicts will key off of the index of the
order the calls were made in by default. Calls can accept an optional `name` argument to name the key instead.
### Unnamed multi-requests:
'endpoint': '/abc',
'expected_status_code': 404
}, {
'endpoint': '/'
async def test_root(json):
assert json[0] == { 'error': 'Not Found', 'status_code': 404 }
assert json[1]['name'] == 'amlservice'
### Named multi-requests:
'name': '404',
'endpoint': '/abc',
'expected_status_code': 404
}, {
'name': 'root',
'endpoint': '/'
async def test_root(json):
assert json['404'] == { 'error': 'Not Found', 'status_code': 404 }
assert json['root']['name'] == 'amlservice'
marker = request.node.get_closest_marker('test_request')
request = marker.args[0]
if isinstance(request, str):
response = await api_client.get(request)
json = await response.json()
return { 'json': json, 'response': response }
elif not isinstance(request, dict):
raise ValueError('test_request fixture expects all arguments passed in as a dict')
if len(marker.args) > 1:
requests = marker.args
return await _test_all_requests(requests, api_client)
response, json = await _test_one_request(request, api_client)
return { 'json': json, 'response': response }
async def _test_all_requests(requests, api_client):
all_responses = { 'json': {}, 'response': {} }
for index, request in enumerate(requests):
name = request.pop('name', index)
response, json = await _test_one_request(request, api_client)
all_responses['response'][name] = response
all_responses['json'][name] = json
return all_responses
async def _test_one_request(request, api_client):
method = request.pop('method', 'GET').lower()
request_fn = getattr(api_client, method)
response = await request_fn(request.pop('endpoint'), **request.pop('request_kwargs', {}))
json = await _assert_request(response, **request)
return response, json
async def _assert_request(
response, expected_status_code=200, is_dict=True,
expected_json_keys=None, assert_deep_json_keys=True, type_checks={}, name=None
if expected_status_code:
assert response.status == expected_status_code
json = await response.json()
if expected_json_keys is not None:
if assert_deep_json_keys:
assert _flattened_deep_json_keys(json) == sorted(expected_json_keys)
assert sorted(json.keys()) == sorted(expected_json_keys)
if is_dict:
assert isinstance(json, dict), f"Expected response json to be a dict, instead it was a {type(json)}"
if type_checks is not None:
for path, expected_type in type_checks.items():
failure_message = '{} is not of type {}'.format(path, expected_type)
if isinstance(expected_type, list):
assert type(pydash.get(json, path)) in expected_type, failure_message
assert isinstance(pydash.get(json, path), expected_type), failure_message
return json
def _flattened_deep_json_keys(json):
Taken from:
if isinstance(json, list):
# force a top-level `parent_key` to prevent any pre-pended characters
json = { '': json }
def _unpack(parent_key, parent_value):
if isinstance(parent_value, dict):
for key, value in parent_value.items():
if parent_key is not '':
traversed_key = '{}/{}'.format(parent_key, key)
else: # e.g., if json = { '': json }
traversed_key = key
yield traversed_key, value
elif isinstance(parent_value, list):
for value in parent_value:
yield parent_key, value
yield parent_key, parent_value
while True:
# Keep unpacking the json file until all values are atomic elements (not dictionary or list)
json = dict(chain.from_iterable(starmap(_unpack, json.items())))
if not any(isinstance(value, dict) for value in json.values()) and \
not any(isinstance(value, list) for value in json.values()):
return sorted(set(json.keys()))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment