p53209761
/
test_ok2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add unraisableexception and threadexception plugins

This commit is contained in:
Ran Benita 2020-11-13 23:47:19 +02:00
parent 148e3c582a
commit d50df85e26
10 changed files with 520 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
changelog/5299.feature.rst Normal file
View File

@ -0,0 +1,2 @@
pytest now warns about unraisable exceptions and unhandled thread exceptions that occur in tests on Python>=3.8.
See :ref:`unraisable` for more information.

6
doc/en/reference.rst
View File

@ -1090,6 +1090,12 @@ Custom warnings generated in some situations such as improper usage or deprecate
.. autoclass:: pytest.PytestUnknownMarkWarning
:show-inheritance:
.. autoclass:: pytest.PytestUnraisableExceptionWarning
:show-inheritance:
.. autoclass:: pytest.PytestUnhandledThreadExceptionWarning
:show-inheritance:
Consult the :ref:`internal-warnings` section in the documentation for more information.

32
doc/en/usage.rst
View File

@ -470,6 +470,38 @@ seconds to finish (not available on Windows).
the command-line using ``-o faulthandler_timeout=X``.
.. _unraisable:
Warning about unraisable exceptions and unhandled thread exceptions
-------------------------------------------------------------------
.. versionadded:: 6.2
.. note::
These features only work on Python>=3.8.
Unhandled exceptions are exceptions that are raised in a situation in which
they cannot propagate to a caller. The most common case is an exception raised
in a :meth:`__del__ <object.__del__>` implementation.
Unhandled thread exceptions are exceptions raised in a :class:`~threading.Thread`
but not handled, causing the thread to terminate uncleanly.
Both types of exceptions are normally considered bugs, but may go unnoticed
because they don't cause the program itself to crash. Pytest detects these
conditions and issues a warning that is visible in the test run summary.
The plugins are automatically enabled for pytest runs, unless the
``-p no:unraisableexception`` (for unraisable exceptions) and
``-p no:threadexception`` (for thread exceptions) options are given on the
command-line.
The warnings may be silenced selectivly using the :ref:`pytest.mark.filterwarnings ref`
mark. The warning categories are :class:`pytest.PytestUnraisableExceptionWarning` and
:class:`pytest.PytestUnhandledThreadExceptionWarning`.
Creating JUnitXML format files
----------------------------------------------------

1
src/_pytest/config/__init__.py
View File

@ -251,6 +251,7 @@ default_plugins = essential_plugins + (
"warnings",
"logging",
"reports",
*(["unraisableexception", "threadexception"] if sys.version_info >= (3, 8) else []),
"faulthandler",
)

90
src/_pytest/threadexception.py Normal file
View File

@ -0,0 +1,90 @@
import threading
import traceback
import warnings
from types import TracebackType
from typing import Any
from typing import Callable
from typing import Generator
from typing import Optional
from typing import Type
import pytest
# Copied from cpython/Lib/test/support/threading_helper.py, with modifications.
class catch_threading_exception:
"""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 set to an
object which is being finalized. Exiting the context manager clears the
stored object.
Usage:
with threading_helper.catch_threading_exception() as cm:
# code spawning a thread which raises an exception
...
# check the thread exception: use cm.args
...
# cm.args attribute no longer exists at this point
# (to break a reference cycle)
"""
def __init__(self) -> None:
# See https://github.com/python/typeshed/issues/4767 regarding the underscore.
self.args: Optional["threading._ExceptHookArgs"] = None
self._old_hook: Optional[Callable[["threading._ExceptHookArgs"], Any]] = None
def _hook(self, args: "threading._ExceptHookArgs") -> None:
self.args = args
def __enter__(self) -> "catch_threading_exception":
self._old_hook = threading.excepthook
threading.excepthook = self._hook
return self
def __exit__(
self,
exc_type: Optional[Type[BaseException]],
exc_val: Optional[BaseException],
exc_tb: Optional[TracebackType],
) -> None:
assert self._old_hook is not None
threading.excepthook = self._old_hook
self._old_hook = None
del self.args
def thread_exception_runtest_hook() -> Generator[None, None, None]:
with catch_threading_exception() as cm:
yield
if cm.args:
if cm.args.thread is not None:
thread_name = cm.args.thread.name
else:
thread_name = "<unknown>"
msg = f"Exception in thread {thread_name}\n\n"
msg += "".join(
traceback.format_exception(
cm.args.exc_type, cm.args.exc_value, cm.args.exc_traceback,
)
)
warnings.warn(pytest.PytestUnhandledThreadExceptionWarning(msg))
@pytest.hookimpl(hookwrapper=True, trylast=True)
def pytest_runtest_setup() -> Generator[None, None, None]:
yield from thread_exception_runtest_hook()
@pytest.hookimpl(hookwrapper=True, tryfirst=True)
def pytest_runtest_call() -> Generator[None, None, None]:
yield from thread_exception_runtest_hook()
@pytest.hookimpl(hookwrapper=True, tryfirst=True)
def pytest_runtest_teardown() -> Generator[None, None, None]:
yield from thread_exception_runtest_hook()

