x/tobiko
Code Issues Proposed changes

Merge "Use pytest timeout and log capturing"

Browse Source
This commit is contained in:
Zuul 2021-01-28 00:36:26 +00:00 committed by Gerrit Code Review
commit cb983697f3
10 changed files with 202 additions and 130 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
extra-requirements.txt
View File

@ -2,4 +2,5 @@ ansi2html # LGPLv3+
pandas # BSD
podman # Apache-2.0
pytest-rerunfailures # MPL-2.0
pytest-timeout # MIT
validations-libs # APACHE-2.0

1
tobiko/__init__.py
View File

@ -109,7 +109,6 @@ skip_test = _skip.skip_test
skip_unless = _skip.skip_unless
skip = _skip.skip
BaseTestCase = _testcase.TestCase
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

82
tobiko/common/_testcase.py
View File

@ -13,24 +13,15 @@
# under the License.
from __future__ import absolute_import
import logging
import os
import sys
import traceback
import typing # noqa
from oslo_log import log
import testtools
from tobiko.common import _config
from tobiko.common import _exception
from tobiko.common import _itimer
from tobiko.common import _logging
from tobiko.common import _time
LOG = log.getLogger(__name__)
os.environ.setdefault('PYTHON', sys.executable)
@ -133,83 +124,22 @@ def discover_test_cases(finder=FINDER, **kwargs):
return finder.discover_test_cases(**kwargs)
class TestCaseTimeoutError(_exception.TobikoException):
message = ("Test case '{testcase_id}' timed out after {timeout} seconds "
"at:\n{stack}")
class TestCase(testtools.TestCase):
_capture_log = False
_capture_log_level = logging.DEBUG
_capture_log_logger = logging.root
_testcase_timeout: _time.Seconds = None
@classmethod
def setUpClass(cls):
super(TestCase, cls).setUpClass()
config = _config.tobiko_config()
cls._capture_log = config.logging.capture_log
cls._testcase_timeout = _time.to_seconds(cls._testcase_timeout or
config.testcase.timeout or
None)
def setUp(self):
super(TestCase, self).setUp()
self._push_test_case()
self._setup_capture_log()
self._setup_testcase_timeout()
def _setup_capture_log(self):
if self._capture_log:
self.useFixture(_logging.CaptureLogFixture(
test_case_id=self.id(),
level=self._capture_log_level,
logger=self._capture_log_logger))
def _setup_testcase_timeout(self):
timeout = self._testcase_timeout
if timeout is not None:
self.useFixture(_itimer.itimer(
delay=timeout,
on_timeout=self._on_testcase_timeout))
def _on_testcase_timeout(self, _signal_number, frame):
stack = traceback.extract_stack(frame)
for test_method_index, summary in enumerate(stack):
if self._testMethodName == summary.name:
stack = stack[test_method_index:]
break
formatted_stack = ''.join(traceback.format_list(stack))
timeout = self._testcase_timeout
raise TestCaseTimeoutError(testcase_id=self.id(), timeout=timeout,
stack=formatted_stack)
def _push_test_case(self):
push_test_case(self)
self.addCleanup(self._pop_test_case)
def _pop_test_case(self):
self.assertIs(self, pop_test_case())
class TestCasesManager(object):
def __init__(self):
self._test_cases: typing.List[TestCase] = []
self._test_cases: typing.List[testtools.TestCase] = []
def get_test_case(self) -> TestCase:
def get_test_case(self) -> testtools.TestCase:
try:
return self._test_cases[-1]
except IndexError:
return DUMMY_TEST_CASE
def pop_test_case(self) -> TestCase:
def pop_test_case(self) -> testtools.TestCase:
return self._test_cases.pop()
def push_test_case(self, test_case: TestCase):
_exception.check_valid_type(test_case, TestCase)
def push_test_case(self, test_case: testtools.TestCase):
_exception.check_valid_type(test_case, testtools.TestCase)
self._test_cases.append(test_case)
@ -231,7 +161,7 @@ def get_test_case(manager: TestCasesManager = TEST_CASES) -> \
return manager.get_test_case()
class DummyTestCase(TestCase):
class DummyTestCase(testtools.TestCase):
def runTest(self):
pass

5
tobiko/config.py
View File

@ -21,8 +21,6 @@ import typing # noqa
from oslo_config import cfg
from oslo_log import log
import testtools
from testtools import monkey
import tobiko
@ -47,7 +45,7 @@ LOGGING_CONF_GROUP_NAME = "logging"
LOGGING_OPTIONS = [
cfg.BoolOpt('capture_log',
default=True,
help="Whenever to capture LOG during test case excecution"),
help="Whenever to capture LOG during test case execution"),
]
HTTP_CONF_GROUP_NAME = "http"
@ -245,7 +243,6 @@ def setup_tobiko_config(conf):
warnings_logger.logger.setLevel(log.ERROR)
tobiko.setup_fixture(HttpProxyFixture)
monkey.patch(testtools, 'TestCase', tobiko.BaseTestCase)
for module_name in CONFIG_MODULES:
module = importlib.import_module(module_name)

