faulthandler.py 4.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119
  1. from __future__ import annotations
  2. from collections.abc import Generator
  3. import os
  4. import sys
  5. from _pytest.config import Config
  6. from _pytest.config.argparsing import Parser
  7. from _pytest.nodes import Item
  8. from _pytest.stash import StashKey
  9. import pytest
  10. fault_handler_original_stderr_fd_key = StashKey[int]()
  11. fault_handler_stderr_fd_key = StashKey[int]()
  12. def pytest_addoption(parser: Parser) -> None:
  13. help_timeout = (
  14. "Dump the traceback of all threads if a test takes "
  15. "more than TIMEOUT seconds to finish"
  16. )
  17. help_exit_on_timeout = (
  18. "Exit the test process if a test takes more than "
  19. "faulthandler_timeout seconds to finish"
  20. )
  21. parser.addini("faulthandler_timeout", help_timeout, default=0.0)
  22. parser.addini(
  23. "faulthandler_exit_on_timeout", help_exit_on_timeout, type="bool", default=False
  24. )
  25. def pytest_configure(config: Config) -> None:
  26. import faulthandler
  27. # at teardown we want to restore the original faulthandler fileno
  28. # but faulthandler has no api to return the original fileno
  29. # so here we stash the stderr fileno to be used at teardown
  30. # sys.stderr and sys.__stderr__ may be closed or patched during the session
  31. # so we can't rely on their values being good at that point (#11572).
  32. stderr_fileno = get_stderr_fileno()
  33. if faulthandler.is_enabled():
  34. config.stash[fault_handler_original_stderr_fd_key] = stderr_fileno
  35. config.stash[fault_handler_stderr_fd_key] = os.dup(stderr_fileno)
  36. faulthandler.enable(file=config.stash[fault_handler_stderr_fd_key])
  37. def pytest_unconfigure(config: Config) -> None:
  38. import faulthandler
  39. faulthandler.disable()
  40. # Close the dup file installed during pytest_configure.
  41. if fault_handler_stderr_fd_key in config.stash:
  42. os.close(config.stash[fault_handler_stderr_fd_key])
  43. del config.stash[fault_handler_stderr_fd_key]
  44. # Re-enable the faulthandler if it was originally enabled.
  45. if fault_handler_original_stderr_fd_key in config.stash:
  46. faulthandler.enable(config.stash[fault_handler_original_stderr_fd_key])
  47. del config.stash[fault_handler_original_stderr_fd_key]
  48. def get_stderr_fileno() -> int:
  49. try:
  50. fileno = sys.stderr.fileno()
  51. # The Twisted Logger will return an invalid file descriptor since it is not backed
  52. # by an FD. So, let's also forward this to the same code path as with pytest-xdist.
  53. if fileno == -1:
  54. raise AttributeError()
  55. return fileno
  56. except (AttributeError, ValueError):
  57. # pytest-xdist monkeypatches sys.stderr with an object that is not an actual file.
  58. # https://docs.python.org/3/library/faulthandler.html#issue-with-file-descriptors
  59. # This is potentially dangerous, but the best we can do.
  60. assert sys.__stderr__ is not None
  61. return sys.__stderr__.fileno()
  62. def get_timeout_config_value(config: Config) -> float:
  63. return float(config.getini("faulthandler_timeout") or 0.0)
  64. def get_exit_on_timeout_config_value(config: Config) -> bool:
  65. exit_on_timeout = config.getini("faulthandler_exit_on_timeout")
  66. assert isinstance(exit_on_timeout, bool)
  67. return exit_on_timeout
  68. @pytest.hookimpl(wrapper=True, trylast=True)
  69. def pytest_runtest_protocol(item: Item) -> Generator[None, object, object]:
  70. timeout = get_timeout_config_value(item.config)
  71. exit_on_timeout = get_exit_on_timeout_config_value(item.config)
  72. if timeout > 0:
  73. import faulthandler
  74. stderr = item.config.stash[fault_handler_stderr_fd_key]
  75. faulthandler.dump_traceback_later(timeout, file=stderr, exit=exit_on_timeout)
  76. try:
  77. return (yield)
  78. finally:
  79. faulthandler.cancel_dump_traceback_later()
  80. else:
  81. return (yield)
  82. @pytest.hookimpl(tryfirst=True)
  83. def pytest_enter_pdb() -> None:
  84. """Cancel any traceback dumping due to timeout before entering pdb."""
  85. import faulthandler
  86. faulthandler.cancel_dump_traceback_later()
  87. @pytest.hookimpl(tryfirst=True)
  88. def pytest_exception_interact() -> None:
  89. """Cancel any traceback dumping due to an interactive exception being
  90. raised."""
  91. import faulthandler
  92. faulthandler.cancel_dump_traceback_later()