x
/
tobiko
Code Issues Proposed changes

Remove last dependency from stestr 

This also removes test cases discovery function
and tobiko-fixture command that where using it

Change-Id: I8ec1cde0b5ada0e63a213aa09a5a877d432bbd73
changes/51/772651/6
Federico Ressi 2021-01-27 12:09:56 +01:00
parent 406240730a
commit 8d66126f1b
10 changed files with 0 additions and 678 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
requirements.txt
View File

@ -18,5 +18,4 @@ python-octaviaclient>=2.2.0 # Apache-2.0
python-openstackclient>=5.4.0 # Apache-2.0
six>=1.15.0 # MIT
sshtunnel>=0.3.1 # MIT
stestr>=3.1.0 # Apache-2.0
testtools>=2.4.0 # MIT

1
tobiko/__init__.py
View File

@ -110,7 +110,6 @@ skip_unless = _skip.skip_unless
skip = _skip.skip
assert_test_case_was_skipped = _testcase.assert_test_case_was_skipped
discover_test_cases = _testcase.discover_test_cases
get_test_case = _testcase.get_test_case
pop_test_case = _testcase.pop_test_case
push_test_case = _testcase.push_test_case

0
tobiko/cmd/__init__.py
View File

52
tobiko/cmd/base.py
View File

@ -1,52 +0,0 @@
# Copyright (c) 2018 Red Hat
# All Rights Reserved.
#
# 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.
from __future__ import absolute_import
import logging
import argparse
from oslo_log import log
from tobiko import config
LOG = log.getLogger(__name__)
class TobikoCMD(object):
"""Manages different command line utilities."""
def __init__(self):
config.CONF.tobiko.use_stderr = True
log.setup(config.CONF.tobiko, 'tobiko')
self.parser = self.get_parser()
self.args = (self.parser).parse_args()
def get_parser(self):
parser = argparse.ArgumentParser(add_help=True)
parser.add_argument('--verbose', '-v', action='count',
help='Make the output more verbose, incremental.')
parser.add_argument('--quiet', '-q', action='count',
help='Make the output less verbose, incremental.')
return parser
def set_stream_handler_logging_level(self):
num_quiet = self.args.quiet or 0
num_verb = self.args.verbose or 0
level = logging.WARNING - (num_verb * 10) + (num_quiet * 10)
root_logger = logging.getLogger()
for handler in root_logger.handlers:
if isinstance(handler, logging.StreamHandler):
handler.setLevel(level)

182
tobiko/cmd/fixture.py
View File

