| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163 |
- from __future__ import annotations
- import collections
- from collections.abc import Callable
- import functools
- import gc
- import sys
- import traceback
- from typing import NamedTuple
- from typing import TYPE_CHECKING
- import warnings
- from _pytest.config import Config
- from _pytest.nodes import Item
- from _pytest.stash import StashKey
- from _pytest.tracemalloc import tracemalloc_message
- import pytest
- if TYPE_CHECKING:
- pass
- if sys.version_info < (3, 11):
- from exceptiongroup import ExceptionGroup
- # This is a stash item and not a simple constant to allow pytester to override it.
- gc_collect_iterations_key = StashKey[int]()
- def gc_collect_harder(iterations: int) -> None:
- for _ in range(iterations):
- gc.collect()
- class UnraisableMeta(NamedTuple):
- msg: str
- cause_msg: str
- exc_value: BaseException | None
- unraisable_exceptions: StashKey[collections.deque[UnraisableMeta | BaseException]] = (
- StashKey()
- )
- def collect_unraisable(config: Config) -> None:
- pop_unraisable = config.stash[unraisable_exceptions].pop
- errors: list[pytest.PytestUnraisableExceptionWarning | RuntimeError] = []
- meta = None
- hook_error = None
- try:
- while True:
- try:
- meta = pop_unraisable()
- except IndexError:
- break
- if isinstance(meta, BaseException):
- hook_error = RuntimeError("Failed to process unraisable exception")
- hook_error.__cause__ = meta
- errors.append(hook_error)
- continue
- msg = meta.msg
- try:
- warnings.warn(pytest.PytestUnraisableExceptionWarning(msg))
- except pytest.PytestUnraisableExceptionWarning as e:
- # This except happens when the warning is treated as an error (e.g. `-Werror`).
- if meta.exc_value is not None:
- # Exceptions have a better way to show the traceback, but
- # warnings do not, so hide the traceback from the msg and
- # set the cause so the traceback shows up in the right place.
- e.args = (meta.cause_msg,)
- e.__cause__ = meta.exc_value
- errors.append(e)
- if len(errors) == 1:
- raise errors[0]
- if errors:
- raise ExceptionGroup("multiple unraisable exception warnings", errors)
- finally:
- del errors, meta, hook_error
- def cleanup(
- *, config: Config, prev_hook: Callable[[sys.UnraisableHookArgs], object]
- ) -> None:
- # A single collection doesn't necessarily collect everything.
- # Constant determined experimentally by the Trio project.
- gc_collect_iterations = config.stash.get(gc_collect_iterations_key, 5)
- try:
- try:
- gc_collect_harder(gc_collect_iterations)
- collect_unraisable(config)
- finally:
- sys.unraisablehook = prev_hook
- finally:
- del config.stash[unraisable_exceptions]
- def unraisable_hook(
- unraisable: sys.UnraisableHookArgs,
- /,
- *,
- append: Callable[[UnraisableMeta | BaseException], object],
- ) -> None:
- try:
- # we need to compute these strings here as they might change after
- # the unraisablehook finishes and before the metadata object is
- # collected by a pytest hook
- err_msg = (
- "Exception ignored in" if unraisable.err_msg is None else unraisable.err_msg
- )
- summary = f"{err_msg}: {unraisable.object!r}"
- traceback_message = "\n\n" + "".join(
- traceback.format_exception(
- unraisable.exc_type,
- unraisable.exc_value,
- unraisable.exc_traceback,
- )
- )
- tracemalloc_tb = "\n" + tracemalloc_message(unraisable.object)
- msg = summary + traceback_message + tracemalloc_tb
- cause_msg = summary + tracemalloc_tb
- append(
- UnraisableMeta(
- msg=msg,
- cause_msg=cause_msg,
- exc_value=unraisable.exc_value,
- )
- )
- except BaseException as e:
- append(e)
- # Raising this will cause the exception to be logged twice, once in our
- # collect_unraisable and once by the unraisablehook calling machinery
- # which is fine - this should never happen anyway and if it does
- # it should probably be reported as a pytest bug.
- raise
- def pytest_configure(config: Config) -> None:
- prev_hook = sys.unraisablehook
- deque: collections.deque[UnraisableMeta | BaseException] = collections.deque()
- config.stash[unraisable_exceptions] = deque
- config.add_cleanup(functools.partial(cleanup, config=config, prev_hook=prev_hook))
- sys.unraisablehook = functools.partial(unraisable_hook, append=deque.append)
- @pytest.hookimpl(trylast=True)
- def pytest_runtest_setup(item: Item) -> None:
- collect_unraisable(item.config)
- @pytest.hookimpl(trylast=True)
- def pytest_runtest_call(item: Item) -> None:
- collect_unraisable(item.config)
- @pytest.hookimpl(trylast=True)
- def pytest_runtest_teardown(item: Item) -> None:
- collect_unraisable(item.config)
|