Move the handlers into their own dir

And separate out json to its own file.
This commit is contained in:
Chris Dent 2016-04-02 15:42:22 +01:00
parent d4cd98e0b3
commit 48b52d246b
5 changed files with 123 additions and 106 deletions

View File

@ -37,6 +37,7 @@ import yaml
from gabbi import case from gabbi import case
from gabbi import handlers from gabbi import handlers
from gabbi.handlers import jsonhandler
from gabbi import httpclient from gabbi import httpclient
from gabbi import reporter from gabbi import reporter
from gabbi import suite as gabbi_suite from gabbi import suite as gabbi_suite
@ -46,7 +47,7 @@ HANDLERS = [
handlers.ForbiddenHeadersResponseHandler, handlers.ForbiddenHeadersResponseHandler,
handlers.HeadersResponseHandler, handlers.HeadersResponseHandler,
handlers.StringResponseHandler, handlers.StringResponseHandler,
handlers.JSONHandler, jsonhandler.JSONHandler,
] ]

View File

@ -12,10 +12,6 @@
# under the License. # under the License.
"""Handlers for processing the body of a response in various ways.""" """Handlers for processing the body of a response in various ways."""
import json
from gabbi import json_parser
class ResponseHandler(object): class ResponseHandler(object):
"""Add functionality for making assertions about an HTTP response. """Add functionality for making assertions about an HTTP response.
@ -82,6 +78,35 @@ class ResponseHandler(object):
return not self.__eq__(other) return not self.__eq__(other)
class ContentHandler(object):
"""A mixin for ResponseHandlers that adds content handling."""
@staticmethod
def accepts(content_type):
"""Return True if this handler can handler this type."""
return False
@staticmethod
def gen_replacer(test):
"""Return a function which does RESPONSE replacing."""
def replacer_func(match):
return match.group('arg')
return replacer_func
@staticmethod
def dumps(data, pretty=False):
"""Return structured data as a string.
If pretty is true, prettify.
"""
return data
@staticmethod
def loads(data):
"""Create structured (Python) data from a stream."""
return data
class StringResponseHandler(ResponseHandler): class StringResponseHandler(ResponseHandler):
"""Test for matching strings in the the response body.""" """Test for matching strings in the the response body."""
@ -141,99 +166,3 @@ class HeadersResponseHandler(ResponseHandler):
test.assertEqual(header_value, response_value, test.assertEqual(header_value, response_value,
'Expect header %s with value %s, got %s' % 'Expect header %s with value %s, got %s' %
(header, header_value, response[header])) (header, header_value, response[header]))
class ContentHandler(object):
"""A mixin for ResponseHandlers that add content handling."""
@staticmethod
def accepts(content_type):
"""Return True if this handler can handler this type."""
return False
@staticmethod
def gen_replacer(test):
"""Return a function which does RESPONSE replacing."""
def replacer_func(match):
return match.group('arg')
return replacer_func
@staticmethod
def dumps(data, pretty=False):
"""Return structured data as a string.
If pretty is true, prettify.
"""
return data
@staticmethod
def loads(data):
"""Create structured (Python) data from a stream."""
return data
class JSONHandler(ResponseHandler, ContentHandler):
test_key_suffix = 'json_paths'
test_key_value = {}
@staticmethod
def accepts(content_type):
content_type = content_type.split(';', 1)[0].strip()
return (content_type.endswith('+json') or
content_type.startswith('application/json'))
@classmethod
def gen_replacer(cls, test):
def replacer_func(match):
path = match.group('arg')
return str(cls.extract_json_path_value(
test.prior.response_data, path))
return replacer_func
@staticmethod
def dumps(data, pretty=False):
if pretty:
return json.dumps(data, indent=2, separators=(',', ': '))
else:
return json.dumps(data)
@staticmethod
def loads(data):
return json.loads(data)
@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."""
# 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 = self.extract_json_path_value(
test.response_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.response_data))
expected = test.replace_template(value)
test.assertEqual(expected, match, 'Unable to match %s as %s, got %s'
% (path, expected, match))

View File

@ -0,0 +1,86 @@
#
# 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 import handlers
from gabbi import json_parser
class JSONHandler(handlers.ResponseHandler,
handlers.ContentHandler):
test_key_suffix = 'json_paths'
test_key_value = {}
@staticmethod
def accepts(content_type):
content_type = content_type.split(';', 1)[0].strip()
return (content_type.endswith('+json') or
content_type.startswith('application/json'))
@classmethod
def gen_replacer(cls, test):
def replacer_func(match):
path = match.group('arg')
return str(cls.extract_json_path_value(
test.prior.response_data, path))
return replacer_func
@staticmethod
def dumps(data, pretty=False):
if pretty:
return json.dumps(data, indent=2, separators=(',', ': '))
else:
return json.dumps(data)
@staticmethod
def loads(data):
return json.loads(data)
@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."""
# 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 = self.extract_json_path_value(
test.response_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.response_data))
expected = test.replace_template(value)
test.assertEqual(expected, match, 'Unable to match %s as %s, got %s'
% (path, expected, match))

View File

@ -19,6 +19,7 @@ import unittest
from gabbi import case from gabbi import case
from gabbi import driver from gabbi import driver
from gabbi import handlers from gabbi import handlers
from gabbi.handlers import jsonhandler
class HandlersTest(unittest.TestCase): class HandlersTest(unittest.TestCase):
@ -83,7 +84,7 @@ class HandlersTest(unittest.TestCase):
self.assertIn(' "location": "house"', msg) self.assertIn(' "location": "house"', msg)
def test_response_json_paths(self): def test_response_json_paths(self):
handler = handlers.JSONHandler(self.test_class) handler = jsonhandler.JSONHandler(self.test_class)
self.test.content_type = "application/json" self.test.content_type = "application/json"
self.test.test_data = {'response_json_paths': { self.test.test_data = {'response_json_paths': {
'$.objects[0].name': 'cow', '$.objects[0].name': 'cow',
@ -98,7 +99,7 @@ class HandlersTest(unittest.TestCase):
self._assert_handler(handler) self._assert_handler(handler)
def test_response_json_paths_fail_data(self): def test_response_json_paths_fail_data(self):
handler = handlers.JSONHandler(self.test_class) handler = jsonhandler.JSONHandler(self.test_class)
self.test.content_type = "application/json" self.test.content_type = "application/json"
self.test.test_data = {'response_json_paths': { self.test.test_data = {'response_json_paths': {
'$.objects[0].name': 'cow', '$.objects[0].name': 'cow',
@ -114,7 +115,7 @@ class HandlersTest(unittest.TestCase):
self._assert_handler(handler) self._assert_handler(handler)
def test_response_json_paths_fail_path(self): def test_response_json_paths_fail_path(self):
handler = handlers.JSONHandler(self.test_class) handler = jsonhandler.JSONHandler(self.test_class)
self.test.content_type = "application/json" self.test.content_type = "application/json"
self.test.test_data = {'response_json_paths': { self.test.test_data = {'response_json_paths': {
'$.objects[1].name': 'cow', '$.objects[1].name': 'cow',

View File

@ -15,10 +15,10 @@
import unittest import unittest
from gabbi import handlers from gabbi.handlers import jsonhandler
extract = handlers.JSONHandler.extract_json_path_value extract = jsonhandler.JSONHandler.extract_json_path_value
nested_data = { nested_data = {
'objects': [ 'objects': [
{'name': 'one', 'value': 'alpha'}, {'name': 'one', 'value': 'alpha'},