Skip to content

Commit

Permalink
Merge pull request #194 from samuelfekete/add_history
Browse files Browse the repository at this point in the history
Add $HISTORY variable
  • Loading branch information
cdent committed Dec 23, 2016
2 parents 3c1a4db + 31d1b1d commit 970fb65
Show file tree
Hide file tree
Showing 8 changed files with 264 additions and 26 deletions.
95 changes: 79 additions & 16 deletions gabbi/case.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@
'LOCATION',
'COOKIE',
'LAST_URL',
'URL',
'HEADERS',
'RESPONSE',
]
Expand Down Expand Up @@ -209,11 +210,21 @@ def _cookie_replace(self, message):
With cookie data from set-cookie in the prior request.
"""
response_cookies = self.prior.response['set-cookie']
return re.sub(self._simple_replacer_regex('COOKIE'),
self._cookie_replacer, message)

def _cookie_replacer(self, match):
"""Replace a regex match with the cookie of a previous response."""
case = match.group('case')
if case:
referred_case = self.history[case]
else:
referred_case = self.prior
response_cookies = referred_case.response['set-cookie']
cookies = http_cookies.SimpleCookie()
cookies.load(response_cookies)
cookie_string = cookies.output(attrs=[], header='', sep=',').strip()
return message.replace('$COOKIE', cookie_string)
return cookie_string

def _headers_replace(self, message):
"""Replace a header indicator in a message with that headers value from
Expand All @@ -225,7 +236,12 @@ def _headers_replace(self, message):
def _header_replacer(self, match):
"""Replace a regex match with the value of a prior header."""
header_key = match.group('arg')
return self.prior.response[header_key.lower()]
case = match.group('case')
if case:
referred_case = self.history[case]
else:
referred_case = self.prior
return referred_case.response[header_key.lower()]

def _last_url_replace(self, message):
"""Replace $LAST_URL in a message.
Expand All @@ -234,12 +250,39 @@ def _last_url_replace(self, message):
"""
return message.replace('$LAST_URL', self.prior.url)

def _url_replace(self, message):
"""Replace $URL in a message.
With the URL used in a previous request.
"""
return re.sub(self._simple_replacer_regex('URL'),
self._url_replacer, message)

def _url_replacer(self, match):
"""Replace a regex match with the value of a previous url."""
case = match.group('case')
if case:
referred_case = self.history[case]
else:
referred_case = self.prior
return referred_case.url

def _location_replace(self, message):
"""Replace $LOCATION in a message.
With the location header from the prior request.
With the location header from a previous request.
"""
return message.replace('$LOCATION', self.prior.location)
return re.sub(self._simple_replacer_regex('LOCATION'),
self._location_replacer, message)

def _location_replacer(self, match):
"""Replace a regex match with the value of a previous location."""
case = match.group('case')
if case:
referred_case = self.history[case]
else:
referred_case = self.prior
return referred_case.location

def _load_data_file(self, filename):
"""Read a file from the current test directory."""
Expand Down Expand Up @@ -290,23 +333,43 @@ def _parse_url(self, url):
return urlparse.urlunsplit((parsed_url.scheme, parsed_url.netloc,
parsed_url.path, query_string, ''))

_history_regex = (
r"(?:\$HISTORY\[(?P<quote1>['\"])(?P<case>.+?)(?P=quote1)\]\.)??"
)

@staticmethod
def _replacer_regex(key):
"""Compose a regular expression for test template variables."""
return r"\$%s\[(?P<quote>['\"])(?P<arg>.+?)(?P=quote)\]" % key
case = HTTPTestCase._history_regex
return r"%s\$%s\[(?P<quote>['\"])(?P<arg>.+?)(?P=quote)\]" % (
case, key)

@staticmethod
def _simple_replacer_regex(key):
"""Compose a regular expression for simple variable replacement."""
case = HTTPTestCase._history_regex
return r"%s\$%s" % (case, key)

def _response_replace(self, message):
"""Replace a content from the prior request with a value."""
replacer_class = self.get_content_handler(
self.prior.response.get('content-type'))
if replacer_class:
replacer_func = replacer_class.gen_replacer(self)
else:
# If no handler can be found use the null replacer,
# which returns "foo" when "$RESPONSE['foo']".
replacer_func = base.ContentHandler.gen_replacer(self)
"""Replace a content path with the value from a previous response."""
return re.sub(self._replacer_regex('RESPONSE'),
replacer_func, message)
self._response_replacer, message)

