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
| 13 | |
| 14 | # Copied from cpython/Lib/test/support/threading_helper.py, with modifications. |
| 15 | class catch_threading_exception: |
| 16 | """Context manager catching threading.Thread exception using |
| 17 | threading.excepthook. |
| 18 | |
| 19 | Storing exc_value using a custom hook can create a reference cycle. The |
| 20 | reference cycle is broken explicitly when the context manager exits. |
| 21 | |
| 22 | Storing thread using a custom hook can resurrect it if it is set to an |
| 23 | object which is being finalized. Exiting the context manager clears the |
| 24 | stored object. |
| 25 | |
| 26 | Usage: |
| 27 | with threading_helper.catch_threading_exception() as cm: |
| 28 | # code spawning a thread which raises an exception |
| 29 | ... |
| 30 | # check the thread exception: use cm.args |
| 31 | ... |
| 32 | # cm.args attribute no longer exists at this point |
| 33 | # (to break a reference cycle) |
| 34 | """ |
| 35 | |
| 36 | def __init__(self) -> None: |
| 37 | self.args: Optional["threading.ExceptHookArgs"] = None |
| 38 | self._old_hook: Optional[Callable[["threading.ExceptHookArgs"], Any]] = None |
| 39 | |
| 40 | def _hook(self, args: "threading.ExceptHookArgs") -> None: |
| 41 | self.args = args |
| 42 | |
| 43 | def __enter__(self) -> "catch_threading_exception": |
| 44 | self._old_hook = threading.excepthook |
| 45 | threading.excepthook = self._hook |
| 46 | return self |
| 47 | |
| 48 | def __exit__( |
| 49 | self, |
| 50 | exc_type: Optional[Type[BaseException]], |
| 51 | exc_val: Optional[BaseException], |
| 52 | exc_tb: Optional[TracebackType], |
| 53 | ) -> None: |
| 54 | assert self._old_hook is not None |
| 55 | threading.excepthook = self._old_hook |
| 56 | self._old_hook = None |
| 57 | del self.args |
| 58 | |
| 59 | |
| 60 | def thread_exception_runtest_hook() -> Generator[None, None, None]: |
no outgoing calls
no test coverage detected