implement exception timeout

attempt to trigger exception in test first before killing process

Author: xoviat

pytest_timeout.py

@@ -10,10 +10,15 @@
 import os
 import signal
 import sys
+import ctypes
+import time
 import threading
 import traceback
 from collections import namedtuple
 from distutils.version import LooseVersion
+from threading import current_thread
+from _pytest.config import ExitCode
+from _pytest.outcomes import Failed
 
 import py
 import pytest
@@ -23,7 +28,7 @@
 if HAVE_SIGALRM:
     DEFAULT_METHOD = "signal"
 else:
-    DEFAULT_METHOD = "thread"
+    DEFAULT_METHOD = "exception"
 TIMEOUT_DESC = """
 Timeout in seconds before dumping the stacks.  Default is 0 which
 means no timeout.
@@ -56,14 +61,14 @@ def pytest_addoption(parser):
     group.addoption(
         "--timeout_method",
         action="store",
-        choices=["signal", "thread"],
+        choices=["signal", "thread", "exception"],
         help="Deprecated, use --timeout-method",
     )
     group.addoption(
         "--timeout-method",
         dest="timeout_method",
         action="store",
-        choices=["signal", "thread"],
+        choices=["signal", "thread", "exception"],
         help=METHOD_DESC,
     )
     parser.addini("timeout", TIMEOUT_DESC)
@@ -216,6 +221,16 @@ def cancel():
 
         item.cancel_timeout = cancel
         timer.start()
+    elif params.method == "exception":
+        timer = threading.Timer(params.timeout, timeout_exception, (current_thread(), item, params.timeout))
+        timer.name = "%s %s" % (__name__, item.nodeid)
+
+        def cancel():
+            timer.cancel()
+            timer.join()
+
+        item.cancel_timeout = cancel
+        timer.start()
 
 
 def timeout_teardown(item):
@@ -417,6 +432,37 @@ def timeout_timer(item, timeout):
         os._exit(1)
 
 
+def timeout_exception(thread, item, timeout):
+    """Dump stack of threads and call os._exit().
+
+    This disables the capturemanager and dumps stdout and stderr.
+    Then the stacks are dumped and os._exit(1) is called.
+    """
+    if is_debugging():
+        return
+    try:
+        __tracebackhide__ = True
+        nthreads = len(threading.enumerate())
+        if nthreads > 1:
+            write_title("Timeout", sep="+")
+        dump_stacks()
+        if nthreads > 1:
+            write_title("Timeout", sep="+")
+    except Exception:
+        traceback.print_exc()
+    finally:
+        sys.stdout.flush()
+        sys.stderr.flush()
+
+    for _ in range(10):
+        ctypes.pythonapi.PyThreadState_SetAsyncExc(
+            ctypes.c_long(thread.ident), ctypes.py_object(Failed),
+        )
+        time.sleep(0.5)
+
+    os._exit(1)
+
+
 def dump_stacks():
     """Dump the stacks of all threads except the current thread."""
     current_ident = threading.current_thread().ident

test_pytest_timeout.py

@@ -72,6 +72,27 @@ def test_foo():
     assert "++ Timeout ++" in result.stderr.lines[-1]
 
 
+def test_exeption(testdir):
+    testdir.makepyfile(
+        """
+        import time
+
+        def test_foo():
+            time.sleep(2)
+    """
+    )
+    result = testdir.runpytest("--timeout=1", "--timeout-method=exception")
+    result.stderr.fnmatch_lines(
+        [
+            "*++ Timeout ++*",
+            "*~~ Stack of MainThread* ~~*",
+            "*File *, line *, in *",
+            "*++ Timeout ++*",
+        ]
+    )
+    assert "++ Timeout ++" in result.stderr.lines[-1]
+
+
 @pytest.mark.skipif(
     hasattr(sys, "pypy_version_info"), reason="pypy coverage seems broken currently"
 )