@ -1,182 +0,0 @@
# Copyright 2018 Red Hat
#
# 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.
from __future__ import absolute_import
import sys
from oslo_log import log
import six
import tobiko
from tobiko.cmd import base
LOG = log.getLogger(__name__)
class FixtureUtil(base.TobikoCMD):
def get_parser(self):
parser = super(FixtureUtil, self).get_parser()
subparsers = parser.add_subparsers(
title='subcommands', description='fixture operation')
subcommand_parser = subparsers.add_parser(
'help', help=('show this help message'))
subcommand_parser.set_defaults(subcommand='help')
for subcommand_name in ['cleanup', 'list', 'setup']:
subcommand_parser = subparsers.add_parser(
subcommand_name, help=(subcommand_name + ' fixtures'))
subcommand_parser.set_defaults(subcommand=subcommand_name)
subcommand_parser.add_argument(
'--config', '-c',
default='.stestr.conf',
help=("Set a stestr config file to use with this command. "
"If one isn't specified then .stestr.conf in the "
"directory that a command is running from is used"))
subcommand_parser.add_argument(
'--repo-type', '-r',
choices=['file'],
default='file',
help=("Select the repo backend to use"))
subcommand_parser.add_argument(
'--repo-url', '-u',
help=("Set the repo url to use. An acceptable value for "
"this depends on the repository type used."))
subcommand_parser.add_argument(
'--test-path', '-t',
help=("Set the test path to use for unittest discovery. If "
"both this and the corresponding config file option "
"are set, this value will be used."))
subcommand_parser.add_argument(
'--top-dir',
help=("Set the top dir to use for unittest discovery. If "
"both this and the corresponding config file option "
"are set, this value will be used."))
subcommand_parser.add_argument(
'--group-regex', '--group_regex', '-g',
help=("Set a group regex to use for grouping tests together "
"in the stestr scheduler. If both this and the "
"corresponding config file option are set this value "
"will be used."))
subcommand_parser.add_argument(
'--blacklist-file', '-b',
help=("Path to a blacklist file, this file contains a "
"separate regex exclude on each newline."))
subcommand_parser.add_argument(
'--whitelist-file', '-w',
help=("Path to a whitelist file, this file contains a "
"separate regex on each newline."))
subcommand_parser.add_argument(
'--black-regex', '-B',
help=("Test rejection regex. If a test cases name matches "
"on re.search() operation , it will be removed from "
"the final test list. Effectively the black-regexp is "
"added to black regexp list, but you do need to edit "
"a file. The black filtering happens after the "
"initial white selection, which by default is "
"everything."))
subcommand_parser.add_argument(
'filters',
nargs='*',
help=("A list of string regex filters to initially apply "
"on the test list. Tests that match any of the "
"regexes will be used. (assuming any other filtering "
"specified also uses it)."))
return parser
def execute(self):
action = self.args.subcommand or 'list'
if action == 'help':
return self.show_help()
if action == 'list':
return self.list_fixtures()
elif action == 'setup':
return self.setup_fixtures()
elif action == 'cleanup':
return self.cleanup_fixtures()
def show_help(self, stream=None):
stream = stream or sys.stdout
self.parser.print_help(stream)
def discover_testcases(self):
args = self.args
return tobiko.discover_test_cases(
config=args.config,
repo_type=args.repo_type,
repo_url=args.repo_url,
test_path=args.test_path,
top_dir=args.top_dir,
group_regex=args.group_regex,
blacklist_file=args.blacklist_file,
whitelist_file=args.whitelist_file,
black_regex=args.black_regex,
filters=args.filters)
def list_fixtures(self, stream=None):
stream = stream or sys.stdout
test_cases = self.discover_testcases()
fixtures_names = tobiko.list_required_fixtures(test_cases)
if fixtures_names:
output = '\n'.join(fixtures_names) + '\n'
if six.PY2:
output = output.decode()
stream.write(output)
def setup_fixtures(self, stream=None):
stream = stream or sys.stdout
test_cases = self.discover_testcases()
success = True
for fixture_name in tobiko.list_required_fixtures(test_cases):
output = fixture_name + '\n'
if six.PY2:
output = output.decode()
stream.write(output)
try:
tobiko.setup_fixture(fixture_name)
except Exception:
LOG.exception('Fixture %r setup failed', fixture_name)
success = False
if not success:
sys.exit(1)
def cleanup_fixtures(self, stream=None):
stream = stream or sys.stdout
test_cases = self.discover_testcases()
success = True
for fixture_name in tobiko.list_required_fixtures(test_cases):
output = fixture_name + '\n'
if six.PY2:
output = output.decode()
stream.write(output)
try:
tobiko.cleanup_fixture(fixture_name)
except Exception:
LOG.exception('Fixture %r cleanup failed', fixture_name)
success = False
if not success:
sys.exit(1)
def main():
"""Create CLI main entry."""
fixture_util = FixtureUtil()
fixture_util.set_stream_handler_logging_level()
fixture_util.execute()
if __name__ == '__main__':
sys.exit(main())

99
tobiko/common/_testcase.py
View File

