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
|
#
# 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.
"""JSON-related content handling."""
import json
from gabbi.exception import GabbiDataLoadError
from gabbi.handlers import base
from gabbi import json_parser
class JSONHandler(base.ContentHandler):
"""A ContentHandler for JSON
* Structured test ``data`` is turned into JSON when request
content-type is JSON.
* Response bodies that are JSON strings are made into Python
data on the test ``response_data`` attribute when the response
content-type is JSON.
* A ``response_json_paths`` response handler is added.
* JSONPaths in $RESPONSE substitutions are supported.
"""
test_key_suffix = 'json_paths'
test_key_value = {}
@staticmethod
def accepts(content_type):
content_type = content_type.lower()
parameters = ''
if ';' in content_type:
content_type, parameters = content_type.split(';', 1)
content_type = content_type.strip()
return (content_type.endswith('+json') or
content_type == 'application/json'
and 'stream=' not in parameters)
@classmethod
def replacer(cls, response_data, match):
return cls.extract_json_path_value(response_data, match)
@staticmethod
def dumps(data, pretty=False, test=None):
if pretty:
return json.dumps(data, indent=2, separators=(',', ': '))
else:
return json.dumps(data)
@staticmethod
def loads(data):
try:
return json.loads(data)
except ValueError as exc:
raise GabbiDataLoadError('unable to parse data') from exc
@staticmethod
def load_data_file(test, file_path):
info = test.load_data_file(file_path)
info = str(info, 'UTF-8')
return json.loads(info)
@staticmethod
def extract_json_path_value(data, path):
"""Extract the value at JSON Path path from the data.
The input data is a Python datastructure, not a JSON string.
"""
path_expr = json_parser.parse(path)
matches = [match.value for match in path_expr.find(data)]
if matches:
if len(matches) > 1:
return matches
else:
return matches[0]
else:
raise ValueError(
"JSONPath '%s' failed to match on data: '%s'" % (path, data))
def action(self, test, path, value=None):
"""Test json_paths against json data."""
# Do template expansion in the left hand side.
lhs_path = test.replace_template(path)
rhs_path = rhs_match = None
try:
lhs_match = self.extract_json_path_value(
test.response_data, lhs_path)
except AttributeError:
raise AssertionError('unable to extract JSON from test results')
except ValueError:
raise AssertionError('left hand side json path %s cannot match '
'%s' % (path, test.response_data))
# read data from disk if the value starts with '<@'
if isinstance(value, str) and value.startswith('<@'):
# Do template expansion in the rhs if rhs_path is provided.
if ':' in value:
value, rhs_path = value.split(':$', 1)
rhs_path = test.replace_template('$' + rhs_path)
value = self.load_data_file(test, value.replace('<@', '', 1))
if rhs_path:
try:
rhs_match = self.extract_json_path_value(value, rhs_path)
except AttributeError:
raise AssertionError('unable to extract JSON from data on '
'disk')
except ValueError:
raise AssertionError('right hand side json path %s cannot '
'match %s' % (rhs_path, value))
# If expected is a string, check to see if it is a regex.
is_regex = isinstance(value, str) and self.is_regex(value)
expected = (rhs_match or
test.replace_template(value, escape_regex=is_regex))
match = lhs_match
if is_regex and not rhs_match:
expected = expected[1:-1]
# match may be a number so stringify
match = str(match)
test.assertRegex(
match, expected,
'Expect jsonpath %s to match /%s/, got %s' %
(path, expected, match))
else:
test.assertEqual(expected, match,
'Unable to match %s as %s, got %s' %
(path, expected, match))
|
