Browse Source

Merge "Use TestSuite and TestResult"

changes/52/821852/2
Zuul 5 months ago committed by Gerrit Code Review
parent
commit
5dfa11d6a4
  1. 74
      tobiko/run/_result.py
  2. 27
      tobiko/run/_run.py

74
tobiko/run/_result.py

@ -0,0 +1,74 @@
# Copyright (c) 2021 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 io
import sys
import typing
import unittest
import tobiko
def get_test_result() -> unittest.TestResult:
return tobiko.setup_fixture(TestResultFixture).result
class TestResultFixture(tobiko.SharedFixture):
verbosity = 2
stream = sys.stderr
description = True
result: unittest.TestResult
def setup_fixture(self):
self.result = TestResult(stream=self.stream,
verbosity=self.verbosity,
description=self.description)
class TestResult(unittest.TextTestResult):
def __init__(self,
stream: typing.TextIO,
description: bool,
verbosity: int):
super().__init__(stream=TextIOWrapper(stream),
descriptions=description,
verbosity=verbosity)
self.buffer = True
def startTest(self, test: unittest.TestCase):
tobiko.push_test_case(test)
super().startTest(test)
def stopTest(self, test: unittest.TestCase) -> None:
super().stopTestRun()
actual_test = tobiko.pop_test_case()
assert actual_test == test
class TextIOWrapper(io.TextIOWrapper):
def __init__(self, stream: typing.TextIO):
super().__init__(buffer=stream.buffer,
encoding='UTF-8',
errors='strict',
line_buffering=True,
write_through=False)
def writeln(self, line: str):
self.write(line + '\n')

27
tobiko/run/_run.py

@ -25,6 +25,7 @@ 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
@ -50,21 +51,25 @@ def run_tests(test_path: typing.Iterable[str],
def run_test_ids(test_ids: typing.Iterable[str]) -> int:
test_classes: typing.Dict[str, typing.List[str]] = \
collections.defaultdict(list)
# regroup test ids my test class keeping test names order
test_ids = list(test_ids)
LOG.info(f'Run {len(test_ids)} test(s)')
for test_id in test_ids:
test_class_id, test_name = test_id.rsplit('.', 1)
test_classes[test_class_id].append(test_name)
result = unittest.TestResult()
# add test cases to the suite ordered by class name
suite = unittest.TestSuite()
for test_class_id, test_names in sorted(test_classes.items()):
LOG.info(f'Enter test class {test_class_id}')
test_class = tobiko.load_object(test_class_id)
for test_name in test_names:
LOG.info(f'Enter test case {test_class_id}.{test_name}')
test_case = test_class(test_name)
test_case.run(result)
LOG.info(f'Exit test case {test_class_id}.{test_name}')
LOG.info(f'Exit test class {test_class_id}')
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')
return result.testsRun
@ -88,7 +93,7 @@ def forked_run_test_ids(test_ids: typing.Iterable[str]) -> int:
def main(test_path: typing.Iterable[str] = None,
test_filename: str = None,
forked: bool = True,
forked: bool = False,
python_path: typing.Iterable[str] = None):
if test_path is None:
test_path = sys.argv[1:]
@ -97,8 +102,8 @@ def main(test_path: typing.Iterable[str] = None,
test_filename=test_filename,
forked=forked,
python_path=python_path)
except Exception as ex:
sys.stderr.write(f'{ex}\n')
except Exception:
LOG.exception("Error running test cases")
sys.exit(1)
else:
sys.exit(0)

Loading…
Cancel
Save