72
tobiko/tests/conftest.py
View File

@ -16,10 +16,71 @@ from __future__ import absolute_import
from datetime import datetime
from oslo_log import log
from py.xml import html # pylint: disable=no-name-in-module,import-error
import pytest
LOG = log.getLogger(__name__)
@pytest.hookimpl
def pytest_configure(config):
configure_caplog(config)
configure_timeout(config)
def configure_caplog(config):
import tobiko
tobiko_config = tobiko.tobiko_config()
if tobiko_config.logging.capture_log is True:
if tobiko_config.debug:
default = 'DEBUG'
else:
default = 'INFO'
else:
default = 'FATAL'
for key in ['log_level',
'log_file_level',
'log_cli_level']:
set_default_inicfg(config, key, default)
default = tobiko_config.logging_default_format_string
if default:
# instance and color are not supported by pytest
default = default.replace('%(instance)s', '')
default = default.replace('%(color)s', '')
if default:
for key in ['log_format',
'log_file_format',
'log_cli_format']:
set_default_inicfg(config, key, default)
default = tobiko_config.log_date_format
if default:
for key in ['log_date_format',
'log_file_date_format',
'log_cli_date_format']:
set_default_inicfg(config, key, default)
def set_default_inicfg(config, key, default):
value = config.inicfg.setdefault(key, default)
if value != default:
LOG.debug(f"Set default inicfg: {key} = {value}")
else:
LOG.debug(f"Keep existing inicfg: {key} = {value}")
def configure_timeout(config):
import tobiko
tobiko_config = tobiko.tobiko_config()
default = tobiko_config.testcase.timeout
if default is not None and default > 0.:
set_default_inicfg(config, 'timeout', default)
def pytest_html_results_table_header(cells):
cells.insert(2, html.th("Description"))
cells.insert(1, html.th("Time", class_="sortable time", col="time"))
@ -37,3 +98,14 @@ def pytest_runtest_makereport(item, call): # pylint: disable=unused-argument
outcome = yield
report = outcome.get_result()
report.description = str(item.function.__doc__)
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_call(item):
# pylint: disable=protected-access
import tobiko
tobiko.push_test_case(item._testcase)
try:
yield
finally:
tobiko.pop_test_case()

41
tobiko/tests/functional/test_testcase.py
View File

@ -1,41 +0,0 @@
# Copyright (c) 2020 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 time
import testtools
class TestCaseTest(testtools.TestCase):
def test_with_timeout(self):
class MyTest(testtools.TestCase):
_testcase_timeout = 1.
def test_busy(self):
while True:
time.sleep(0.)
test_case = MyTest('test_busy')
test_result = testtools.TestResult()
test_case.run(test_result)
reported_test_case, reported_error = test_result.errors[-1]
self.assertIs(test_case, reported_test_case)
self.assertIn('TestCaseTimeoutError', reported_error)

112
tobiko/tests/unit/test_conftest.py Normal file
View File

@ -0,0 +1,112 @@
# Copyright 2021 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
from oslo_log import log
import mock
import tobiko
from tobiko.tests import unit
from tobiko.tests import conftest
LOG = log.getLogger(__name__)
class CaplogTest(unit.TobikoUnitTest):
def test_configure_caplog(self):
pytest_config = mock.MagicMock(inicfg={
'log_level': '<existing>',
'log_format': '<existing>',
'log_date_format': '<existing>'})
conftest.configure_caplog(pytest_config)
self.assertEqual('<existing>', pytest_config.inicfg['log_level'])
self.assertEqual('<existing>', pytest_config.inicfg['log_format'])
self.assertEqual('<existing>', pytest_config.inicfg['log_date_format'])
def test_configure_caplog_debug(self):
self.patch_caplog_config(capture_log=True, debug=True)
pytest_config = mock.MagicMock(inicfg={})
conftest.configure_caplog(pytest_config)
self.assertEqual('DEBUG', pytest_config.inicfg['log_level'])
def test_configure_caplog_info(self):
self.patch_caplog_config(capture_log=True, debug=False)
pytest_config = mock.MagicMock(inicfg={})
conftest.configure_caplog(pytest_config)
self.assertEqual('INFO', pytest_config.inicfg['log_level'])
def test_configure_caplog_fatal(self):
self.patch_caplog_config(capture_log=False)
pytest_config = mock.MagicMock(inicfg={})
conftest.configure_caplog(pytest_config)
self.assertEqual('FATAL', pytest_config.inicfg['log_level'])
def test_configure_caplog_log_format(self):
self.patch_caplog_config(log_format='<some-format>')
pytest_config = mock.MagicMock(inicfg={})
conftest.configure_caplog(pytest_config)
self.assertEqual('<some-format>', pytest_config.inicfg['log_format'])
def test_configure_caplog_log_date_format(self):
self.patch_caplog_config(log_date_format='<some-format>')
pytest_config = mock.MagicMock(inicfg={})
conftest.configure_caplog(pytest_config)
self.assertEqual('<some-format>',
pytest_config.inicfg['log_date_format'])
def patch_caplog_config(self, capture_log=False, debug=False,
log_format=None, log_date_format=None):
tobiko_config = self.patch(tobiko, 'tobiko_config').return_value
tobiko_config.logging.capture_log = capture_log
tobiko_config.debug = debug
tobiko_config.logging_default_format_string = log_format
tobiko_config.log_date_format = log_date_format
class TimeoutTest(unit.TobikoUnitTest):
def test_configure_timeout_existing(self):
pytest_config = mock.MagicMock(inicfg={'timeout': '<existing>'})
conftest.configure_timeout(pytest_config)
self.assertEqual('<existing>', pytest_config.inicfg['timeout'])
def test_configure_timeout_none(self):
self.patch_timeout_config(timeout=None)
pytest_config = mock.MagicMock(inicfg={})
conftest.configure_timeout(pytest_config)
self.assertNotIn('timeout', pytest_config.inicfg)
def test_configure_timeout_zero(self):
self.patch_timeout_config(timeout=0.)
pytest_config = mock.MagicMock(inicfg={})
conftest.configure_timeout(pytest_config)
self.assertNotIn('timeout', pytest_config.inicfg)
def test_configure_timeout_negative(self):
self.patch_timeout_config(timeout=-1.)
pytest_config = mock.MagicMock(inicfg={})
conftest.configure_timeout(pytest_config)
self.assertNotIn('timeout', pytest_config.inicfg)
def test_configure_timeout_positive(self):
self.patch_timeout_config(timeout=10.)
pytest_config = mock.MagicMock(inicfg={})
conftest.configure_timeout(pytest_config)
self.assertEqual(10., pytest_config.inicfg['timeout'])
def patch_timeout_config(self, timeout):
tobiko_config = self.patch(tobiko, 'tobiko_config').return_value
tobiko_config.testcase.timeout = timeout

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

