| from __future__ import annotations |
|
|
| from collections.abc import Generator |
| import os |
| import sys |
|
|
| from _pytest.config import Config |
| from _pytest.config.argparsing import Parser |
| from _pytest.nodes import Item |
| from _pytest.stash import StashKey |
| import pytest |
|
|
|
|
| fault_handler_original_stderr_fd_key = StashKey[int]() |
| fault_handler_stderr_fd_key = StashKey[int]() |
|
|
|
|
| def pytest_addoption(parser: Parser) -> None: |
| help = ( |
| "Dump the traceback of all threads if a test takes " |
| "more than TIMEOUT seconds to finish" |
| ) |
| parser.addini("faulthandler_timeout", help, default=0.0) |
|
|
|
|
| def pytest_configure(config: Config) -> None: |
| import faulthandler |
|
|
| |
| |
| |
| |
| |
| stderr_fileno = get_stderr_fileno() |
| if faulthandler.is_enabled(): |
| config.stash[fault_handler_original_stderr_fd_key] = stderr_fileno |
| config.stash[fault_handler_stderr_fd_key] = os.dup(stderr_fileno) |
| faulthandler.enable(file=config.stash[fault_handler_stderr_fd_key]) |
|
|
|
|
| def pytest_unconfigure(config: Config) -> None: |
| import faulthandler |
|
|
| faulthandler.disable() |
| |
| if fault_handler_stderr_fd_key in config.stash: |
| os.close(config.stash[fault_handler_stderr_fd_key]) |
| del config.stash[fault_handler_stderr_fd_key] |
| |
| if fault_handler_original_stderr_fd_key in config.stash: |
| faulthandler.enable(config.stash[fault_handler_original_stderr_fd_key]) |
| del config.stash[fault_handler_original_stderr_fd_key] |
|
|
|
|
| def get_stderr_fileno() -> int: |
| try: |
| fileno = sys.stderr.fileno() |
| |
| |
| if fileno == -1: |
| raise AttributeError() |
| return fileno |
| except (AttributeError, ValueError): |
| |
| |
| |
| assert sys.__stderr__ is not None |
| return sys.__stderr__.fileno() |
|
|
|
|
| def get_timeout_config_value(config: Config) -> float: |
| return float(config.getini("faulthandler_timeout") or 0.0) |
|
|
|
|
| @pytest.hookimpl(wrapper=True, trylast=True) |
| def pytest_runtest_protocol(item: Item) -> Generator[None, object, object]: |
| timeout = get_timeout_config_value(item.config) |
| if timeout > 0: |
| import faulthandler |
|
|
| stderr = item.config.stash[fault_handler_stderr_fd_key] |
| faulthandler.dump_traceback_later(timeout, file=stderr) |
| try: |
| return (yield) |
| finally: |
| faulthandler.cancel_dump_traceback_later() |
| else: |
| return (yield) |
|
|
|
|
| @pytest.hookimpl(tryfirst=True) |
| def pytest_enter_pdb() -> None: |
| """Cancel any traceback dumping due to timeout before entering pdb.""" |
| import faulthandler |
|
|
| faulthandler.cancel_dump_traceback_later() |
|
|
|
|
| @pytest.hookimpl(tryfirst=True) |
| def pytest_exception_interact() -> None: |
| """Cancel any traceback dumping due to an interactive exception being |
| raised.""" |
| import faulthandler |
|
|
| faulthandler.cancel_dump_traceback_later() |
|
|