| | from __future__ import annotations |
| |
|
| | import threading |
| | import traceback |
| | from types import TracebackType |
| | from typing import Any |
| | from typing import Callable |
| | from typing import Generator |
| | from typing import TYPE_CHECKING |
| | import warnings |
| |
|
| | import pytest |
| |
|
| |
|
| | if TYPE_CHECKING: |
| | from typing_extensions import Self |
| |
|
| |
|
| | |
| | class catch_threading_exception: |
| | """Context manager catching threading.Thread exception using |
| | threading.excepthook. |
| | |
| | Storing exc_value using a custom hook can create a reference cycle. The |
| | reference cycle is broken explicitly when the context manager exits. |
| | |
| | Storing thread using a custom hook can resurrect it if it is set to an |
| | object which is being finalized. Exiting the context manager clears the |
| | stored object. |
| | |
| | Usage: |
| | with threading_helper.catch_threading_exception() as cm: |
| | # code spawning a thread which raises an exception |
| | ... |
| | # check the thread exception: use cm.args |
| | ... |
| | # cm.args attribute no longer exists at this point |
| | # (to break a reference cycle) |
| | """ |
| |
|
| | def __init__(self) -> None: |
| | self.args: threading.ExceptHookArgs | None = None |
| | self._old_hook: Callable[[threading.ExceptHookArgs], Any] | None = None |
| |
|
| | def _hook(self, args: threading.ExceptHookArgs) -> None: |
| | self.args = args |
| |
|
| | def __enter__(self) -> Self: |
| | self._old_hook = threading.excepthook |
| | threading.excepthook = self._hook |
| | return self |
| |
|
| | def __exit__( |
| | self, |
| | exc_type: type[BaseException] | None, |
| | exc_val: BaseException | None, |
| | exc_tb: TracebackType | None, |
| | ) -> None: |
| | assert self._old_hook is not None |
| | threading.excepthook = self._old_hook |
| | self._old_hook = None |
| | del self.args |
| |
|
| |
|
| | def thread_exception_runtest_hook() -> Generator[None]: |
| | with catch_threading_exception() as cm: |
| | try: |
| | yield |
| | finally: |
| | if cm.args: |
| | thread_name = ( |
| | "<unknown>" if cm.args.thread is None else cm.args.thread.name |
| | ) |
| | msg = f"Exception in thread {thread_name}\n\n" |
| | msg += "".join( |
| | traceback.format_exception( |
| | cm.args.exc_type, |
| | cm.args.exc_value, |
| | cm.args.exc_traceback, |
| | ) |
| | ) |
| | warnings.warn(pytest.PytestUnhandledThreadExceptionWarning(msg)) |
| |
|
| |
|
| | @pytest.hookimpl(wrapper=True, trylast=True) |
| | def pytest_runtest_setup() -> Generator[None]: |
| | yield from thread_exception_runtest_hook() |
| |
|
| |
|
| | @pytest.hookimpl(wrapper=True, tryfirst=True) |
| | def pytest_runtest_call() -> Generator[None]: |
| | yield from thread_exception_runtest_hook() |
| |
|
| |
|
| | @pytest.hookimpl(wrapper=True, tryfirst=True) |
| | def pytest_runtest_teardown() -> Generator[None]: |
| | yield from thread_exception_runtest_hook() |
| |
|