@ -25,105 +25,6 @@ from tobiko.common import _exception
os.environ.setdefault('PYTHON', sys.executable)
class TestCasesFinder(object):
def __init__(self, config=None, repo_type=None, repo_url=None,
test_path=None, top_dir=None, group_regex=None,
blacklist_file=None, whitelist_file=None, black_regex=None,
filters=None):
"""
:param str config: The path to the stestr config file. Must be a
string.
:param str repo_type: This is the type of repository to use. Valid
choices are 'file' and 'sql'.
:param str repo_url: The url of the repository to use.
:param str test_path: Set the test path to use for unittest discovery.
If both this and the corresponding config file option are set, this
value will be used.
:param str top_dir: The top dir to use for unittest discovery. This
takes precedence over the value in the config file. (if one is
present in the config file)
:param str group_regex: Set a group regex to use for grouping tests
together in the stestr scheduler. If both this and the
corresponding config file option are set this value will be used.
:param str blacklist_file: Path to a blacklist file, this file contains
a separate regex exclude on each newline.
:param str whitelist_file: Path to a whitelist file, this file contains
a separate regex on each newline.
:param str black_regex: Test rejection regex. If a test cases name
matches on re.search() operation, it will be removed from the final
test list.
:param list filters: A list of string regex filters to initially apply
on the test list. Tests that match any of the regexes will be used.
(assuming any other filtering specified also uses it)
"""
self.config = config or '.stestr.conf'
self.repo_type = repo_type or 'file'
self.repo_url = repo_url
self.test_path = test_path
self.top_dir = top_dir
self.group_regex = group_regex
self.blacklist_file = blacklist_file
self.whitelist_file = whitelist_file
self.black_regex = black_regex
self.filters = filters
def discover_test_cases(self, **kwargs):
"""Iterate over test_ids for a project
This method will print the test_ids for tests in a project. You can
filter the output just like with the run command to see exactly what
will be run.
"""
from stestr import config_file
params = dict(config=self.config, repo_type=self.repo_type,
repo_url=self.repo_url, test_path=self.test_path,
top_dir=self.top_dir, group_regex=self.group_regex,
blacklist_file=self.blacklist_file,
whitelist_file=self.whitelist_file,
black_regex=self.black_regex, filters=self.filters)
if kwargs:
params.update(kwargs)
ids = None
config = params.pop('config')
conf = config_file.TestrConf(config)
filters = params.pop('filters')
blacklist_file = params.pop('blacklist_file')
whitelist_file = params.pop('whitelist_file')
black_regex = params.pop('black_regex')
cmd = conf.get_run_command(
regexes=filters, repo_type=params['repo_type'],
repo_url=params['repo_url'], group_regex=params['group_regex'],
blacklist_file=blacklist_file, whitelist_file=whitelist_file,
black_regex=black_regex, test_path=params['test_path'],
top_dir=params['top_dir'])
not_filtered = filters is None and blacklist_file is None\
and whitelist_file is None and black_regex is None
try:
cmd.setUp()
# List tests if the fixture has not already needed to to filter.
if not_filtered:
ids = cmd.list_tests()
else:
ids = cmd.test_ids
except SystemExit as ex:
raise RuntimeError("Error discovering test cases IDs with "
f"parameters: {params}") from ex
finally:
cmd.cleanUp()
return sorted(ids)
FINDER = TestCasesFinder()
def discover_test_cases(finder=FINDER, **kwargs):
return finder.discover_test_cases(**kwargs)
class TestCasesManager(object):
def __init__(self):

0
tobiko/tests/unit/commands/__init__.py
View File

33
tobiko/tests/unit/commands/test_base.py
View File

@ -1,33 +0,0 @@
# Copyright (c) 2018 Red Hat
# All Rights Reserved.
#
# 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.
from __future__ import absolute_import
import sys
from tobiko.cmd import base
from tobiko.tests.unit import openstack
class TobikoCMDTest(openstack.OpenstackTest):
command_name = 'tobiko-base'
command_class = base.TobikoCMD
def test_init(self, argv=None):
self.patch_argv(argv=argv)
return self.command_class()
def patch_argv(self, argv=None):
return self.patch(sys, 'argv', [self.command_name] + (argv or []))

273
tobiko/tests/unit/commands/test_fixture.py
View File

