x
/
tobiko
Code Issues Proposed changes

Fix test runner 

- disable forked execution (by default)
- accept string as test path
- raise FileNotFoundError when test case files are not found
- create a new test result instance when any is given
- raise RunTestCasesFailed after test case errors or failures
- use 'spawn' context to create workers pool

Change-Id: Ifde3d43d023e7508ca099bee0a9e9bab0c639789
This commit is contained in:
Federico Ressi 2022-03-25 11:46:17 +01:00 committed by Omer Schwartz
parent df5c26df8e
commit 7453dd6f4f
7 changed files with 99 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
tobiko/run/__init__.py
View File

@ -17,9 +17,14 @@ from __future__ import absolute_import
from tobiko.run import _discover
from tobiko.run import _find
from tobiko.run import _run
discover_test_ids = _discover.discover_test_ids
find_test_ids = _discover.find_test_ids
forked_discover_test_ids = _discover.forked_discover_test_ids
find_test_files = _find.find_test_files
run_tests = _run.run_tests
run_test_ids = _run.run_test_ids

2
tobiko/run/_config.py
View File

@ -36,7 +36,7 @@ class RunConfigFixture(tobiko.SharedFixture):
@property
def forked(self) -> bool:
return self.workers_count != 1
return self.workers_count is not None and self.workers_count != 1
def run_confing(obj=None) -> RunConfigFixture:

2
tobiko/run/_discover.py
View File

