/usr/lib/python3/dist-packages/gabbi/handlers.py is in python3-gabbi 1.12.0-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 | #
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Handlers for processing the body of a response in various ways."""
class ResponseHandler(object):
"""Add functionality for making assertions about an HTTP response.
A subclass may implement two methods: ``action`` and ``preprocess``.
``preprocess`` takes one argument, the ``TestCase``. It is called exactly
once for each test before looping across the assertions. It is used,
rarely, to copy the ``test.output`` into a useful form (such as a parsed
DOM).
``action`` takes two or three arguments. If ``test_key_value`` is a list
``action`` is called with the test case and a single list item. If
``test_key_value`` is a dict then ``action`` is called with the test case
and a key and value pair.
"""
test_key_suffix = ''
test_key_value = []
def __init__(self, test_class):
self._key = 'response_%s' % self.test_key_suffix
self._register(test_class)
def __call__(self, test):
if test.test_data[self._key]:
self.preprocess(test)
for item in test.test_data[self._key]:
try:
value = test.test_data[self._key][item]
except (TypeError, KeyError):
value = None
self.action(test, item, value=value)
def preprocess(self, test):
"""Do any pre-single-test preprocessing."""
pass
def action(self, test, item, value=None):
"""Test an individual entry for this response handler.
If the entry is a key value pair the key is in item and the
value in value. Otherwise the entry is considered a single item
from a list.
"""
pass
def _register(self, test_class):
"""Register this handler on the provided test class."""
test_class.base_test[self._key] = self.test_key_value
if self not in test_class.response_handlers:
test_class.response_handlers.append(self)
def __eq__(self, other):
if isinstance(other, ResponseHandler):
return self.__class__ == other.__class__
return False
def __ne__(self, other):
return not self.__eq__(other)
class StringResponseHandler(ResponseHandler):
"""Test for matching strings in the the response body."""
test_key_suffix = 'strings'
test_key_value = []
def action(self, test, expected, value=None):
expected = test.replace_template(expected)
test.assert_in_or_print_output(expected, test.output)
class JSONResponseHandler(ResponseHandler):
"""Test for matching json paths in the json_data."""
test_key_suffix = 'json_paths'
test_key_value = {}
def action(self, test, path, value=None):
"""Test json_paths against json data."""
# NOTE: This process has some advantages over other process that
# might come along because the JSON data has already been
# processed (to provided for the magic template replacing).
# Other handlers that want access to data structures will need
# to do their own processing.
try:
match = test.extract_json_path_value(test.json_data, path)
except AttributeError:
raise AssertionError('unable to extract JSON from test results')
except ValueError:
raise AssertionError('json path %s cannot match %s' %
(path, test.json_data))
expected = test.replace_template(value)
test.assertEqual(expected, match, 'Unable to match %s as %s, got %s'
% (path, expected, match))
class ForbiddenHeadersResponseHandler(ResponseHandler):
"""Test that listed headers are not in the response."""
test_key_suffix = 'forbidden_headers'
test_key_value = []
def action(self, test, forbidden, value=None):
# normalize forbidden header to lower case
forbidden = test.replace_template(forbidden).lower()
test.assertNotIn(forbidden, test.response,
'Forbidden header %s found in response' % forbidden)
class HeadersResponseHandler(ResponseHandler):
"""Compare expected headers with actual headers.
If a header value is wrapped in ``/`` it is treated as a raw
regular expression.
"""
test_key_suffix = 'headers'
test_key_value = {}
def action(self, test, header, value=None):
header = header.lower() # case-insensitive comparison
response = test.response
header_value = test.replace_template(value)
try:
response_value = response[header]
except KeyError:
raise AssertionError(
"'%s' header not present in response: %s" % (
header, response.keys()))
if header_value.startswith('/') and header_value.endswith('/'):
header_value = header_value.strip('/').rstrip('/')
test.assertRegexpMatches(
response_value, header_value,
'Expect header %s to match /%s/, got %s' %
(header, header_value, response_value))
else:
test.assertEqual(header_value, response[header],
'Expect header %s with value %s, got %s' %
(header, header_value, response[header]))
|