Block timeout introduction

avocado/core/test.py

@@ -24,11 +24,14 @@
 import logging
 import os
 import shutil
+import signal
 import sys
 import tempfile
+import threading
 import time
 import unittest
 import warnings
+from contextlib import contextmanager
 
 from avocado.core import exceptions, parameters
 from avocado.core.settings import settings
@@ -514,6 +517,29 @@ def phase(self):
         """
         return self.__phase
 
+    @contextmanager
+    def wait_max(self, timeout):
+        """
+        Context manager for getting block of code with its specific timeout.
+
+        Usage:
+            with self.wait_max(3):
+                # code which should take max 3 seconds
+                ...
+
+        :param timeout: Timeout in seconds for block of code.
+        :type timeout: int
+        """
+
+        def raise_timeout():
+            os.kill(os.getpid(), signal.SIGTERM)
+
+        timeout = timeout * self.params.get("timeout_factor", default=1.0)
+        alarm = threading.Timer(timeout, raise_timeout)
+        alarm.start()
+        yield timeout
+        alarm.cancel()
+
     def __str__(self):
         return str(self.name)
 

docs/source/guides/writer/chapters/writing.rst

@@ -773,6 +773,19 @@ runner task, making it raise a
 process is specific to spawner implementation, for more information
 see :class:`avocado.core.plugin_interfaces.Spawner.terminate_task`.
 
+
+Block Timeout
+-------------
+On more complex (and thus usually) longer tests, there may be multiple
+steps to complete. It may be known that some of these steps should not
+take more than a small percentage of the overall expected time for the
+test as a whole. Therefore, it is not convenient to set the timeout for
+the whole test, but it would be better to have timeout for each of those
+steps. For such use-case avocado supports `wait_max` context manager,
+which let you set specific timeout (in seconds) for a block of code:
+
+.. literalinclude:: ../../../../../examples/tests/blocktimeouttest.py
+
 Timeout Factor
 ~~~~~~~~~~~~~~
 
@@ -810,6 +823,9 @@ test logs.  For the previous test execution it shows::
    ...
    [stdlog] 2023-11-29 11:16:23,746 test             L0354 DEBUG|   actual timeout: 6.0
 
+
+.. note:: Be aweare that timeout factor will also affect timeouts created by `wait_max` context manager.
+
 Skipping Tests
 --------------
 

examples/tests/blocktimeouttest.py

@@ -0,0 +1,22 @@
+import time
+
+from avocado import Test
+
+
+class TimeoutTest(Test):
+    """
+    Functional test for avocado. Throw a TestTimeoutError.
+
+    :param sleep_time: How long should the test sleep
+    """
+
+    def test(self):
+        """
+        This should throw a TestTimeoutError.
+        """
+        with self.wait_max(3):
+            sleep_time = float(self.params.get("sleep_time", default=5.0))
+            self.log.info(
+                "Sleeping for %.2f seconds (2 more than the timeout)", sleep_time
+            )
+            time.sleep(sleep_time)

selftests/check.py

@@ -29,7 +29,7 @@
     "nrunner-requirement": 28,
     "unit": 678,
     "jobs": 11,
-    "functional-parallel": 312,
+    "functional-parallel": 313,
     "functional-serial": 7,
     "optional-plugins": 0,
     "optional-plugins-golang": 2,

selftests/functional/basic.py

@@ -543,6 +543,32 @@ def test_runner_timeout(self):
         # Ensure no test aborted error messages show up
         self.assertNotIn(b"TestAbortError: Test aborted unexpectedly", output)
 
+    def test_runner_block_timeout(self):
+        cmd_line = (
+            f"{AVOCADO} run --disable-sysinfo --job-results-dir "
+            f"{self.tmpdir.name} examples/tests/blocktimeouttest.py"
+        )
+        result = process.run(cmd_line, ignore_status=True)
+        json_path = os.path.join(self.tmpdir.name, "latest", "results.json")
+        with open(json_path, encoding="utf-8") as json_file:
+            result_json = json.load(json_file)
+        output = result.stdout
+        expected_rc = exit_codes.AVOCADO_JOB_INTERRUPTED
+        unexpected_rc = exit_codes.AVOCADO_FAIL
+        self.assertNotEqual(
+            result.exit_status,
+            unexpected_rc,
+            f"Avocado crashed (rc {unexpected_rc}):\n{result}",
+        )
+        self.assertEqual(
+            result.exit_status,
+            expected_rc,
+            f"Avocado did not return rc {expected_rc}:\n{result}",
+        )
+        self.assertIn("Timeout reached", result_json["tests"][0]["fail_reason"])
+        # Ensure no test aborted error messages show up
+        self.assertNotIn(b"TestAbortError: Test aborted unexpectedly", output)
+
     def test_runner_timeout_factor(self):
         cmd_line = (
             f"{AVOCADO} run --disable-sysinfo --job-results-dir "