Allow to configure a timeout for test cases execution

Change-Id: Ia605db936c160c79a80f007adff5e7d39876f8e0

tobiko/common/_testcase.py

@@ -16,14 +16,18 @@ from __future__ import absolute_import
 import logging
 import os
 import sys
+import traceback
 import typing  # noqa
 
 from oslo_log import log
 from stestr import config_file
 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__)
@@ -128,22 +132,32 @@ 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()
-        from tobiko import config
-        cls._capture_log = config.CONF.tobiko.logging.capture_log
+        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:
@@ -152,6 +166,25 @@ class TestCase(testtools.TestCase):
                 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)

tobiko/config.py

@@ -61,6 +61,15 @@ HTTP_OPTIONS = [
                help="Don't use proxy server to connect to listed hosts")]
 
 
+TESTCASE_CONF_GROUP_NAME = "testcase"
+
+TESTCASE_OPTIONS = [
+    cfg.FloatOpt('timeout',
+                 default=None,
+                 help=("Timeout (in seconds) used for interrupting test case "
+                       "execution"))]
+
+
 def workspace_config_files(project=None, prog=None):
     project = project or 'tobiko'
     filenames = []
@@ -191,6 +200,9 @@ def register_tobiko_options(conf):
     conf.register_opts(
         group=cfg.OptGroup(HTTP_CONF_GROUP_NAME), opts=HTTP_OPTIONS)
 
+    conf.register_opts(
+        group=cfg.OptGroup(TESTCASE_CONF_GROUP_NAME), opts=TESTCASE_OPTIONS)
+
     for module_name in CONFIG_MODULES:
         module = importlib.import_module(module_name)
         if hasattr(module, 'register_tobiko_options'):
@@ -203,8 +215,15 @@ def list_http_options():
     ]
 
 
+def list_testcase_options():
+    return [
+        (TESTCASE_CONF_GROUP_NAME, itertools.chain(TESTCASE_OPTIONS))
+    ]
+
+
 def list_tobiko_options():
-    all_options = list_http_options()
+    all_options = (list_http_options() +
+                   list_testcase_options())
 
     for module_name in CONFIG_MODULES:
         module = importlib.import_module(module_name)

tobiko/tests/functional/test_testcase.py

@@ -0,0 +1,41 @@
+# 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)