openstack/os-testr
Code Issues Proposed changes

Seperate regex builder logic into a seperate module

Browse Source 
The regex building logic is independently useful and will likely start
to be used in other places. This commits splits it out from the ostestr
cli and makes it an independent module that just contains the pieces
necessary for building a selection regex.

Change-Id: Ic8494d0f54357fdafd650b40219e6ad7fd5a65ad
This commit is contained in:
Matthew Treinish 2016-05-04 12:59:20 -04:00
parent a5bd26aaa8
commit 73bf72c71b
No known key found for this signature in database
GPG Key ID: FD12A0F214C9E177
5 changed files with 231 additions and 193 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

101
os_testr/os_testr.py → os_testr/ostestr.py
View File

@ -23,6 +23,8 @@ import pbr.version
from subunit import run as subunit_run
from testtools import run as testtools_run
from os_testr import regex_builder as rb
__version__ = pbr.version.VersionInfo('os_testr').version_string()
@ -95,91 +97,6 @@ def get_parser(args):
return parser.parse_known_args(args)
def _get_test_list(regex, env=None):
env = env or copy.deepcopy(os.environ)
proc = subprocess.Popen(['testr', 'list-tests', regex], env=env,
stdout=subprocess.PIPE)
out = proc.communicate()[0]
raw_test_list = out.split('\n')
bad = False
test_list = []
exclude_list = ['OS_', 'CAPTURE', 'TEST_TIMEOUT', 'PYTHON',
'subunit.run discover']
for line in raw_test_list:
for exclude in exclude_list:
if exclude in line:
bad = True
break
elif not line:
bad = True
break
if not bad:
test_list.append(line)
bad = False
return test_list
def print_skips(regex, message):
test_list = _get_test_list(regex)
if test_list:
if message:
print(message)
else:
print('Skipped because of regex %s:' % regex)
for test in test_list:
print(test)
# Extra whitespace to separate
print('\n')
def path_to_regex(path):
root, _ = os.path.splitext(path)
return root.replace('/', '.')
def get_regex_from_whitelist_file(file_path):
lines = []
for line in open(file_path).read().splitlines():
split_line = line.strip().split('#')
# Before the # is the regex
line_regex = split_line[0].strip()
if line_regex:
lines.append(line_regex)
return '|'.join(lines)
def construct_regex(blacklist_file, whitelist_file, regex, print_exclude):
if not blacklist_file:
exclude_regex = ''
else:
black_file = open(blacklist_file, 'r')
exclude_regex = ''
for line in black_file:
raw_line = line.strip()
split_line = raw_line.split('#')
# Before the # is the regex
line_regex = split_line[0].strip()
if len(split_line) > 1:
# After the # is a comment
comment = split_line[1].strip()
else:
comment = ''
if line_regex:
if print_exclude:
print_skips(line_regex, comment)
if exclude_regex:
exclude_regex = '|'.join([line_regex, exclude_regex])
else:
exclude_regex = line_regex
if exclude_regex:
exclude_regex = "^((?!" + exclude_regex + ").)*$"
if regex:
exclude_regex += regex
if whitelist_file:
exclude_regex += '%s' % get_regex_from_whitelist_file(whitelist_file)
return exclude_regex
def call_testr(regex, subunit, pretty, list_tests, slowest, parallel, concur,
until_failure, color, others=None):
others = others or []
@ -206,7 +123,7 @@ def call_testr(regex, subunit, pretty, list_tests, slowest, parallel, concur,
# This workaround is necessary because of lp bug 1411804 it's super hacky
# and makes tons of unfounded assumptions, but it works for the most part
if (subunit or pretty) and until_failure:
test_list = _get_test_list(regex, env)
test_list = rb._get_test_list(regex, env)
count = 0
failed = False
if not test_list:
@ -292,7 +209,7 @@ def _select_and_call_runner(opts, exclude_regex, others):
return 2
test_to_run = opts.no_discover or opts.pdb
if test_to_run.find('/') != -1:
test_to_run = path_to_regex(test_to_run)
test_to_run = rb.path_to_regex(test_to_run)
ec = call_subunit_run(test_to_run, opts.pretty, opts.subunit)
return ec
@ -318,13 +235,13 @@ def main():
print(msg)
exit(5)
if opts.path:
regex = path_to_regex(opts.path)
regex = rb.path_to_regex(opts.path)
else:
regex = opts.regex
exclude_regex = construct_regex(opts.blacklist_file,
opts.whitelist_file,
regex,
opts.print_exclude)
exclude_regex = rb.construct_regex(opts.blacklist_file,
opts.whitelist_file,
regex,
opts.print_exclude)
exit(_select_and_call_runner(opts, exclude_regex, others))
if __name__ == '__main__':

102
os_testr/regex_builder.py Normal file
View File

@ -0,0 +1,102 @@
# Copyright 2016 Hewlett-Packard Development Company, L.P.
#
# 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.
import copy
import os
import subprocess
def _get_test_list(regex, env=None):
env = env or copy.deepcopy(os.environ)
proc = subprocess.Popen(['testr', 'list-tests', regex], env=env,
stdout=subprocess.PIPE)
out = proc.communicate()[0]
raw_test_list = out.split('\n')
bad = False
test_list = []
exclude_list = ['OS_', 'CAPTURE', 'TEST_TIMEOUT', 'PYTHON',
'subunit.run discover']
for line in raw_test_list:
for exclude in exclude_list:
if exclude in line:
bad = True
break
elif not line:
bad = True
break
if not bad:
test_list.append(line)
bad = False
return test_list
def print_skips(regex, message):
test_list = _get_test_list(regex)
if test_list:
if message:
print(message)
else:
print('Skipped because of regex %s:' % regex)
for test in test_list:
print(test)
# Extra whitespace to separate
print('\n')
def path_to_regex(path):
root, _ = os.path.splitext(path)
return root.replace('/', '.')
def get_regex_from_whitelist_file(file_path):
lines = []
for line in open(file_path).read().splitlines():
split_line = line.strip().split('#')
# Before the # is the regex
line_regex = split_line[0].strip()
if line_regex:
lines.append(line_regex)
return '|'.join(lines)
def construct_regex(blacklist_file, whitelist_file, regex, print_exclude):
if not blacklist_file:
exclude_regex = ''
else:
black_file = open(blacklist_file, 'r')
exclude_regex = ''
for line in black_file:
raw_line = line.strip()
split_line = raw_line.split('#')
# Before the # is the regex
line_regex = split_line[0].strip()
if len(split_line) > 1:
# After the # is a comment
comment = split_line[1].strip()
else:
comment = ''
if line_regex:
if print_exclude:
print_skips(line_regex, comment)
if exclude_regex:
exclude_regex = '|'.join([line_regex, exclude_regex])
else:
exclude_regex = line_regex
if exclude_regex:
exclude_regex = "^((?!" + exclude_regex + ").)*$"
if regex:
exclude_regex += regex
if whitelist_file:
exclude_regex += '%s' % get_regex_from_whitelist_file(whitelist_file)
return exclude_regex

117
os_testr/tests/test_ostestr.py Normal file
View File

@ -0,0 +1,117 @@
# -*- coding: utf-8 -*-
# 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_os_testr
----------------------------------
Tests for `os_testr` module.
"""
import mock
from os_testr import ostestr as os_testr
from os_testr.tests import base
class TestGetParser(base.TestCase):
def test_pretty(self):
namespace = os_testr.get_parser(['--pretty'])
self.assertEqual(True, namespace[0].pretty)
namespace = os_testr.get_parser(['--no-pretty'])
self.assertEqual(False, namespace[0].pretty)
self.assertRaises(SystemExit, os_testr.get_parser,
['--no-pretty', '--pretty'])
def test_slowest(self):
namespace = os_testr.get_parser(['--slowest'])
self.assertEqual(True, namespace[0].slowest)
namespace = os_testr.get_parser(['--no-slowest'])
self.assertEqual(False, namespace[0].slowest)
self.assertRaises(SystemExit, os_testr.get_parser,
['--no-slowest', '--slowest'])
def test_parallel(self):
namespace = os_testr.get_parser(['--parallel'])
self.assertEqual(True, namespace[0].parallel)
namespace = os_testr.get_parser(['--serial'])
self.assertEqual(False, namespace[0].parallel)
self.assertRaises(SystemExit, os_testr.get_parser,
['--parallel', '--serial'])
class TestCallers(base.TestCase):
def test_no_discover(self):
namespace = os_testr.get_parser(['-n', 'project.tests.foo'])
def _fake_exit(arg):
self.assertTrue(arg)
def _fake_run(*args, **kwargs):
return 'project.tests.foo' in args
with mock.patch.object(os_testr, 'exit', side_effect=_fake_exit), \
mock.patch.object(os_testr, 'get_parser', return_value=namespace), \
mock.patch.object(os_testr,
'call_subunit_run',
side_effect=_fake_run):
os_testr.main()
def test_no_discover_path(self):
namespace = os_testr.get_parser(['-n', 'project/tests/foo'])
def _fake_exit(arg):
self.assertTrue(arg)
def _fake_run(*args, **kwargs):
return 'project.tests.foo' in args
with mock.patch.object(os_testr, 'exit', side_effect=_fake_exit), \
mock.patch.object(os_testr, 'get_parser', return_value=namespace), \
mock.patch.object(os_testr,
'call_subunit_run',
side_effect=_fake_run):
os_testr.main()
def test_pdb(self):
namespace = os_testr.get_parser(['--pdb', 'project.tests.foo'])
def _fake_exit(arg):
self.assertTrue(arg)
def _fake_run(*args, **kwargs):
return 'project.tests.foo' in args
with mock.patch.object(os_testr, 'exit', side_effect=_fake_exit), \
mock.patch.object(os_testr, 'get_parser', return_value=namespace), \
mock.patch.object(os_testr,
'call_subunit_run',
side_effect=_fake_run):
os_testr.main()
def test_pdb_path(self):
namespace = os_testr.get_parser(['--pdb', 'project/tests/foo'])
def _fake_exit(arg):
self.assertTrue(arg)
def _fake_run(*args, **kwargs):
return 'project.tests.foo' in args
with mock.patch.object(os_testr, 'exit', side_effect=_fake_exit), \
mock.patch.object(os_testr, 'get_parser', return_value=namespace), \
mock.patch.object(os_testr,
'call_subunit_run',
side_effect=_fake_run):
os_testr.main()

102
os_testr/tests/test_os_testr.py → os_testr/tests/test_regex_builder.py
View File

@ -12,17 +12,11 @@
# License for the specific language governing permissions and limitations
# under the License.
"""
test_os_testr
----------------------------------
Tests for `os_testr` module.
"""
import mock
import six
from os_testr import os_testr
from os_testr import regex_builder as os_testr
from os_testr.tests import base
@ -35,98 +29,6 @@ class TestPathToRegex(base.TestCase):
self.assertEqual("openstack.tests.network.v2", result)
class TestGetParser(base.TestCase):
def test_pretty(self):
namespace = os_testr.get_parser(['--pretty'])
self.assertEqual(True, namespace[0].pretty)
namespace = os_testr.get_parser(['--no-pretty'])
self.assertEqual(False, namespace[0].pretty)
self.assertRaises(SystemExit, os_testr.get_parser,
['--no-pretty', '--pretty'])
def test_slowest(self):
namespace = os_testr.get_parser(['--slowest'])
self.assertEqual(True, namespace[0].slowest)
namespace = os_testr.get_parser(['--no-slowest'])
self.assertEqual(False, namespace[0].slowest)
self.assertRaises(SystemExit, os_testr.get_parser,
['--no-slowest', '--slowest'])
def test_parallel(self):
namespace = os_testr.get_parser(['--parallel'])
self.assertEqual(True, namespace[0].parallel)
namespace = os_testr.get_parser(['--serial'])
self.assertEqual(False, namespace[0].parallel)
self.assertRaises(SystemExit, os_testr.get_parser,
['--parallel', '--serial'])
class TestCallers(base.TestCase):
def test_no_discover(self):
namespace = os_testr.get_parser(['-n', 'project.tests.foo'])
def _fake_exit(arg):
self.assertTrue(arg)
def _fake_run(*args, **kwargs):
return 'project.tests.foo' in args
with mock.patch.object(os_testr, 'exit', side_effect=_fake_exit), \
mock.patch.object(os_testr, 'get_parser', return_value=namespace), \
mock.patch.object(os_testr,
'call_subunit_run',
side_effect=_fake_run):
os_testr.main()
def test_no_discover_path(self):
namespace = os_testr.get_parser(['-n', 'project/tests/foo'])
def _fake_exit(arg):
self.assertTrue(arg)
def _fake_run(*args, **kwargs):
return 'project.tests.foo' in args
with mock.patch.object(os_testr, 'exit', side_effect=_fake_exit), \
mock.patch.object(os_testr, 'get_parser', return_value=namespace), \
mock.patch.object(os_testr,
'call_subunit_run',
side_effect=_fake_run):
os_testr.main()
def test_pdb(self):
namespace = os_testr.get_parser(['--pdb', 'project.tests.foo'])
def _fake_exit(arg):
self.assertTrue(arg)
def _fake_run(*args, **kwargs):
return 'project.tests.foo' in args
with mock.patch.object(os_testr, 'exit', side_effect=_fake_exit), \
mock.patch.object(os_testr, 'get_parser', return_value=namespace), \
mock.patch.object(os_testr,
'call_subunit_run',
side_effect=_fake_run):
os_testr.main()
def test_pdb_path(self):
namespace = os_testr.get_parser(['--pdb', 'project/tests/foo'])
def _fake_exit(arg):
self.assertTrue(arg)
def _fake_run(*args, **kwargs):
return 'project.tests.foo' in args
with mock.patch.object(os_testr, 'exit', side_effect=_fake_exit), \
mock.patch.object(os_testr, 'get_parser', return_value=namespace), \
mock.patch.object(os_testr,
'call_subunit_run',
side_effect=_fake_run):
os_testr.main()
class TestConstructRegex(base.TestCase):
def test_regex_passthrough(self):
result = os_testr.construct_regex(None, None, 'fake_regex', False)

2
setup.cfg
View File

@ -25,7 +25,7 @@ packages =
[entry_points]
console_scripts =
subunit-trace = os_testr.subunit_trace:main
ostestr = os_testr.os_testr:main
ostestr = os_testr.ostestr:main
subunit2html = os_testr.subunit2html:main
generate-subunit = os_testr.generate_subunit:main