import gc
|
import logging
|
import os
|
import subprocess
|
from pathlib import Path
|
|
import pytest
|
|
from traceback import (
|
extract_tb,
|
print_exception,
|
format_exception,
|
)
|
from traceback import _cause_message # type: ignore
|
import sys
|
import re
|
|
from .tutil import slow
|
from .._multierror import MultiError, concat_tb, NonBaseMultiError
|
from ... import TrioDeprecationWarning
|
from ..._core import open_nursery
|
|
if sys.version_info < (3, 11):
|
from exceptiongroup import ExceptionGroup
|
|
|
class NotHashableException(Exception):
|
code = None
|
|
def __init__(self, code):
|
super().__init__()
|
self.code = code
|
|
def __eq__(self, other):
|
if not isinstance(other, NotHashableException):
|
return False
|
return self.code == other.code
|
|
|
async def raise_nothashable(code):
|
raise NotHashableException(code)
|
|
|
def raiser1():
|
raiser1_2()
|
|
|
def raiser1_2():
|
raiser1_3()
|
|
|
def raiser1_3():
|
raise ValueError("raiser1_string")
|
|
|
def raiser2():
|
raiser2_2()
|
|
|
def raiser2_2():
|
raise KeyError("raiser2_string")
|
|
|
def raiser3():
|
raise NameError
|
|
|
def get_exc(raiser):
|
try:
|
raiser()
|
except Exception as exc:
|
return exc
|
|
|
def get_tb(raiser):
|
return get_exc(raiser).__traceback__
|
|
|
def test_concat_tb():
|
|
tb1 = get_tb(raiser1)
|
tb2 = get_tb(raiser2)
|
|
# These return a list of (filename, lineno, fn name, text) tuples
|
# https://docs.python.org/3/library/traceback.html#traceback.extract_tb
|
entries1 = extract_tb(tb1)
|
entries2 = extract_tb(tb2)
|
|
tb12 = concat_tb(tb1, tb2)
|
assert extract_tb(tb12) == entries1 + entries2
|
|
tb21 = concat_tb(tb2, tb1)
|
assert extract_tb(tb21) == entries2 + entries1
|
|
# Check degenerate cases
|
assert extract_tb(concat_tb(None, tb1)) == entries1
|
assert extract_tb(concat_tb(tb1, None)) == entries1
|
assert concat_tb(None, None) is None
|
|
# Make sure the original tracebacks didn't get mutated by mistake
|
assert extract_tb(get_tb(raiser1)) == entries1
|
assert extract_tb(get_tb(raiser2)) == entries2
|
|
|
def test_MultiError():
|
exc1 = get_exc(raiser1)
|
exc2 = get_exc(raiser2)
|
|
assert MultiError([exc1]) is exc1
|
m = MultiError([exc1, exc2])
|
assert m.exceptions == (exc1, exc2)
|
assert "ValueError" in str(m)
|
assert "ValueError" in repr(m)
|
|
with pytest.raises(TypeError):
|
MultiError(object())
|
with pytest.raises(TypeError):
|
MultiError([KeyError(), ValueError])
|
|
|
def test_MultiErrorOfSingleMultiError():
|
# For MultiError([MultiError]), ensure there is no bad recursion by the
|
# constructor where __init__ is called if __new__ returns a bare MultiError.
|
exceptions = (KeyError(), ValueError())
|
a = MultiError(exceptions)
|
b = MultiError([a])
|
assert b == a
|
assert b.exceptions == exceptions
|
|
|
async def test_MultiErrorNotHashable():
|
exc1 = NotHashableException(42)
|
exc2 = NotHashableException(4242)
|
exc3 = ValueError()
|
assert exc1 != exc2
|
assert exc1 != exc3
|
|
with pytest.raises(MultiError):
|
async with open_nursery() as nursery:
|
nursery.start_soon(raise_nothashable, 42)
|
nursery.start_soon(raise_nothashable, 4242)
|
|
|
def test_MultiError_filter_NotHashable():
|
excs = MultiError([NotHashableException(42), ValueError()])
|
|
def handle_ValueError(exc):
|
if isinstance(exc, ValueError):
|
return None
|
else:
|
return exc
|
|
with pytest.warns(TrioDeprecationWarning):
|
filtered_excs = MultiError.filter(handle_ValueError, excs)
|
|
assert isinstance(filtered_excs, NotHashableException)
|
|
|
def make_tree():
|
# Returns an object like:
|
# MultiError([
|
# MultiError([
|
# ValueError,
|
# KeyError,
|
# ]),
|
# NameError,
|
# ])
|
# where all exceptions except the root have a non-trivial traceback.
|
exc1 = get_exc(raiser1)
|
exc2 = get_exc(raiser2)
|
exc3 = get_exc(raiser3)
|
|
# Give m12 a non-trivial traceback
|
try:
|
raise MultiError([exc1, exc2])
|
except BaseException as m12:
|
return MultiError([m12, exc3])
|
|
|
def assert_tree_eq(m1, m2):
|
if m1 is None or m2 is None:
|
assert m1 is m2
|
return
|
assert type(m1) is type(m2)
|
assert extract_tb(m1.__traceback__) == extract_tb(m2.__traceback__)
|
assert_tree_eq(m1.__cause__, m2.__cause__)
|
assert_tree_eq(m1.__context__, m2.__context__)
|
if isinstance(m1, MultiError):
|
assert len(m1.exceptions) == len(m2.exceptions)
|
for e1, e2 in zip(m1.exceptions, m2.exceptions):
|
assert_tree_eq(e1, e2)
|
|
|
def test_MultiError_filter():
|
def null_handler(exc):
|
return exc
|
|
m = make_tree()
|
assert_tree_eq(m, m)
|
with pytest.warns(TrioDeprecationWarning):
|
assert MultiError.filter(null_handler, m) is m
|
|
assert_tree_eq(m, make_tree())
|
|
# Make sure we don't pick up any detritus if run in a context where
|
# implicit exception chaining would like to kick in
|
m = make_tree()
|
try:
|
raise ValueError
|
except ValueError:
|
with pytest.warns(TrioDeprecationWarning):
|
assert MultiError.filter(null_handler, m) is m
|
assert_tree_eq(m, make_tree())
|
|
def simple_filter(exc):
|
if isinstance(exc, ValueError):
|
return None
|
if isinstance(exc, KeyError):
|
return RuntimeError()
|
return exc
|
|
with pytest.warns(TrioDeprecationWarning):
|
new_m = MultiError.filter(simple_filter, make_tree())
|
|
assert isinstance(new_m, MultiError)
|
assert len(new_m.exceptions) == 2
|
# was: [[ValueError, KeyError], NameError]
|
# ValueError disappeared & KeyError became RuntimeError, so now:
|
assert isinstance(new_m.exceptions[0], RuntimeError)
|
assert isinstance(new_m.exceptions[1], NameError)
|
|
# implicit chaining:
|
assert isinstance(new_m.exceptions[0].__context__, KeyError)
|
|
# also, the traceback on the KeyError incorporates what used to be the
|
# traceback on its parent MultiError
|
orig = make_tree()
|
# make sure we have the right path
|
assert isinstance(orig.exceptions[0].exceptions[1], KeyError)
|
# get original traceback summary
|
orig_extracted = (
|
extract_tb(orig.__traceback__)
|
+ extract_tb(orig.exceptions[0].__traceback__)
|
+ extract_tb(orig.exceptions[0].exceptions[1].__traceback__)
|
)
|
|
def p(exc):
|
print_exception(type(exc), exc, exc.__traceback__)
|
|
p(orig)
|
p(orig.exceptions[0])
|
p(orig.exceptions[0].exceptions[1])
|
p(new_m.exceptions[0].__context__)
|
# compare to the new path
|
assert new_m.__traceback__ is None
|
new_extracted = extract_tb(new_m.exceptions[0].__context__.__traceback__)
|
assert orig_extracted == new_extracted
|
|
# check preserving partial tree
|
def filter_NameError(exc):
|
if isinstance(exc, NameError):
|
return None
|
return exc
|
|
m = make_tree()
|
with pytest.warns(TrioDeprecationWarning):
|
new_m = MultiError.filter(filter_NameError, m)
|
# with the NameError gone, the other branch gets promoted
|
assert new_m is m.exceptions[0]
|
|
# check fully handling everything
|
def filter_all(exc):
|
return None
|
|
with pytest.warns(TrioDeprecationWarning):
|
assert MultiError.filter(filter_all, make_tree()) is None
|
|
|
def test_MultiError_catch():
|
# No exception to catch
|
|
def noop(_):
|
pass # pragma: no cover
|
|
with pytest.warns(TrioDeprecationWarning), MultiError.catch(noop):
|
pass
|
|
# Simple pass-through of all exceptions
|
m = make_tree()
|
with pytest.raises(MultiError) as excinfo:
|
with pytest.warns(TrioDeprecationWarning), MultiError.catch(lambda exc: exc):
|
raise m
|
assert excinfo.value is m
|
# Should be unchanged, except that we added a traceback frame by raising
|
# it here
|
assert m.__traceback__ is not None
|
assert m.__traceback__.tb_frame.f_code.co_name == "test_MultiError_catch"
|
assert m.__traceback__.tb_next is None
|
m.__traceback__ = None
|
assert_tree_eq(m, make_tree())
|
|
# Swallows everything
|
with pytest.warns(TrioDeprecationWarning), MultiError.catch(lambda _: None):
|
raise make_tree()
|
|
def simple_filter(exc):
|
if isinstance(exc, ValueError):
|
return None
|
if isinstance(exc, KeyError):
|
return RuntimeError()
|
return exc
|
|
with pytest.raises(MultiError) as excinfo:
|
with pytest.warns(TrioDeprecationWarning), MultiError.catch(simple_filter):
|
raise make_tree()
|
new_m = excinfo.value
|
assert isinstance(new_m, MultiError)
|
assert len(new_m.exceptions) == 2
|
# was: [[ValueError, KeyError], NameError]
|
# ValueError disappeared & KeyError became RuntimeError, so now:
|
assert isinstance(new_m.exceptions[0], RuntimeError)
|
assert isinstance(new_m.exceptions[1], NameError)
|
# Make sure that Python did not successfully attach the old MultiError to
|
# our new MultiError's __context__
|
assert not new_m.__suppress_context__
|
assert new_m.__context__ is None
|
|
# check preservation of __cause__ and __context__
|
v = ValueError()
|
v.__cause__ = KeyError()
|
with pytest.raises(ValueError) as excinfo:
|
with pytest.warns(TrioDeprecationWarning), MultiError.catch(lambda exc: exc):
|
raise v
|
assert isinstance(excinfo.value.__cause__, KeyError)
|
|
v = ValueError()
|
context = KeyError()
|
v.__context__ = context
|
with pytest.raises(ValueError) as excinfo:
|
with pytest.warns(TrioDeprecationWarning), MultiError.catch(lambda exc: exc):
|
raise v
|
assert excinfo.value.__context__ is context
|
assert not excinfo.value.__suppress_context__
|
|
for suppress_context in [True, False]:
|
v = ValueError()
|
context = KeyError()
|
v.__context__ = context
|
v.__suppress_context__ = suppress_context
|
distractor = RuntimeError()
|
with pytest.raises(ValueError) as excinfo:
|
|
def catch_RuntimeError(exc):
|
if isinstance(exc, RuntimeError):
|
return None
|
else:
|
return exc
|
|
with pytest.warns(TrioDeprecationWarning):
|
with MultiError.catch(catch_RuntimeError):
|
raise MultiError([v, distractor])
|
assert excinfo.value.__context__ is context
|
assert excinfo.value.__suppress_context__ == suppress_context
|
|
|
@pytest.mark.skipif(
|
sys.implementation.name != "cpython", reason="Only makes sense with refcounting GC"
|
)
|
def test_MultiError_catch_doesnt_create_cyclic_garbage():
|
# https://github.com/python-trio/trio/pull/2063
|
gc.collect()
|
old_flags = gc.get_debug()
|
|
def make_multi():
|
# make_tree creates cycles itself, so a simple
|
raise MultiError([get_exc(raiser1), get_exc(raiser2)])
|
|
def simple_filter(exc):
|
if isinstance(exc, ValueError):
|
return Exception()
|
if isinstance(exc, KeyError):
|
return RuntimeError()
|
assert False, "only ValueError and KeyError should exist" # pragma: no cover
|
|
try:
|
gc.set_debug(gc.DEBUG_SAVEALL)
|
with pytest.raises(MultiError):
|
# covers MultiErrorCatcher.__exit__ and _multierror.copy_tb
|
with pytest.warns(TrioDeprecationWarning), MultiError.catch(simple_filter):
|
raise make_multi()
|
gc.collect()
|
assert not gc.garbage
|
finally:
|
gc.set_debug(old_flags)
|
gc.garbage.clear()
|
|
|
def assert_match_in_seq(pattern_list, string):
|
offset = 0
|
print("looking for pattern matches...")
|
for pattern in pattern_list:
|
print("checking pattern:", pattern)
|
reobj = re.compile(pattern)
|
match = reobj.search(string, offset)
|
assert match is not None
|
offset = match.end()
|
|
|
def test_assert_match_in_seq():
|
assert_match_in_seq(["a", "b"], "xx a xx b xx")
|
assert_match_in_seq(["b", "a"], "xx b xx a xx")
|
with pytest.raises(AssertionError):
|
assert_match_in_seq(["a", "b"], "xx b xx a xx")
|
|
|
def test_base_multierror():
|
"""
|
Test that MultiError() with at least one base exception will return a MultiError
|
object.
|
"""
|
|
exc = MultiError([ZeroDivisionError(), KeyboardInterrupt()])
|
assert type(exc) is MultiError
|
|
|
def test_non_base_multierror():
|
"""
|
Test that MultiError() without base exceptions will return a NonBaseMultiError
|
object.
|
"""
|
|
exc = MultiError([ZeroDivisionError(), ValueError()])
|
assert type(exc) is NonBaseMultiError
|
assert isinstance(exc, ExceptionGroup)
|
|
|
def run_script(name, use_ipython=False):
|
import trio
|
|
trio_path = Path(trio.__file__).parent.parent
|
script_path = Path(__file__).parent / "test_multierror_scripts" / name
|
|
env = dict(os.environ)
|
print("parent PYTHONPATH:", env.get("PYTHONPATH"))
|
if "PYTHONPATH" in env: # pragma: no cover
|
pp = env["PYTHONPATH"].split(os.pathsep)
|
else:
|
pp = []
|
pp.insert(0, str(trio_path))
|
pp.insert(0, str(script_path.parent))
|
env["PYTHONPATH"] = os.pathsep.join(pp)
|
print("subprocess PYTHONPATH:", env.get("PYTHONPATH"))
|
|
if use_ipython:
|
lines = [script_path.read_text(), "exit()"]
|
|
cmd = [
|
sys.executable,
|
"-u",
|
"-m",
|
"IPython",
|
# no startup files
|
"--quick",
|
"--TerminalIPythonApp.code_to_run=" + "\n".join(lines),
|
]
|
else:
|
cmd = [sys.executable, "-u", str(script_path)]
|
print("running:", cmd)
|
completed = subprocess.run(
|
cmd, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT
|
)
|
print("process output:")
|
print(completed.stdout.decode("utf-8"))
|
return completed
|
|
|
def check_simple_excepthook(completed):
|
assert_match_in_seq(
|
[
|
"in <module>",
|
"MultiError",
|
"--- 1 ---",
|
"in exc1_fn",
|
"ValueError",
|
"--- 2 ---",
|
"in exc2_fn",
|
"KeyError",
|
],
|
completed.stdout.decode("utf-8"),
|
)
|
|
|
try:
|
import IPython
|
except ImportError: # pragma: no cover
|
have_ipython = False
|
else:
|
have_ipython = True
|
|
need_ipython = pytest.mark.skipif(not have_ipython, reason="need IPython")
|
|
|
@slow
|
@need_ipython
|
def test_ipython_exc_handler():
|
completed = run_script("simple_excepthook.py", use_ipython=True)
|
check_simple_excepthook(completed)
|
|
|
@slow
|
@need_ipython
|
def test_ipython_imported_but_unused():
|
completed = run_script("simple_excepthook_IPython.py")
|
check_simple_excepthook(completed)
|
|
|
@slow
|
@need_ipython
|
def test_ipython_custom_exc_handler():
|
# Check we get a nice warning (but only one!) if the user is using IPython
|
# and already has some other set_custom_exc handler installed.
|
completed = run_script("ipython_custom_exc.py", use_ipython=True)
|
assert_match_in_seq(
|
[
|
# The warning
|
"RuntimeWarning",
|
"IPython detected",
|
"skip installing Trio",
|
# The MultiError
|
"MultiError",
|
"ValueError",
|
"KeyError",
|
],
|
completed.stdout.decode("utf-8"),
|
)
|
# Make sure our other warning doesn't show up
|
assert "custom sys.excepthook" not in completed.stdout.decode("utf-8")
|
|
|
@slow
|
@pytest.mark.skipif(
|
not Path("/usr/lib/python3/dist-packages/apport_python_hook.py").exists(),
|
reason="need Ubuntu with python3-apport installed",
|
)
|
def test_apport_excepthook_monkeypatch_interaction():
|
completed = run_script("apport_excepthook.py")
|
stdout = completed.stdout.decode("utf-8")
|
|
# No warning
|
assert "custom sys.excepthook" not in stdout
|
|
# Proper traceback
|
assert_match_in_seq(
|
["--- 1 ---", "KeyError", "--- 2 ---", "ValueError"],
|
stdout,
|
)
|