93
src/_pytest/unraisableexception.py Normal file
View File

@ -0,0 +1,93 @@
import sys
import traceback
import warnings
from types import TracebackType
from typing import Any
from typing import Callable
from typing import Generator
from typing import Optional
from typing import Type
import pytest
# Copied from cpython/Lib/test/support/__init__.py, with modifications.
class catch_unraisable_exception:
"""Context manager catching unraisable exception using sys.unraisablehook.
Storing the exception value (cm.unraisable.exc_value) creates a reference
cycle. The reference cycle is broken explicitly when the context manager
exits.
Storing the object (cm.unraisable.object) can resurrect it if it is set to
an object which is being finalized. Exiting the context manager clears the
stored object.
Usage:
with catch_unraisable_exception() as cm:
# code creating an "unraisable exception"
...
# check the unraisable exception: use cm.unraisable
...
# cm.unraisable attribute no longer exists at this point
# (to break a reference cycle)
"""
def __init__(self) -> None:
self.unraisable: Optional["sys.UnraisableHookArgs"] = None
self._old_hook: Optional[Callable[["sys.UnraisableHookArgs"], Any]] = None
def _hook(self, unraisable: "sys.UnraisableHookArgs") -> None:
# Storing unraisable.object can resurrect an object which is being
# finalized. Storing unraisable.exc_value creates a reference cycle.
self.unraisable = unraisable
def __enter__(self) -> "catch_unraisable_exception":
self._old_hook = sys.unraisablehook
sys.unraisablehook = self._hook
return self
def __exit__(
self,
exc_type: Optional[Type[BaseException]],
exc_val: Optional[BaseException],
exc_tb: Optional[TracebackType],
) -> None:
assert self._old_hook is not None
sys.unraisablehook = self._old_hook
self._old_hook = None
del self.unraisable
def unraisable_exception_runtest_hook() -> Generator[None, None, None]:
with catch_unraisable_exception() as cm:
yield
if cm.unraisable:
if cm.unraisable.err_msg is not None:
err_msg = cm.unraisable.err_msg
else:
err_msg = "Exception ignored in"
msg = f"{err_msg}: {cm.unraisable.object!r}\n\n"
msg += "".join(
traceback.format_exception(
cm.unraisable.exc_type,
cm.unraisable.exc_value,
cm.unraisable.exc_traceback,
)
)
warnings.warn(pytest.PytestUnraisableExceptionWarning(msg))
@pytest.hookimpl(hookwrapper=True, tryfirst=True)
def pytest_runtest_setup() -> Generator[None, None, None]:
yield from unraisable_exception_runtest_hook()
@pytest.hookimpl(hookwrapper=True, tryfirst=True)
def pytest_runtest_call() -> Generator[None, None, None]:
yield from unraisable_exception_runtest_hook()
@pytest.hookimpl(hookwrapper=True, tryfirst=True)
def pytest_runtest_teardown() -> Generator[None, None, None]:
yield from unraisable_exception_runtest_hook()

22
src/_pytest/warning_types.py
View File

