Provide a $LOCATION, $NETLOC, $SCHEME variables in YAML

This makes it easier to make references to prior requests and
confirm URL construction.

If there is a Location header in a response, then the TestCase
recieves a location attribute to be referred to by subsequent
requests. This is to encourage a more RESTful approach.

In a future commit and reference to all the headers and body
text may be saved, assuming it seems useful.
This commit is contained in:
Chris Dent 2014-12-17 17:33:02 +00:00
parent 305e5820b3
commit 66799d4df2
3 changed files with 55 additions and 5 deletions

View File

@ -98,6 +98,8 @@ class HTTPTestCase(testtools.TestCase):
If provided with a full URL, just return that. If SSL is requested
set the scheme appropriately.
Scheme and netloc are saved for later use in comparisons.
"""
parsed_url = urlparse.urlsplit(url)
url_scheme = parsed_url[0]
@ -111,16 +113,26 @@ class HTTPTestCase(testtools.TestCase):
scheme = 'https'
full_url = urlparse.urlunsplit((scheme, netloc, parsed_url[2],
parsed_url[3], ''))
self.scheme = scheme
self.netloc = netloc
else:
full_url = url
self.scheme = url_scheme
self.netloc = parsed_url[1]
return full_url
def _run_test(self):
"""Make an HTTP request."""
test = self.test_data
http = self.http
base_url = test['url']
full_url = self._parse_url(test['url'], test['ssl'])
if '$LOCATION' in base_url:
# Let AttributeError raise
base_url = base_url.replace('$LOCATION', self.prior.location)
full_url = self._parse_url(base_url, test['ssl'])
method = test['method'].upper()
headers = test['request_headers']
if method == 'GET' or method == 'DELETE':
@ -137,6 +149,11 @@ class HTTPTestCase(testtools.TestCase):
headers=headers,
body=body
)
# Set location attribute for follow on requests
if 'location' in response:
self.location = response['location']
self._assert_response(response, content, test['status'],
headers=test['response_headers'],
expected=test['expected'],
@ -152,7 +169,11 @@ class HTTPTestCase(testtools.TestCase):
if headers:
for header in headers:
self.assertEqual(headers[header], response[header])
header_value = headers[header].replace('$SCHEME', self.scheme)
header_value = header_value.replace('$NETLOC', self.netloc)
self.assertEqual(header_value, response[header],
'Expect header %s with value %s, got %s' %
(header, header_value, response[header]))
output = content.decode('utf-8')
# Compare strings in response body

View File

@ -1,8 +1,6 @@
#
# Tests for testing gabbi, using the built in SimpleWsgi app.
#
# TODO(chdent): This probably needs a top level of some kind, so testwide settings
# can be made.
tests:
- name: get simple page
@ -15,6 +13,7 @@ tests:
response_headers:
allow: GET, PUT, POST, DELETE, PATCH
x-gabbi-method: DIE
x-gabbi-url: $SCHEME://$NETLOC/
- name: query returned
url: /somewhere?foo=1&bar=2&bar=3
@ -28,6 +27,17 @@ tests:
- "2"
- "3"
- name: simple post
url: /named/thing
method: POST
response_headers:
location: $SCHEME://$NETLOC/named/thing
- name: use prior location
url: $LOCATION
response_headers:
x-gabbi-url: $SCHEME://$NETLOC/named/thing
- name: google
url: https://google.com/

View File

@ -41,9 +41,25 @@ class SimpleWsgi(object):
query_data = urlparse.parse_qs(environ.get('QUERY_STRING', ''))
query_output = json.dumps(query_data)
request_url = environ.get('REQUEST_URI',
environ.get('RAW_URI', 'unknown'))
path, query, fragment = urlparse.urlsplit(request_url)[2:]
server_name = environ.get('SERVER_NAME')
server_port = 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
request_url = urlparse.urlunsplit((server_scheme, netloc, path,
query, fragment))
headers = [
('X-Gabbi-method', request_method),
('Content-Type', 'application/json')
('Content-Type', 'application/json'),
('X-Gabbi-url', request_url),
]
if request_method not in METHODS:
@ -52,6 +68,9 @@ class SimpleWsgi(object):
start_response('405 Method Not Allowed', headers)
return []
if request_method.startswith('P'):
headers.append(('Location', request_url))
start_response('200 OK', headers)
return [query_output.encode('utf-8')]