@ -1,273 +0,0 @@
# Copyright (c) 2018 Red Hat
# All Rights Reserved.
#
# 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.
from __future__ import absolute_import
import argparse
import io
import os
import subprocess
import sys
import tobiko
from tobiko.cmd import fixture as _fixture
from tobiko.tests.unit import test_fixture
from tobiko.tests import unit
MyFixture = test_fixture.MyFixture
MyFixture2 = test_fixture.MyFixture2
MyRequiredFixture = test_fixture.MyRequiredFixture
canonical_name = test_fixture.canonical_name
class ExitCalled(Exception):
pass
class FixtureUtilTest(unit.TobikoUnitTest):
command_name = 'tobiko-fixture'
command_class = _fixture.FixtureUtil
test_path = os.path.dirname(__file__)
top_dir = os.path.dirname(os.path.dirname(tobiko.__file__))
required_fixture = tobiko.required_fixture(MyRequiredFixture)
def setUp(self):
super(FixtureUtilTest, self).setUp()
self.mock_error = self.patch(argparse.ArgumentParser, 'error',
side_effect=self.fail)
def patch_argv(self, subcommand=None, arguments=None, filters=None,
**options):
subcommand = subcommand or 'list'
arguments = arguments or []
if options:
arguments += make_options(**options)
if filters:
arguments += list(filters)
argv = [self.command_name, subcommand] + arguments
return self.patch(sys, 'argv', argv)
def test_init(self, subcommand=None, arguments=None, filters=None,
config_file=None, repo_type=None,
repo_url=None, test_path=None, top_dir=None,
group_regex=None, blacklist_file=None,
whitelist_file=None, black_regex=None):
self.patch_argv(subcommand=subcommand, arguments=arguments)
command = self.command_class()
self.mock_error.assert_not_called()
args = command.args
self.assertIsNotNone(args)
self.assertEqual(filters or [], args.filters)
self.assertEqual(config_file or '.stestr.conf', args.config)
self.assertEqual(subcommand or 'list', args.subcommand)
self.assertEqual(repo_type or 'file', args.repo_type)
self.assertEqual(repo_url, args.repo_url)
self.assertEqual(test_path, args.test_path)
self.assertEqual(top_dir, args.top_dir)
self.assertEqual(group_regex, args.group_regex)
self.assertEqual(blacklist_file, args.blacklist_file)
self.assertEqual(whitelist_file, whitelist_file)
self.assertEqual(black_regex, args.black_regex)
def test_init_with_list(self):
self.test_init(subcommand='list')
def test_init_with_seutp(self):
self.test_init(subcommand='setup')
def test_init_with_cleanup(self):
self.test_init(subcommand='cleanup')
def test_init_with_c(self):
self.test_init(arguments=['-c', 'some-config-file'],
config_file='some-config-file')
def test_init_with_config(self):
self.test_init(arguments=['--config', 'some-config-file'],
config_file='some-config-file')
def test_init_with_r(self):
self.test_init(arguments=['-r', 'file'], repo_type='file')
def test_init_with_repo_type_file(self):
self.test_init(arguments=['--repo-type', 'file'], repo_type='file')
def test_init_with_repo_type_sql(self):
self.assertRaises(self.failureException, self.test_init,
arguments=['--repo-type', 'sql'])
def test_init_with_u(self):
self.test_init(arguments=['-u', 'some-url'], repo_url='some-url')
def test_init_with_repo_url(self):
self.test_init(arguments=['--repo-url', 'some-url'],
repo_url='some-url')
def test_init_with_filters(self):
self.test_init(arguments=['a', 'b', 'c'], filters=['a', 'b', 'c'])
def test_init_with_t(self):
self.test_init(arguments=['-t', 'some/test/path'],
test_path='some/test/path')
def test_init_with_test_path(self):
self.test_init(arguments=['--test-path', 'some/test/path'],
test_path='some/test/path')
def test_init_with_top_dir(self):
self.test_init(arguments=['--top-dir', 'some/top/dir'],
top_dir='some/top/dir')
def test_init_with_g(self):
self.test_init(arguments=['-g', 'some-regex'],
group_regex='some-regex')
def test_init_with_group_regex(self):
self.test_init(arguments=['--group-regex', 'some-regex'],
group_regex='some-regex')
def test_init_with_group_regex2(self):
self.test_init(arguments=['--group_regex', 'some-regex'],
group_regex='some-regex')
def test_init_with_b(self):
self.test_init(arguments=['-b', 'some/blacklist-file'],
blacklist_file='some/blacklist-file')
def test_init_with_blacklist_file(self):
self.test_init(arguments=['--blacklist-file', 'some/blacklist-file'],
blacklist_file='some/blacklist-file')
def test_init_with_w(self):
self.test_init(arguments=['-w', 'some/whitelist-file'],
whitelist_file='some/whitelist-file')
def test_init_with_whitelist_file(self):
self.test_init(arguments=['--whitelist-file', 'some/whitelist-file'],
whitelist_file='some/whitelist-file')
def test_init_with_B(self):
self.test_init(arguments=['-B', 'some-black-regex'],
black_regex='some-black-regex')
def test_init_with_blacklist_regex(self):
self.test_init(arguments=['--black-regex', 'some-black-regex'],
black_regex='some-black-regex')
def test_main(self, subcommand=None, config_file=None, repo_type=None,
repo_url=None, test_path=None, top_dir=None,
group_regex=None, blacklist_file=None, whitelist_file=None,
black_regex=None, filters=None):
test_path = test_path or self.test_path
top_dir = top_dir or self.top_dir
self.setup_file_repo(top_dir=top_dir)
self.patch_argv(subcommand=subcommand, config_file=config_file,
repo_type=repo_type, repo_url=repo_url,
test_path=test_path, top_dir=top_dir,
group_regex=group_regex,
blacklist_file=blacklist_file,
whitelist_file=whitelist_file,
black_regex=black_regex, filters=filters)
stdout = self.patch(sys, 'stdout', io.StringIO())
_fixture.main()
self.mock_error.assert_not_called()
return stdout
def test_list(self, fixture1=MyFixture, fixture2=MyFixture2):
stdout = self.test_main(subcommand='list')
written_lines = stdout.getvalue().splitlines()
self.assertIn(canonical_name(fixture1), written_lines)
self.assertIn(canonical_name(fixture2), written_lines)
self.assertIn(canonical_name(MyRequiredFixture),
written_lines)
def test_list_with_filters(self, fixture=MyFixture):
stdout = self.test_main(subcommand='list', filters=[self.id()])
written_lines = stdout.getvalue().splitlines()
self.assertEqual([canonical_name(fixture),
canonical_name(MyRequiredFixture)],
written_lines)
self.assertIn(canonical_name(MyRequiredFixture),
written_lines)
def test_setup(self, fixture=MyFixture, fixture2=MyFixture2):
stdout = self.test_main(subcommand='setup')
written_lines = stdout.getvalue().splitlines()
for obj in [fixture, fixture2, MyRequiredFixture]:
self.assertIn(canonical_name(obj), written_lines)
tobiko.get_fixture(obj).setup_fixture.assert_called_once_with()
tobiko.get_fixture(obj).cleanup_fixture.assert_not_called()
def test_setup_with_filters(self, fixture=MyFixture):
stdout = self.test_main(subcommand='setup', filters=[self.id()])
written_lines = stdout.getvalue().splitlines()
self.assertEqual([canonical_name(fixture),
canonical_name(MyRequiredFixture)],
written_lines)
for obj in [fixture, MyRequiredFixture]:
tobiko.get_fixture(obj).setup_fixture.assert_called_once_with()
tobiko.get_fixture(obj).cleanup_fixture.assert_not_called()
def test_cleanup(self, fixture=MyFixture, fixture2=MyFixture2):
stdout = self.test_main(subcommand='cleanup')
written_lines = stdout.getvalue().splitlines()
for obj in [fixture, fixture2, MyRequiredFixture]:
self.assertIn(canonical_name(obj), written_lines)
tobiko.get_fixture(obj).setup_fixture.assert_not_called()
tobiko.get_fixture(obj).cleanup_fixture.assert_called_once_with()
def test_cleanup_with_filters(self, fixture=MyFixture):
stdout = self.test_main(subcommand='cleanup', filters=[self.id()])
written_lines = stdout.getvalue().splitlines()
self.assertEqual([canonical_name(fixture),
canonical_name(MyRequiredFixture)],
written_lines)
for obj in [fixture, MyRequiredFixture]:
tobiko.get_fixture(obj).setup_fixture.assert_not_called()
tobiko.get_fixture(obj).cleanup_fixture.assert_called_once_with()
def setup_file_repo(self, top_dir):
if not os.path.isdir(os.path.join(top_dir, '.stestr')):
command = ['stestr', '--top-dir', top_dir, 'init']
subprocess.check_call(command)
def make_options(config_file=None, repo_type=None, repo_url=None,
test_path=None, top_dir=None, group_regex=None,
blacklist_file=None, whitelist_file=None,
black_regex=None):
options = []
if config_file:
options += ['--config', config_file]
if repo_type:
options += ['--repo-type', repo_type]
if repo_url:
options += ['--repo-url', repo_url]
if test_path:
options += ['--test-path', test_path]
if top_dir:
options += ['--top-dir', top_dir]
if group_regex:
options += ['--group-regex', group_regex]
if blacklist_file:
options += ['--blacklist-file', blacklist_file]
if whitelist_file:
options += ['--whitelist-file', whitelist_file]
if black_regex:
options += ['--black-regex', black_regex]
return options

