/usr/lib/rads/venv/lib/python3.13/site-packages/_pytest
Edit: /usr/lib/rads/venv/lib/python3.13/site-packages/_pytest/unraisableexception.py (5179B)
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)