python · ericsnowcurrently
diff --git a/Lib/test/test__xxsubinterpreters.py b/Lib/test/test__xxsubinterpreters.py
@@ -1,3 +1,4 @@
+import builtins
 from collections import namedtuple
 import contextlib
 import itertools
@@ -885,10 +886,11 @@ def assert_run_failed(self, exctype, msg=None):
             yield
         if msg is None:
             self.assertEqual(str(caught.exception).split(':')[0],
-                             str(exctype))
+                             exctype.__name__)
         else:
             self.assertEqual(str(caught.exception),
-                             "{}: {}".format(exctype, msg))
+                             "{}: {}".format(exctype.__name__, msg))
+        self.assertIsInstance(caught.exception.__cause__, exctype)
 
     def test_invalid_syntax(self):
         with self.assert_run_failed(SyntaxError):
@@ -1079,6 +1081,301 @@ def f():
         self.assertEqual(retcode, 0)
 
 
+def build_exception(exctype, /, *args, **kwargs):
+    # XXX Use __qualname__?
+    name = exctype.__name__
+    argreprs = [repr(a) for a in args]
+    if kwargs:
+        kwargreprs = [f'{k}={v!r}' for k, v in kwargs.items()]
+        script = f'{name}({", ".join(argreprs)}, {", ".join(kwargreprs)})'
+    else:
+        script = f'{name}({", ".join(argreprs)})'
+    expected = exctype(*args, **kwargs)
+    return script, expected
+
+
+def build_exceptions(self, *exctypes, default=None, custom=None, bases=True):
+    if not exctypes:
+        raise NotImplementedError
+    if not default:
+        default = ((), {})
+    elif isinstance(default, str):
+        default = ((default,), {})
+    elif type(default) is not tuple:
+        raise NotImplementedError
+    elif len(default) != 2:
+        default = (default, {})
+    elif type(default[0]) is not tuple:
+        default = (default, {})
+    elif type(default[1]) is not dict:
+        default = (default, {})
+    # else leave it alone
+
+    for exctype in exctypes:
+        customtype = None
+        values = default
+        if custom:
+            if exctype in custom:
+                customtype = exctype
+            elif bases:
+                for customtype in custom:
+                    if issubclass(exctype, customtype):
+                        break
+                else:
+                    customtype = None
+        if customtype is not None:
+            values = custom[customtype]
+            if values is None:
+                continue
+        args, kwargs = values
+        script, expected = build_exception(exctype, *args, **kwargs)
+        yield exctype, customtype, script, expected
+
+
+try:
+    raise Exception
+except Exception as exc:
+    assert exc.__traceback__ is not None
+    Traceback = type(exc.__traceback__)
+
+
+class RunFailedTests(TestBase):
+
+    BUILTINS = [v
+                for v in vars(builtins).values()
+                if (type(v) is type
+                    and issubclass(v, Exception)
+                    #and issubclass(v, BaseException)
+                    )
+                ]
+    BUILTINS_SPECIAL = [
+        # These all have extra attributes (i.e. args/kwargs)
+        SyntaxError,
+        ImportError,
+        UnicodeError,
+        OSError,
+        SystemExit,
+        StopIteration,
+    ]
+
+    @classmethod
+    def build_exceptions(cls, exctypes=None, default=(), custom=None):
+        if exctypes is None:
+            exctypes = cls.BUILTINS
+            if custom is None:
+                # Skip the "special" ones.
+                custom = {et: None for et in cls.BUILTINS_SPECIAL}
+        yield from build_exceptions(*exctypes, default=default, custom=custom)
+
+    def assertExceptionsEqual(self, exc, expected, *, chained=True):
+        if type(expected) is type:
+            self.assertIs(type(exc), expected)
+            return
+        elif not isinstance(exc, Exception):
+            self.assertEqual(exc, expected)
+        elif not isinstance(expected, Exception):
+            self.assertEqual(exc, expected)
+        else:
+            # Plain equality doesn't work, so we have to compare manually.
+            self.assertIs(type(exc), type(expected))
+            self.assertEqual(exc.args, expected.args)
+            self.assertEqual(exc.__reduce__(), expected.__reduce__())
+            if chained:
+                self.assertExceptionsEqual(exc.__context__,
+                                           expected.__context__)
+                self.assertExceptionsEqual(exc.__cause__,
+                                           expected.__cause__)
+                self.assertEqual(exc.__suppress_context__,
+                                 expected.__suppress_context__)
+
+    def assertTracebacksEqual(self, tb, expected):
+        if not isinstance(tb, Traceback):
+            self.assertEqual(tb, expected)
+        elif not isinstance(expected, Traceback):
+            self.assertEqual(tb, expected)
+        else:
+            self.assertEqual(tb.tb_frame.f_code.co_name,
+                             expected.tb_frame.f_code.co_name)
+            self.assertEqual(tb.tb_frame.f_code.co_filename,
+                             expected.tb_frame.f_code.co_filename)
+            self.assertEqual(tb.tb_lineno, expected.tb_lineno)
+            self.assertTracebacksEqual(tb.tb_next, expected.tb_next)
+
+    # XXX Move this to TestBase?
+    @contextlib.contextmanager
+    def expected_run_failure(self, expected):
+        exctype = expected if type(expected) is type else type(expected)
+
+        with self.assertRaises(interpreters.RunFailedError) as caught:
+            yield caught
+        exc = caught.exception
+
+        modname = exctype.__module__
+        if modname == 'builtins' or modname == '__main__':
+            exctypename = exctype.__name__
+        else:
+            exctypename = f'{modname}.{exctype.__name__}'
+        if exctype is expected:
+            self.assertEqual(str(exc).split(':')[0], exctypename)
+        else:
+            self.assertEqual(str(exc), f'{exctypename}: {expected}')
+        self.assertExceptionsEqual(exc.__cause__, expected)
+        if exc.__cause__ is not None:
+            self.assertIsNotNone(exc.__cause__.__traceback__)
+
+    def test_builtin_exceptions(self):
+        interpid = interpreters.create()
+        msg = '<a message>'
+        for i, info in enumerate(self.build_exceptions(
+            default=msg,
+            custom={
+                SyntaxError: ((msg, '<stdin>', 1, 3, 'a +?'), {}),
+                ImportError: ((msg,), {'name': 'spam', 'path': '/x/spam.py'}),
+                UnicodeError: None,
+                #UnicodeError: ((), {}),
+                #OSError: ((), {}),
+                SystemExit: ((1,), {}),
+                StopIteration: (('<a value>',), {}),
+            },
+        )):
+            exctype, _, script, expected = info
+            testname = f'{i+1} - {script}'
+            script = f'raise {script}'
+
+            with self.subTest(testname):
+                with self.expected_run_failure(expected):
+                    interpreters.run_string(interpid, script)
+
+    def test_custom_exception_from___main__(self):
+        script = dedent("""
+            class SpamError(Exception):
+                def __init__(self, q):
+                    super().__init__(f'got {q}')
+                    self.q = q
+            raise SpamError('eggs')
+            """)
+        expected = Exception(f'SpamError: got {"eggs"}')
+
+        interpid = interpreters.create()
+        with self.assertRaises(interpreters.RunFailedError) as caught:
+            interpreters.run_string(interpid, script)
+        cause = caught.exception.__cause__
+
+        self.assertExceptionsEqual(cause, expected)
+
+    class SpamError(Exception):
+        # The normal Exception.__reduce__() produces a funny result
+        # here.  So we have to use a custom __new__().
+        def __new__(cls, q):
+            if type(q) is SpamError:
+                return q
+            return super().__new__(cls, q)
+        def __init__(self, q):
+            super().__init__(f'got {q}')
+            self.q = q
+
+    def test_custom_exception(self):
+        script = dedent("""
+            import test.test__xxsubinterpreters
+            SpamError = test.test__xxsubinterpreters.RunFailedTests.SpamError
+            raise SpamError('eggs')
+            """)
+        try:
+            ns = {}
+            exec(script, ns, ns)
+        except Exception as exc:
+            expected = exc
+
+        interpid = interpreters.create()
+        with self.expected_run_failure(expected):
+            interpreters.run_string(interpid, script)
+
+    class SpamReducedError(Exception):
+        def __init__(self, q):
+            super().__init__(f'got {q}')
+            self.q = q
+        def __reduce__(self):
+            return (type(self), (self.q,), {})
+
+    def test_custom___reduce__(self):
+        script = dedent("""
+            import test.test__xxsubinterpreters
+            SpamError = test.test__xxsubinterpreters.RunFailedTests.SpamReducedError
+            raise SpamError('eggs')
+            """)
+        try:
+            exec(script, (ns := {'__name__': '__main__'}), ns)
+        except Exception as exc:
+            expected = exc
+
+        interpid = interpreters.create()
+        with self.expected_run_failure(expected):
+            interpreters.run_string(interpid, script)
+
+    def test_traceback_propagated(self):
+        script = dedent("""
+            def do_spam():
+                raise Exception('uh-oh')
+            def do_eggs():
+                return do_spam()
+            class Spam:
+                def do(self):
+                    return do_eggs()
+            def get_handler():
+                def handler():
+                    return Spam().do()
+                return handler
+            go = (lambda: get_handler()())
+            def iter_all():
+                yield from (go() for _ in [True])
+                yield None
+            def main():
+                for v in iter_all():
+                    pass
+            main()
+            """)
+        try:
+            ns = {}
+            exec(script, ns, ns)
+        except Exception as exc:
+            expected = exc
+            expectedtb = exc.__traceback__.tb_next
+
+        interpid = interpreters.create()
+        with self.expected_run_failure(expected) as caught:
+            interpreters.run_string(interpid, script)
+        exc = caught.exception
+
+        self.assertTracebacksEqual(exc.__cause__.__traceback__,
+                                   expectedtb)
+
+    def test_chained_exceptions(self):
+        script = dedent("""
+            try:
+                raise ValueError('msg 1')
+            except Exception as exc1:
+                try:
+                    raise TypeError('msg 2')
+                except Exception as exc2:
+                    try:
+                        raise IndexError('msg 3') from exc2
+                    except Exception:
+                        raise AttributeError('msg 4')
+            """)
+        try:
+            exec(script, {}, {})
+        except Exception as exc:
+            expected = exc
+
+        interpid = interpreters.create()
+        with self.expected_run_failure(expected) as caught:
+            interpreters.run_string(interpid, script)
+        exc = caught.exception
+
+        # ...just to be sure.
+        self.assertIs(type(exc.__cause__), AttributeError)
+
+
 ##################################
 # channel tests