@ -75,7 +75,7 @@ class TestCaseTest(unit.TobikoUnitTest):
def test_get_test_case_out_of_context(self):
manager = tobiko.TestCasesManager()
result = tobiko.get_test_case(manager=manager)
self.assertIsInstance(result, tobiko.BaseTestCase)
self.assertIsInstance(result, testtools.TestCase)
self.assertEqual('tobiko.common._testcase.DummyTestCase.runTest',
result.id())

8
tools/run_tests.py
View File

@ -137,12 +137,6 @@ def log_environ():
capture_stdout=False)
def debug_test_cases():
common.execute_python('-m testtools.run {posargs}',
posargs=common.get_posargs(),
capture_stdout=False)
def run_test_cases():
xdist_options = ''
if TOX_NUM_PROCESSES != '1':
@ -153,7 +147,9 @@ def run_test_cases():
common.execute(f"pytest "
f"{xdist_options} "
f"{rerun_options} "
f"--log-file={TOX_REPORT_LOG} "
f"--junitxml={TOX_REPORT_XML} "
f"--junit-prefix={TOX_REPORT_NAME} "
f"--html={TOX_REPORT_HTML} --self-contained-html "
f"{common.get_posargs()}",
capture_stdout=False)

8
tox.ini
View File

@ -20,12 +20,14 @@ passenv =
OS_*
TOBIKO_*
TOX_*
PYTEST_*
setenv =
OS_LOG_CAPTURE = {env:OS_LOG_CAPTURE:true}
OS_STDOUT_CAPTURE = {env:OS_STDOUT_CAPTURE:true}
OS_STDERR_CAPTURE = {env:OS_STDERR_CAPTURE:true}
OS_TEST_PATH = {toxinidir}/tobiko/tests/unit
PS1 = [tobiko@{envname}] {env:PS1:}
PYTEST_TIMEOUT = {env:PYTEST_TIMEOUT:300}
PYTHONWARNINGS = ignore::Warning,{env:PYTHONWARNINGS:}
RUN_TESTS_EXTRA_ARGS = {env:OS_TEST_PATH}
TOBIKO_PREVENT_CREATE = {env:TOBIKO_PREVENT_CREATE:false}
@ -140,7 +142,9 @@ envdir = {[testenv:py3]envdir}
passenv =
{[testenv]passenv}
*_proxy
setenv = {[testenv]setenv}
setenv =
{[testenv]setenv}
PYTEST_TIMEOUT = 1200
[testenv:venv]
@ -210,6 +214,7 @@ setenv =
{[integration]setenv}
OS_TEST_PATH = {toxinidir}/tobiko/tests/faults
TOX_NUM_PROCESSES = 1
PYTEST_TIMEOUT = 3600
[testenv:instanceha]
@ -221,6 +226,7 @@ setenv =
{[integration]setenv}
OS_TEST_PATH = {toxinidir}/tobiko/tests/faults/iha
TOX_NUM_PROCESSES = 1
PYTEST_TIMEOUT = 3600
# --- CI workflow test environments -------------------------------------------