Pytest breaks code that relies on threading.excepthook

[klamann]

I'm using a custom threading.excepthook in my application. When an exception is raised in a thread, I expect that the code in my custom excepthook function is executed, otherwise my application will fail.

When I run my application normally, the excepthook works as expected

import threading

def excepthook(args: threading.ExceptHookArgs):
    print("what happens next is very important...")

threading.excepthook = excepthook

def foo():
    raise ValueError("oh no!")

t = threading.Thread(target=foo)
t.start()
t.join()

Output:

what happens next is very important...

But when I try to test the same function with pytest, the excepthook is replaced with some PytestUnhandledThreadExceptionWarning that I did not ask for and my own excepthook gets ignored.

import threading

def excepthook(args: threading.ExceptHookArgs):
    print("what happens next is very important...")

threading.excepthook = excepthook

def test_excepthook():
    def foo():
        raise ValueError("oh no!")
    t = threading.Thread(target=foo)
    t.start()
    t.join()

Output:

============================= test session starts =============================
platform win32 -- Python 3.8.10, pytest-6.2.4, py-1.10.0, pluggy-0.13.1 -- D:\Users\*snip*\.venv\Scripts\python.exe
cachedir: .pytest_cache
rootdir: D:\Users\*snip*\tests, configfile: pytest.ini
plugins: cov-2.12.1, mock-3.6.1, pycharm-0.7.0
collecting ... collected 1 item

excepthook_test.py::test_excepthook PASSED                               [100%]

============================== warnings summary ===============================
excepthook_test.py::test_excepthook
  D:\Users\*snip*\.venv\lib\site-packages\_pytest\threadexception.py:75: PytestUnhandledThreadExceptionWarning: Exception in thread Thread-1
  
  Traceback (most recent call last):
    File "C:\*snip*\Python38\lib\threading.py", line 932, in _bootstrap_inner
      self.run()
    File "C:\*snip*\Python38\lib\threading.py", line 870, in run
      self._target(*self._args, **self._kwargs)
    File "D:\Users\*snip*\tests\excepthook_test.py", line 10, in foo
      raise ValueError("oh no!")
  ValueError: oh no!
  
    warnings.warn(pytest.PytestUnhandledThreadExceptionWarning(msg))

-- Docs: https://docs.pytest.org/en/stable/warnings.html
======================== 1 passed, 1 warning in 0.12s =========================

Process finished with exit code 0

It looks like this function is responsible and it gets added to each test function automatically. How can I prevent that from happening?

Running pytest 6.2.4 on Python 3.8 on Windows.

[graingert]

you can disable the plugin https://docs.pytest.org/en/6.2.x/usage.html#warning-about-unraisable-exceptions-and-unhandled-thread-exceptions with -p no:threadexception

but rather than rely on global state, you could also refactor your application and test to set your thread hook in a context manager:

import contextlib
import threading

@contextlib.contextmanager
def threadhook_context(new_hook):
    old_hook, threading.excepthook = threading.excepthook, new_hook
    try:
        yield old_hook
    finally:
        assert threading.excepthook is new_hook
        threading.excepthook = old_hook

def excepthook(args: threading.ExceptHookArgs):
    print("what happens next is very important...")

def foo():
     raise ValueError("oh no!")

def some_system_under_test():
    with threadhook_context(excepthook):
        t = threading.Thread(target=foo)
        t.start()
        t.join()

def test_excepthook(capsys):
    assert some_system_under_test() is None
    assert capsys.readouterr().out == "what happens next is very important...\n"

this is how trio manages it: https://github.com/python-trio/trio/blob/289610534afb73ab34a982d846c040a8e39b3ed4/trio/_core/tests/tutil.py#L99-L109

[klamann]

that's an interesting approach... I was wondering if there was a way to make the excepthook thread-local. Though I would prefer to have a subclass of threading.Thread instead of a context manager for each invocation of a thread that relies on the excepthook. But that will probably break when I have two threads running at the same time with conflicting excepthook functions...

well, I'm not happy with how threading.excepthook is implemented in Python, but I think your solution might work for me. Thanks!

[graingert]

@klamann the threading.excepthook is supposed to be a last recourse for unhandled exceptions - you should only use it to debug exceptions that are accidentally leaked by your script.

It sounds like a ThreadPoolExecutor would fit your use case better:

import concurrent.futures

def foo():
    raise ValueError("hello")

def callback(fut):
    if fut.exception():
        print("what happens next is very important...")

def some_system():
    with concurrent.futures.ThreadPoolExecutor() as tpe:
        fut = tpe.submit(foo)
        fut.add_done_callback(callback)

def test_foo(capsys):
    assert some_system() is None
    assert capsys.readouterr().out == "what happens next is very important...\n"

[klamann]

this was very helpful, thank you so much!