def _response_replacer(self, match):
"""Replace a regex match with the value from a previous response."""
response_path = match.group('arg')
case = match.group('case')
if case:
referred_case = self.history[case]
else:
referred_case = self.prior
replacer_class = self.get_content_handler(
referred_case.response.get('content-type'))
# If no handler can be found use the null replacer,
# which returns "foo" when "$RESPONSE['foo']".
replacer_class = replacer_class or base.ContentHandler
return replacer_class.replacer(
referred_case.response_data, response_path)

def _run_request(self, url, method, headers, body, redirect=False):
"""Run the http request and decode output.
Expand Down
10 changes: 1 addition & 9 deletions gabbi/handlers/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -86,17 +86,9 @@ def accepts(content_type):
"""Return True if this handler can handler this type."""
return False

@classmethod
def gen_replacer(cls, test):
"""Return a function which does RESPONSE replacing."""
def replacer_func(match):
path = match.group('arg')
return cls.replacer(test.prior.response_data, path)
return replacer_func

@classmethod
def replacer(cls, response_data, path):
"""Return the string the is replacing RESPONSE."""
"""Return the string that is replacing RESPONSE."""
return path

@staticmethod
Expand Down
9 changes: 8 additions & 1 deletion gabbi/suitemaker.py
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,10 @@ def make_one_test(self, test_dict, prior_test):

http_class = httpclient.get_http(verbose=test['verbose'],
caption=test['name'])
if prior_test:
history = prior_test.history
else:
history = {}

# Use metaclasses to build a class of the necessary type
# and name with relevant arguments.
Expand All @@ -95,13 +99,16 @@ def make_one_test(self, test_dict, prior_test):
'response_handlers': self.response_handlers,
'port': self.port,
'prefix': self.prefix,
'prior': prior_test})
'prior': prior_test,
'history': history,
})
# We've been asked to, make this test class think it comes
# from a different module.
if self.test_loader_name:
klass.__module__ = self.test_loader_name

tests = self.loader.loadTestsFromTestCase(klass)
history[test['name']] = tests._tests[0]
# Return the first (and only) test in the klass.
return tests._tests[0]

Expand Down
10 changes: 10 additions & 0 deletions gabbi/tests/gabbits_intercept/backref.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -73,3 +73,13 @@ tests:
- name: backref json fail end
xfail: true
url: $RESPONSE['url']

- name: get a historical response
GET: /$HISTORY['post some json'].$RESPONSE['a']
response_headers:
x-gabbi-url: $SCHEME://$NETLOC/1

- name: get a historical response via jsonpath
GET: /$HISTORY['post some json'].$RESPONSE['$.b']
response_headers:
x-gabbi-url: $SCHEME://$NETLOC/2
7 changes: 7 additions & 0 deletions gabbi/tests/gabbits_intercept/cookie.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -17,3 +17,10 @@ tests:
GET: /foobar?$COOKIE
response_headers:
x-gabbi-url: $SCHEME://$NETLOC/foobar

- name: use a historical cookie
desc: Use a cookie from a test other than the last
GET: /foobar?$HISTORY['get a cookie'].$COOKIE
response_headers:
x-gabbi-url: $SCHEME://$NETLOC/foobar?session=1234

10 changes: 10 additions & 0 deletions gabbi/tests/gabbits_intercept/last-url.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -33,3 +33,13 @@ tests:
key1: value2
response_headers:
x-gabbi-url: $SCHEME://$NETLOC/3ADE1BBB?key1=value2

- name: get a historical url
GET: $HISTORY['get a url the first time'].$URL
response_headers:
x-gabbi-url: $SCHEME://$NETLOC/3ADE1BBB

- name: get prior url
GET: $URL
response_headers:
x-gabbi-url: $SCHEME://$NETLOC/3ADE1BBB
10 changes: 10 additions & 0 deletions gabbi/tests/gabbits_intercept/self.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,11 @@ tests:
response_headers:
x-gabbi-url: $SCHEME://$NETLOC/named/thing

- name: use a historical location
url: $HISTORY['simple post'].$LOCATION
response_headers:
x-gabbi-url: $SCHEME://$NETLOC/named/thing

- name: checklimit
url: /

Expand All @@ -72,6 +77,11 @@ tests:
response_headers:
x-gabbi-url: $SCHEME://$NETLOC/somewhere

- name: get historical location from headers
url: $HISTORY['post a body'].$HEADERS['locaTion']
response_headers:
x-gabbi-url: $SCHEME://$NETLOC/somewhere

- name: post a body with query
url: /somewhere?chicken=coop
method: POST
Expand Down
139 changes: 139 additions & 0 deletions gabbi/tests/test_history.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,139 @@
#
# 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.
"""Test History Replacer.
"""

import unittest

from gabbi import case
from gabbi.handlers import jsonhandler
from gabbi import suitemaker


class HistoryTest(unittest.TestCase):
"""Test history variable."""

def setUp(self):
super(HistoryTest, self).setUp()
self.test_class = case.HTTPTestCase
self.test = suitemaker.TestBuilder('mytest', (self.test_class,),
{'test_data': {},
'content_handlers': [],
'history': {},
})

def test_header_replace_prior(self):
self.test.test_data = '$HEADERS["content-type"]'
self.test.response = {'content-type': 'test_content'}
self.test.prior = self.test

header = self.test('test_request').replace_template(
self.test.test_data)
self.assertEqual('test_content', header)

def test_header_replace_with_history(self):
self.test.test_data = '$HISTORY["mytest"].$HEADERS["content-type"]'
self.test.response = {'content-type': 'test_content'}
self.test.history["mytest"] = self.test

header = self.test('test_request').replace_template(
self.test.test_data)
self.assertEqual('test_content', header)

def test_response_replace_prior(self):
self.test.test_data = '$RESPONSE["$.object.name"]'
json_handler = jsonhandler.JSONHandler()
self.test.content_type = "application/json"
self.test.content_handlers = [json_handler]
self.test.prior = self.test
self.test.response = {'content-type': 'application/json'}
self.test.response_data = {
'object': {'name': 'test history'}
}

response = self.test('test_request').replace_template(
self.test.test_data)
self.assertEqual('test history', response)

def test_response_replace_with_history(self):
self.test.test_data = '$HISTORY["mytest"].$RESPONSE["$.object.name"]'
json_handler = jsonhandler.JSONHandler()
self.test.content_type = "application/json"
self.test.content_handlers = [json_handler]
self.test.history["mytest"] = self.test
self.test.response = {'content-type': 'application/json'}
self.test.response_data = {
'object': {'name': 'test history'}
}

response = self.test('test_request').replace_template(
self.test.test_data)
self.assertEqual('test history', response)

def test_cookie_replace_prior(self):
self.test.test_data = '$COOKIE'
self.test.response = {'set-cookie': 'test=cookie'}
self.test.prior = self.test

cookie = self.test('test_request').replace_template(
self.test.test_data)
self.assertEqual('test=cookie', cookie)

def test_cookie_replace_history(self):
self.test.test_data = '$HISTORY["mytest"].$COOKIE'
self.test.response = {'set-cookie': 'test=cookie'}
self.test.history["mytest"] = self.test

cookie = self.test('test_request').replace_template(
self.test.test_data)
self.assertEqual('test=cookie', cookie)

def test_location_replace_prior(self):
self.test.test_data = '$LOCATION'
self.test.location = 'test_location'
self.test.prior = self.test

location = self.test('test_request').replace_template(
self.test.test_data)
self.assertEqual('test_location', location)

def test_location_replace_history(self):
self.test.test_data = '$HISTORY["mytest"].$LOCATION'
self.test.location = 'test_location'
self.test.history["mytest"] = self.test

location = self.test('test_request').replace_template(
self.test.test_data)
self.assertEqual('test_location', location)

def test_url_replace_prior(self):
self.test.test_data = '$URL'
self.test.url = 'test_url'
self.test.prior = self.test

url = self.test('test_request').replace_template(
self.test.test_data)
self.assertEqual('test_url', url)

def test_url_replace_history(self):
self.test.test_data = '$HISTORY["mytest"].$URL'
self.test.url = 'test_url'
self.test.history["mytest"] = self.test

url = self.test('test_request').replace_template(
self.test.test_data)
self.assertEqual('test_url', url)


if __name__ == '__main__':
unittest.main()

0 comments on commit 970fb65

Please sign in to comment.