| 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() |
|
|
