8a031149b6
This change implements template substitution ($ENVIRON, $RESPONSE,
etc) in the left hand side of a response_json_paths test. This enables
testing in situations where responses may include a server side
generated identifier as the key in an object structure.
As things stand, this is the bare minimum implementation that assumes
that the test creator will be responsible for dealing with the
vargrancies of quoting that will be required in order to generate
legitimate JSON path expressions. For example if there is a uuid (with
hyphens) at $RESPONSE['$.id'] then this expression is likely to fail:
$.nested.structure.$RESPONSE['$.id'].name: foobar
as it will evaluate to something like
$.nested.structure.ADC8AAFC-D564-40D1-9724-7680D3C010C2.name: foobar
which _may_ be treated as an arithemtic expression by the json parser.
The test creator would need to do:
$.nested.structure["$RESPONSE['$.id']"].name: foobar
or variants thereof to get a useful expression.
Docs to follow if this is considered an okay way to implement the
feature.
Fixes #170
135 lines
5.0 KiB
Python
135 lines
5.0 KiB
Python
#
|
|
# 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.
|
|
"""
|
|
SimpleWsgi provides a WSGI callable that can be used in tests to
|
|
reflect posted data and otherwise confirm headers and queries.
|
|
"""
|
|
|
|
import json
|
|
|
|
from six.moves.urllib import parse as urlparse
|
|
|
|
|
|
CURRENT_POLL = 0
|
|
METHODS = ['GET', 'PUT', 'POST', 'DELETE', 'PATCH']
|
|
|
|
|
|
class SimpleWsgi(object):
|
|
"""A simple wsgi application to use in tests."""
|
|
|
|
def __call__(self, environ, start_response):
|
|
global METHODS
|
|
global CURRENT_POLL
|
|
|
|
request_method = environ['REQUEST_METHOD'].upper()
|
|
query_data = urlparse.parse_qs(environ.get('QUERY_STRING', ''))
|
|
request_url = environ.get('REQUEST_URI',
|
|
environ.get('RAW_URI', 'unknown'))
|
|
path_info = environ.get('PATH_INFO', '')
|
|
accept_header = environ.get('HTTP_ACCEPT')
|
|
content_type_header = environ.get('CONTENT_TYPE', '')
|
|
|
|
full_request_url = self._fully_qualify(environ, request_url)
|
|
|
|
if accept_header:
|
|
response_content_type = accept_header
|
|
else:
|
|
# JSON doesn't need a charset but we throw one in here
|
|
# to exercise the decoding code
|
|
response_content_type = (
|
|
'application/json ; charset=utf-8 ; stop=no')
|
|
|
|
headers = [
|
|
('X-Gabbi-method', request_method),
|
|
('Content-Type', response_content_type),
|
|
('X-Gabbi-url', full_request_url),
|
|
]
|
|
|
|
if request_method == 'DIE':
|
|
raise Exception('because you asked me to')
|
|
|
|
if request_method not in METHODS:
|
|
headers.append(
|
|
('Allow', ', '.join(METHODS)))
|
|
start_response('405 Method Not Allowed', headers)
|
|
return []
|
|
|
|
if request_method.startswith('P'):
|
|
body = environ['wsgi.input'].read()
|
|
if body:
|
|
if not content_type_header:
|
|
start_response('400 Bad request', headers)
|
|
return []
|
|
if content_type_header == 'application/json':
|
|
body_data = json.loads(body.decode('utf-8'))
|
|
if query_data:
|
|
query_data.update(body_data)
|
|
else:
|
|
query_data = body_data
|
|
headers.append(('Location', full_request_url))
|
|
|
|
if path_info == '/presenter':
|
|
start_response('200 OK', [('Content-Type', 'text/html')])
|
|
return [b"""<!DOCTYPE html>
|
|
<html>
|
|
<head>
|
|
<meta charset="utf-8">
|
|
<title>Hello World</title>
|
|
</head>
|
|
<body>
|
|
<h1>Hello World</h1>
|
|
<p>lorem ipsum dolor sit amet</p>
|
|
</body>
|
|
</html>
|
|
"""]
|
|
elif path_info.startswith('/poller'):
|
|
if CURRENT_POLL == 0:
|
|
CURRENT_POLL = int(query_data.get('count', [5])[0])
|
|
start_response('400 Bad Reqest', [])
|
|
return []
|
|
else:
|
|
CURRENT_POLL -= 1
|
|
if CURRENT_POLL > 0:
|
|
start_response('400 Bad Reqest', [])
|
|
return []
|
|
else:
|
|
CURRENT_POLL = 0
|
|
# fall through if we've ended the loop
|
|
elif path_info == '/cookie':
|
|
headers.append(('Set-Cookie', 'session=1234; domain=.example.com'))
|
|
elif path_info == '/jsonator':
|
|
json_data = json.dumps({query_data['key'][0]:
|
|
query_data['value'][0]})
|
|
start_response('200 OK', [('Content-Type', 'application/json')])
|
|
return [json_data.encode('utf-8')]
|
|
|
|
start_response('200 OK', headers)
|
|
|
|
query_output = json.dumps(query_data)
|
|
return [query_output.encode('utf-8')]
|
|
|
|
@staticmethod
|
|
def _fully_qualify(environ, url):
|
|
"""Turn a URL path into a fully qualified URL."""
|
|
split_url = urlparse.urlsplit(url)
|
|
server_name = environ.get('SERVER_NAME')
|
|
server_port = str(environ.get('SERVER_PORT'))
|
|
server_scheme = environ.get('wsgi.url_scheme')
|
|
if server_port not in ['80', '443']:
|
|
netloc = '%s:%s' % (server_name, server_port)
|
|
else:
|
|
netloc = server_name
|
|
|
|
return urlparse.urlunsplit((server_scheme, netloc, split_url.path,
|
|
split_url.query, split_url.fragment))
|