@ -90,6 +90,28 @@ class PytestUnknownMarkWarning(PytestWarning):
__module__ = "pytest"
@final
class PytestUnraisableExceptionWarning(PytestWarning):
"""An unraisable exception was reported.
Unraisable exceptions are exceptions raised in :meth:`__del__ <object.__del__>`
implementations and similar situations when the exception cannot be raised
as normal.
"""
__module__ = "pytest"
@final
class PytestUnhandledThreadExceptionWarning(PytestWarning):
"""An unhandled exception occurred in a :class:`~threading.Thread`.
Such exceptions don't propagate normally.
"""
__module__ = "pytest"
_W = TypeVar("_W", bound=PytestWarning)

4
src/pytest/__init__.py
View File

@ -44,7 +44,9 @@ from _pytest.warning_types import PytestConfigWarning
from _pytest.warning_types import PytestDeprecationWarning
from _pytest.warning_types import PytestExperimentalApiWarning
from _pytest.warning_types import PytestUnhandledCoroutineWarning
from _pytest.warning_types import PytestUnhandledThreadExceptionWarning
from _pytest.warning_types import PytestUnknownMarkWarning
from _pytest.warning_types import PytestUnraisableExceptionWarning
from _pytest.warning_types import PytestWarning
set_trace = __pytestPDB.set_trace
@ -85,7 +87,9 @@ __all__ = [
"PytestDeprecationWarning",
"PytestExperimentalApiWarning",
"PytestUnhandledCoroutineWarning",
"PytestUnhandledThreadExceptionWarning",
"PytestUnknownMarkWarning",
"PytestUnraisableExceptionWarning",
"PytestWarning",
"raises",
"register_assert_rewrite",

137
testing/test_threadexception.py Normal file
View File

@ -0,0 +1,137 @@
import sys
import pytest
from _pytest.pytester import Pytester
if sys.version_info < (3, 8):
pytest.skip("threadexception plugin needs Python>=3.8", allow_module_level=True)
@pytest.mark.filterwarnings("default")
def test_unhandled_thread_exception(pytester: Pytester) -> None:
pytester.makepyfile(
test_it="""
import threading
def test_it():
def oops():
raise ValueError("Oops")
t = threading.Thread(target=oops, name="MyThread")
t.start()
t.join()
def test_2(): pass
"""
)
result = pytester.runpytest()
assert result.ret == 0
assert result.parseoutcomes() == {"passed": 2, "warnings": 1}
result.stdout.fnmatch_lines(
[
"*= warnings summary =*",
"test_it.py::test_it",
" * PytestUnhandledThreadExceptionWarning: Exception in thread MyThread",
" ",
" Traceback (most recent call last):",
" ValueError: Oops",
" ",
" warnings.warn(pytest.PytestUnhandledThreadExceptionWarning(msg))",
]
)
@pytest.mark.filterwarnings("default")
def test_unhandled_thread_exception_in_setup(pytester: Pytester) -> None:
pytester.makepyfile(
test_it="""
import threading
import pytest
@pytest.fixture
def threadexc():
def oops():
raise ValueError("Oops")
t = threading.Thread(target=oops, name="MyThread")
t.start()
t.join()
def test_it(threadexc): pass
def test_2(): pass
"""
)
result = pytester.runpytest()
assert result.ret == 0
assert result.parseoutcomes() == {"passed": 2, "warnings": 1}
result.stdout.fnmatch_lines(
[
"*= warnings summary =*",
"test_it.py::test_it",
" * PytestUnhandledThreadExceptionWarning: Exception in thread MyThread",
" ",
" Traceback (most recent call last):",
" ValueError: Oops",
" ",
" warnings.warn(pytest.PytestUnhandledThreadExceptionWarning(msg))",
]
)
@pytest.mark.filterwarnings("default")
def test_unhandled_thread_exception_in_teardown(pytester: Pytester) -> None:
pytester.makepyfile(
test_it="""
import threading
import pytest
@pytest.fixture
def threadexc():
def oops():
raise ValueError("Oops")
yield
t = threading.Thread(target=oops, name="MyThread")
t.start()
t.join()
def test_it(threadexc): pass
def test_2(): pass
"""
)
result = pytester.runpytest()
assert result.ret == 0
assert result.parseoutcomes() == {"passed": 2, "warnings": 1}
result.stdout.fnmatch_lines(
[
"*= warnings summary =*",
"test_it.py::test_it",
" * PytestUnhandledThreadExceptionWarning: Exception in thread MyThread",
" ",
" Traceback (most recent call last):",
" ValueError: Oops",
" ",
" warnings.warn(pytest.PytestUnhandledThreadExceptionWarning(msg))",
]
)
@pytest.mark.filterwarnings("error::pytest.PytestUnhandledThreadExceptionWarning")
def test_unhandled_thread_exception_warning_error(pytester: Pytester) -> None:
pytester.makepyfile(
test_it="""
import threading
import pytest
def test_it():
def oops():
raise ValueError("Oops")
t = threading.Thread(target=oops, name="MyThread")
t.start()
t.join()
def test_2(): pass
"""
)
result = pytester.runpytest()
assert result.ret == pytest.ExitCode.TESTS_FAILED
assert result.parseoutcomes() == {"passed": 1, "failed": 1}

133
testing/test_unraisableexception.py Normal file
View File

@ -0,0 +1,133 @@
import sys
import pytest
from _pytest.pytester import Pytester
if sys.version_info < (3, 8):
pytest.skip("unraisableexception plugin needs Python>=3.8", allow_module_level=True)
@pytest.mark.filterwarnings("default")
def test_unraisable(pytester: Pytester) -> None:
pytester.makepyfile(
test_it="""
class BrokenDel:
def __del__(self):
raise ValueError("del is broken")
def test_it():
obj = BrokenDel()
del obj
def test_2(): pass
"""
)
result = pytester.runpytest()
assert result.ret == 0
assert result.parseoutcomes() == {"passed": 2, "warnings": 1}
result.stdout.fnmatch_lines(
[
"*= warnings summary =*",
"test_it.py::test_it",
" * PytestUnraisableExceptionWarning: Exception ignored in: <function BrokenDel.__del__ at *>",
" ",
" Traceback (most recent call last):",
" ValueError: del is broken",
" ",
" warnings.warn(pytest.PytestUnraisableExceptionWarning(msg))",
]
)
@pytest.mark.filterwarnings("default")
def test_unraisable_in_setup(pytester: Pytester) -> None:
pytester.makepyfile(
test_it="""
import pytest
class BrokenDel:
def __del__(self):
raise ValueError("del is broken")
@pytest.fixture
def broken_del():
obj = BrokenDel()
del obj
def test_it(broken_del): pass
def test_2(): pass
"""
)
result = pytester.runpytest()
assert result.ret == 0
assert result.parseoutcomes() == {"passed": 2, "warnings": 1}
result.stdout.fnmatch_lines(
[
"*= warnings summary =*",
"test_it.py::test_it",
" * PytestUnraisableExceptionWarning: Exception ignored in: <function BrokenDel.__del__ at *>",
" ",
" Traceback (most recent call last):",
" ValueError: del is broken",
" ",
" warnings.warn(pytest.PytestUnraisableExceptionWarning(msg))",
]
)
@pytest.mark.filterwarnings("default")
def test_unraisable_in_teardown(pytester: Pytester) -> None:
pytester.makepyfile(
test_it="""
import pytest
class BrokenDel:
def __del__(self):
raise ValueError("del is broken")
@pytest.fixture
def broken_del():
yield
obj = BrokenDel()
del obj
def test_it(broken_del): pass
def test_2(): pass
"""
)
result = pytester.runpytest()
assert result.ret == 0
assert result.parseoutcomes() == {"passed": 2, "warnings": 1}
result.stdout.fnmatch_lines(
[
"*= warnings summary =*",
"test_it.py::test_it",
" * PytestUnraisableExceptionWarning: Exception ignored in: <function BrokenDel.__del__ at *>",
" ",
" Traceback (most recent call last):",
" ValueError: del is broken",
" ",
" warnings.warn(pytest.PytestUnraisableExceptionWarning(msg))",
]
)
@pytest.mark.filterwarnings("error::pytest.PytestUnraisableExceptionWarning")
def test_unraisable_warning_error(pytester: Pytester) -> None:
pytester.makepyfile(
test_it="""
class BrokenDel:
def __del__(self) -> None:
raise ValueError("del is broken")
def test_it() -> None:
obj = BrokenDel()
del obj
def test_2(): pass
"""
)
result = pytester.runpytest()
assert result.ret == pytest.ExitCode.TESTS_FAILED
assert result.parseoutcomes() == {"passed": 1, "failed": 1}