37
tobiko/tests/unit/test_testcase.py
View File

@ -13,49 +13,12 @@
# under the License.
from __future__ import absolute_import
import os
import subprocess
import testtools
import tobiko
from tobiko.tests import unit
class DiscoverTestCasesTest(unit.TobikoUnitTest):
test_path = os.path.dirname(__file__)
def setUp(self):
super(DiscoverTestCasesTest, self).setUp()
top_dir = os.path.abspath(self.test_path)
while os.path.isdir(top_dir) and top_dir != os.path.sep:
if os.path.isfile(os.path.join(top_dir, 'tox.ini')):
break
top_dir = os.path.dirname(top_dir)
else:
self.fail("'tox.ini' file not found in any parent "
f"of directory '{self.test_path}'")
if not os.path.isdir(os.path.join(top_dir, '.stestr')):
subprocess.run(['stestr', 'init'], cwd=top_dir, check=True)
self.top_dir = top_dir
self.repo_url = top_dir
# Move to top directory
original_work_dir = os.getcwd()
os.chdir(self.top_dir)
self.addCleanup(os.chdir, original_work_dir)
def test_discover_testcases(self):
testcases = tobiko.discover_test_cases(test_path=self.test_path,
top_dir=self.top_dir,
repo_url=self.repo_url,
filters=[self.id()])
self.assertIn(self.id(), testcases)
class TestCaseTest(unit.TobikoUnitTest):
def setUp(self):