openstack/deb-python-gabbi
Code Issues Proposed changes

Runner cleanup (#141)

Browse Source 
Extract methods from the run() method in runner.py to make the
processing of information easier to test and easier to assemble.
Most methods were kept in runner.py but loading yaml was
extracted to utils.py where it can be reused by driver.py.
This commit is contained in:
Chris Dent 2016-06-06 11:42:05 +01:00
parent 2013e4ed74
commit e2964930a9
4 changed files with 93 additions and 62 deletions

12
gabbi/driver.py

@ -25,18 +25,16 @@ An entire directory of YAML files is a TestSuite of TestSuites.
import glob
import inspect
import io
import os
import unittest
from unittest import suite
import uuid
import yaml
from gabbi import case
from gabbi import handlers
from gabbi import reporter
from gabbi import suitemaker
from gabbi import utils
def build_tests(path, loader, host=None, port=8001, intercept=None,
@ -79,7 +77,7 @@ def build_tests(path, loader, host=None, port=8001, intercept=None,
for test_file in glob.iglob('%s/*.yaml' % path):
if intercept:
host = str(uuid.uuid4())
suite_dict = load_yaml(test_file)
suite_dict = utils.load_yaml(yaml_file=test_file)
test_base_name = '%s_%s' % (
test_loader_name, os.path.splitext(os.path.basename(test_file))[0])
@ -124,12 +122,6 @@ def py_test_generator(test_dir, host=None, port=8001, intercept=None,
yield 'stop_%s' % test._tests[0].__class__.__name__, test.stop
def load_yaml(yaml_file):
"""Read and parse any YAML file. Let exceptions flow where they may."""
with io.open(yaml_file, encoding='utf-8') as source:
return yaml.safe_load(source.read())
def test_suite_from_yaml(loader, test_base_name, test_yaml, test_directory,
host, port, fixture_module, intercept, prefix=''):
"""Legacy wrapper retained for backwards compatibility."""

70
gabbi/runner.py

@ -18,12 +18,12 @@ import sys
import unittest
from six.moves.urllib import parse as urlparse
import yaml
from gabbi import case
from gabbi import handlers
from gabbi.reporter import ConciseTestRunner
from gabbi import suitemaker
from gabbi import utils
def run():
@ -91,39 +91,15 @@ def run():
help='Custom response handler. Should be an import path of the '
'form package.module or package.module:class.'
)
args = parser.parse_args()
force_ssl = False
split_url = urlparse.urlsplit(args.target)
if split_url.scheme:
target = split_url.netloc
prefix = split_url.path
if split_url.scheme == 'https':
force_ssl = True
else:
target = args.target
prefix = args.prefix
if ':' in target and '[' not in target:
host, port = target.rsplit(':', 1)
elif ']:' in target:
host, port = target.rsplit(':', 1)
else:
host = target
port = None
host = host.replace('[', '').replace(']', '')
host, port, prefix, force_ssl = process_target_args(
args.target, args.prefix)
# Initialize response handlers.
custom_response_handlers = []
for import_path in args.response_handlers or []:
for handler in load_response_handlers(import_path):
custom_response_handlers.append(handler)
for handler in handlers.RESPONSE_HANDLERS + custom_response_handlers:
handler(case.HTTPTestCase)
initialize_handlers(args.response_handlers)
data = yaml.safe_load(sys.stdin.read())
# Only override the default if we are forcing a change, there may
# already be a default.
data = utils.load_yaml(handle=sys.stdin)
if force_ssl:
if 'defaults' in data:
data['defaults']['ssl'] = True
@ -137,6 +113,40 @@ def run():
sys.exit(not result.wasSuccessful())
def process_target_args(target, prefix):
"""Turn the argparse args into a host, port and prefix."""
force_ssl = False
split_url = urlparse.urlparse(target)
if split_url.scheme:
if split_url.scheme == 'https':
force_ssl = True
return split_url.hostname, split_url.port, split_url.path, force_ssl
else:
target = target
prefix = prefix
if ':' in target and '[' not in target:
host, port = target.rsplit(':', 1)
elif ']:' in target:
host, port = target.rsplit(':', 1)
else:
host = target
port = None
host = host.replace('[', '').replace(']', '')
return host, port, prefix, force_ssl
def initialize_handlers(response_handlers):
custom_response_handlers = []
for import_path in response_handlers or []:
for handler in load_response_handlers(import_path):
custom_response_handlers.append(handler)
for handler in handlers.RESPONSE_HANDLERS + custom_response_handlers:
handler(case.HTTPTestCase)
def load_response_handlers(import_path):
"""Load and return custom response handlers from the given Python package
or module.

55
gabbi/tests/test_runner.py

@ -14,7 +14,6 @@
"""
import copy
import mock
import sys
import unittest
from uuid import uuid4
@ -232,38 +231,50 @@ class RunnerTest(unittest.TestCase):
class RunnerHostArgParse(unittest.TestCase):
@mock.patch('sys.exit')
@mock.patch('sys.stdin')
@mock.patch('gabbi.suitemaker.test_suite_from_dict')
@mock.patch('yaml.safe_load', return_value={})
def _test_hostport(self, url_or_host, expected_host,
portmock_yaml, mock_test_suite, mock_read, mock_exit,
provided_prefix=None, expected_port=None,
expected_prefix=None, expected_data=None):
sys.argv = ['gabbi-run', url_or_host]
if provided_prefix:
sys.argv.append(provided_prefix)
runner.run()
expected_prefix=None, expected_ssl=False):
host, port, prefix, ssl = runner.process_target_args(
url_or_host, provided_prefix)
expected_data = expected_data or {}
# normalize hosts, they are case insensitive
self.assertEqual(expected_host.lower(), host.lower())
# port can be a string or int depending on the inputs
self.assertEqual(expected_port, port)
self.assertEqual(expected_prefix, prefix)
self.assertEqual(expected_ssl, ssl)
mock_test_suite.assert_called_with(
unittest.defaultTestLoader, 'input', expected_data,
'.', expected_host, expected_port, None, None,
prefix=expected_prefix
)
def test_plain_url_no_port(self):
self._test_hostport('http://foobar.com/news',
'foobar.com',
expected_port=None,
expected_prefix='/news')
def test_plain_url(self):
def test_plain_url_with_port(self):
self._test_hostport('http://foobar.com:80/news',
'foobar.com',
expected_port='80',
expected_port=80,
expected_prefix='/news')
def test_ssl_url(self):
self._test_hostport('https://foobar.com/news',
'foobar.com',
expected_prefix='/news',
expected_data={'defaults': {'ssl': True}})
expected_ssl=True)
def test_ssl_port80_url(self):
self._test_hostport('https://foobar.com:80/news',
'foobar.com',
expected_prefix='/news',
expected_port=80,
expected_ssl=True)
def test_ssl_port_url(self):
self._test_hostport('https://foobar.com:999/news',
'foobar.com',
expected_prefix='/news',
expected_port=999,
expected_ssl=True)
def test_simple_hostport(self):
self._test_hostport('foobar.com:999',
@ -281,14 +292,14 @@ class RunnerHostArgParse(unittest.TestCase):
self._test_hostport(
'http://[FEDC:BA98:7654:3210:FEDC:BA98:7654:3210]:999/news',
'FEDC:BA98:7654:3210:FEDC:BA98:7654:3210',
expected_port='999',
expected_port=999,
expected_prefix='/news')
def test_ipv6_url_localhost(self):
self._test_hostport(
'http://[::1]:999/news',
'::1',
expected_port='999',
expected_port=999,
expected_prefix='/news')
def test_ipv6_host_localhost(self):

18
gabbi/utils.py

@ -12,8 +12,11 @@
# under the License.
"""Utility functions grab bag."""
import io
import os
import yaml
try: # Python 3
ConnectionRefused = ConnectionRefusedError
@ -99,6 +102,21 @@ def get_colorizer(stream):
return lambda x, y: y
def load_yaml(handle=None, yaml_file=None):
"""Read and parse any YAML file or filehandle.
Let exceptions flow where they may.
If no file or handle is provided, read from STDIN.
"""
if yaml_file:
with io.open(yaml_file, encoding='utf-8') as source:
return yaml.safe_load(source.read())
# This will intentionally raise AttributeError if handle is None.
return yaml.safe_load(handle.read())
def not_binary(content_type):
"""Decide if something is content we'd like to treat as a string."""
return (content_type.startswith('text/') or