@ -32,7 +32,7 @@ from tobiko.run import _worker
LOG = log.getLogger(__name__)
def find_test_ids(test_path: typing.Iterable[str],
def find_test_ids(test_path: typing.Union[str, typing.Iterable[str]],
test_filename: str = None,
python_path: typing.Iterable[str] = None,
forked: bool = None,

26
tobiko/run/_find.py
View File

@ -28,15 +28,17 @@ from tobiko.run import _config
LOG = log.getLogger(__name__)
def find_test_files(test_path: typing.Iterable[str] = None,
def find_test_files(test_path: typing.Union[str, typing.Iterable[str]] = None,
test_filename: str = None,
config: _config.RunConfigFixture = None) \
-> typing.List[str]:
config = _config.run_confing(config)
if test_path:
test_path = list(test_path)
if not test_path:
if test_path is None:
test_path = config.test_path
elif isinstance(test_path, str):
test_path = [test_path]
else:
test_path = list(test_path)
if not test_filename:
test_filename = config.test_filename
test_files: typing.List[str] = []
@ -56,14 +58,20 @@ def find_test_files(test_path: typing.Iterable[str] = None,
LOG.debug("Find test files...\n"
f" dir: '{find_dir}'\n"
f" name: '{find_name}'\n")
output = subprocess.check_output(
['find', find_dir, '-name', find_name],
universal_newlines=True)
f" name: '{find_name}'")
try:
output = subprocess.check_output(
['find', find_dir, '-name', find_name],
universal_newlines=True)
except subprocess.CalledProcessError as ex:
LOG.exception("Test files not found.")
raise FileNotFoundError('Test files not found: \n'
f" dir: '{find_dir}'\n"
f" name: '{find_name}'") from ex
for line in output.splitlines():
line = line.strip()
if line:
assert os.path.isfile(line)
test_files.append(line)
LOG.debug("Found test file(s):\n"

48
tobiko/run/_run.py
View File

@ -25,32 +25,31 @@ from oslo_log import log
import tobiko
from tobiko.run import _config
from tobiko.run import _discover
from tobiko.run import _result
from tobiko.run import _worker
LOG = log.getLogger(__name__)
def run_tests(test_path: typing.Iterable[str],
def run_tests(test_path: typing.Union[str, typing.Iterable[str]],
test_filename: str = None,
python_path: typing.Iterable[str] = None,
forked: bool = None,
config: _config.RunConfigFixture = None):
config: _config.RunConfigFixture = None,
result: unittest.TestResult = None):
test_ids = _discover.find_test_ids(test_path=test_path,
test_filename=test_filename,
python_path=python_path,
forked=forked,
config=config)
if forked:
forked_run_test_ids(test_ids=test_ids)
else:
run_test_ids(test_ids=test_ids)
return run_test_ids(test_ids=test_ids, result=result)
def run_test_ids(test_ids: typing.Iterable[str]) -> int:
def run_test_ids(test_ids: typing.List[str],
result: unittest.TestResult = None) \
-> int:
test_classes: typing.Dict[str, typing.List[str]] = \
collections.defaultdict(list)
# run the test suite
if result is None:
result = unittest.TestResult()
# regroup test ids my test class keeping test names order
test_ids = list(test_ids)
@ -66,41 +65,30 @@ def run_test_ids(test_ids: typing.Iterable[str]) -> int:
test = test_class(test_name)
suite.addTest(test)
# run the test suite
result = tobiko.setup_fixture(_result.TestResultFixture()).result
LOG.info(f'Run {len(test_ids)} test(s)')
suite.run(result)
LOG.info(f'{result.testsRun} test(s) run')
if result.testsRun and (result.errors or result.failures):
raise RunTestCasesFailed(
errors='\n'.join(str(e) for e in result.errors),
failures='\n'.join(str(e) for e in result.failures))
return result.testsRun
def forked_run_test_ids(test_ids: typing.Iterable[str]) -> int:
test_classes: typing.Dict[str, typing.List[str]] = \
collections.defaultdict(list)
test_ids = list(test_ids)
LOG.info(f'Run {len(test_ids)} test(s)')
for test_id in test_ids:
test_class_id, _ = test_id.rsplit('.', 1)
test_classes[test_class_id].append(test_id)
results = [_worker.call_async(run_test_ids, test_ids=grouped_ids)
for _, grouped_ids in sorted(test_classes.items())]
count = 0
for result in results:
count += result.get()
LOG.info(f'{count} test(s) run')
return count
class RunTestCasesFailed(tobiko.TobikoException):
message = ('Test case execution failed:\n'
'{errors}\n'
'{failures}\n')
def main(test_path: typing.Iterable[str] = None,
test_filename: str = None,
forked: bool = False,
python_path: typing.Iterable[str] = None):
if test_path is None:
test_path = sys.argv[1:]
try:
run_tests(test_path=test_path,
test_filename=test_filename,
forked=forked,
python_path=python_path)
except Exception:
LOG.exception("Error running test cases")

12
tobiko/run/_worker.py
View File

@ -15,7 +15,8 @@
# under the License.
from __future__ import absolute_import
import multiprocessing.pool
import multiprocessing
from multiprocessing import pool
import typing
import tobiko
@ -26,7 +27,7 @@ class WorkersPoolFixture(tobiko.SharedFixture):
config = tobiko.required_fixture(_config.RunConfigFixture)
pool: multiprocessing.pool.Pool
pool: pool.Pool
workers_count: int = 0
def __init__(self, workers_count: int = None):
@ -39,14 +40,15 @@ class WorkersPoolFixture(tobiko.SharedFixture):
if not workers_count:
workers_count = self.config.workers_count
self.workers_count = workers_count or 0
self.pool = multiprocessing.pool.Pool(processes=workers_count or None)
context = multiprocessing.get_context('spawn')
self.pool = context.Pool(processes=workers_count or None)
def workers_pool() -> multiprocessing.pool.Pool:
def workers_pool() -> pool.Pool:
return tobiko.setup_fixture(WorkersPoolFixture).pool
def call_async(func: typing.Callable,
*args,
**kwargs) -> multiprocessing.pool.AsyncResult:
**kwargs):
return workers_pool().apply_async(func, args=args, kwds=kwargs)

50
tobiko/tests/functional/run/test_run.py Normal file
View File

@ -0,0 +1,50 @@
# Copyright (c) 2022 Red Hat, Inc.
#
# 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 functools
import os
import typing
import unittest
import testtools
from tobiko import run
def nested_test_case(test_method: typing.Callable[[testtools.TestCase], None]):
@functools.wraps(test_method)
def wrapper(self: unittest.TestCase):
nested_counter = int(os.environ.get('NESTED_TEST_CASE', 0))
if not nested_counter:
os.environ['NESTED_TEST_CASE'] = str(nested_counter + 1)
try:
test_method(self)
finally:
if nested_counter:
os.environ['NESTED_TEST_CASE'] = str(nested_counter)
else:
os.environ.pop('NESTED_TEST_CASE')
return wrapper
class RunTestsTest(unittest.TestCase):
@nested_test_case
def test_run_tests(self):
result = run.run_tests(__file__)
self.assertGreater(result, 0)