When an exception is uncaught in Interpreter.exec_sync(), it helps to show that exception's error display if uncaught in the calling interpreter. We do so here by generating a TracebackException in the subinterpreter and passing it between interpreters using pickle.
const char *module;
} type;
const char *msg;
+ const char *pickled;
+ Py_ssize_t pickled_len;
} _PyXI_excinfo;
raise AttributeError(name)
+_EXEC_FAILURE_STR = """
+{superstr}
+
+Uncaught in the interpreter:
+
+{formatted}
+""".strip()
+
class ExecFailure(RuntimeError):
def __init__(self, excinfo):
msg = excinfo.formatted
if not msg:
- if excinfo.type and snapshot.msg:
- msg = f'{snapshot.type.__name__}: {snapshot.msg}'
+ if excinfo.type and excinfo.msg:
+ msg = f'{excinfo.type.__name__}: {excinfo.msg}'
else:
- msg = snapshot.type.__name__ or snapshot.msg
+ msg = excinfo.type.__name__ or excinfo.msg
super().__init__(msg)
- self.snapshot = excinfo
+ self.excinfo = excinfo
+
+ def __str__(self):
+ try:
+ formatted = ''.join(self.excinfo.tbexc.format()).rstrip()
+ except Exception:
+ return super().__str__()
+ else:
+ return _EXEC_FAILURE_STR.format(
+ superstr=super().__str__(),
+ formatted=formatted,
+ )
def create():
with self.assertRaises(interpreters.ExecFailure):
interp.exec_sync('raise Exception')
+ def test_display_preserved_exception(self):
+ tempdir = self.temp_dir()
+ modfile = self.make_module('spam', tempdir, text="""
+ def ham():
+ raise RuntimeError('uh-oh!')
+
+ def eggs():
+ ham()
+ """)
+ scriptfile = self.make_script('script.py', tempdir, text="""
+ from test.support import interpreters
+
+ def script():
+ import spam
+ spam.eggs()
+
+ interp = interpreters.create()
+ interp.exec_sync(script)
+ """)
+
+ stdout, stderr = self.assert_python_failure(scriptfile)
+ self.maxDiff = None
+ interpmod_line, = (l for l in stderr.splitlines() if ' exec_sync' in l)
+ # File "{interpreters.__file__}", line 179, in exec_sync
+ self.assertEqual(stderr, dedent(f"""\
+ Traceback (most recent call last):
+ File "{scriptfile}", line 9, in <module>
+ interp.exec_sync(script)
+ ~~~~~~~~~~~~~~~~^^^^^^^^
+ {interpmod_line.strip()}
+ raise ExecFailure(excinfo)
+ test.support.interpreters.ExecFailure: RuntimeError: uh-oh!
+
+ Uncaught in the interpreter:
+
+ Traceback (most recent call last):
+ File "{scriptfile}", line 6, in script
+ spam.eggs()
+ ~~~~~~~~~^^
+ File "{modfile}", line 6, in eggs
+ ham()
+ ~~~^^
+ File "{modfile}", line 3, in ham
+ raise RuntimeError('uh-oh!')
+ RuntimeError: uh-oh!
+ """))
+ self.assertEqual(stdout, '')
+
def test_in_thread(self):
interp = interpreters.create()
script, file = _captured_script('print("it worked!", end="")')
import contextlib
import os
+import os.path
+import subprocess
+import sys
+import tempfile
import threading
from textwrap import dedent
import unittest
+from test import support
+from test.support import os_helper
+
from test.support import interpreters
self.addCleanup(lambda: ensure_closed(w))
return r, w
+ def temp_dir(self):
+ tempdir = tempfile.mkdtemp()
+ tempdir = os.path.realpath(tempdir)
+ self.addCleanup(lambda: os_helper.rmtree(tempdir))
+ return tempdir
+
+ def make_script(self, filename, dirname=None, text=None):
+ if text:
+ text = dedent(text)
+ if dirname is None:
+ dirname = self.temp_dir()
+ filename = os.path.join(dirname, filename)
+
+ os.makedirs(os.path.dirname(filename), exist_ok=True)
+ with open(filename, 'w', encoding='utf-8') as outfile:
+ outfile.write(text or '')
+ return filename
+
+ def make_module(self, name, pathentry=None, text=None):
+ if text:
+ text = dedent(text)
+ if pathentry is None:
+ pathentry = self.temp_dir()
+ else:
+ os.makedirs(pathentry, exist_ok=True)
+ *subnames, basename = name.split('.')
+
+ dirname = pathentry
+ for subname in subnames:
+ dirname = os.path.join(dirname, subname)
+ if os.path.isdir(dirname):
+ pass
+ elif os.path.exists(dirname):
+ raise Exception(dirname)
+ else:
+ os.mkdir(dirname)
+ initfile = os.path.join(dirname, '__init__.py')
+ if not os.path.exists(initfile):
+ with open(initfile, 'w'):
+ pass
+ filename = os.path.join(dirname, basename + '.py')
+
+ with open(filename, 'w', encoding='utf-8') as outfile:
+ outfile.write(text or '')
+ return filename
+
+ @support.requires_subprocess()
+ def run_python(self, *argv):
+ proc = subprocess.run(
+ [sys.executable, *argv],
+ capture_output=True,
+ text=True,
+ )
+ return proc.returncode, proc.stdout, proc.stderr
+
+ def assert_python_ok(self, *argv):
+ exitcode, stdout, stderr = self.run_python(*argv)
+ self.assertNotEqual(exitcode, 1)
+ return stdout, stderr
+
+ def assert_python_failure(self, *argv):
+ exitcode, stdout, stderr = self.run_python(*argv)
+ self.assertNotEqual(exitcode, 0)
+ return stdout, stderr
+
def tearDown(self):
clean_up_interpreters()
/* convenience utilities */
/*************************/
+static const char *
+_copy_raw_string(const char *str, Py_ssize_t len)
+{
+ size_t size = len + 1;
+ if (len <= 0) {
+ size = strlen(str) + 1;
+ }
+ char *copied = PyMem_RawMalloc(size);
+ if (copied == NULL) {
+ return NULL;
+ }
+ if (len <= 0) {
+ strcpy(copied, str);
+ }
+ else {
+ memcpy(copied, str, size);
+ }
+ return copied;
+}
+
static const char *
_copy_string_obj_raw(PyObject *strobj)
{
return copied;
}
+
+static int
+_pickle_object(PyObject *obj, const char **p_pickled, Py_ssize_t *p_len)
+{
+ assert(!PyErr_Occurred());
+ PyObject *picklemod = PyImport_ImportModule("_pickle");
+ if (picklemod == NULL) {
+ PyErr_Clear();
+ picklemod = PyImport_ImportModule("pickle");
+ if (picklemod == NULL) {
+ return -1;
+ }
+ }
+ PyObject *dumps = PyObject_GetAttrString(picklemod, "dumps");
+ Py_DECREF(picklemod);
+ if (dumps == NULL) {
+ return -1;
+ }
+ PyObject *pickledobj = PyObject_CallOneArg(dumps, obj);
+ Py_DECREF(dumps);
+ if (pickledobj == NULL) {
+ return -1;
+ }
+
+ char *pickled = NULL;
+ Py_ssize_t len = 0;
+ if (PyBytes_AsStringAndSize(pickledobj, &pickled, &len) < 0) {
+ Py_DECREF(pickledobj);
+ return -1;
+ }
+ const char *copied = _copy_raw_string(pickled, len);
+ Py_DECREF(pickledobj);
+ if (copied == NULL) {
+ return -1;
+ }
+
+ *p_pickled = copied;
+ *p_len = len;
+ return 0;
+}
+
+static int
+_unpickle_object(const char *pickled, Py_ssize_t size, PyObject **p_obj)
+{
+ assert(!PyErr_Occurred());
+ PyObject *picklemod = PyImport_ImportModule("_pickle");
+ if (picklemod == NULL) {
+ PyErr_Clear();
+ picklemod = PyImport_ImportModule("pickle");
+ if (picklemod == NULL) {
+ return -1;
+ }
+ }
+ PyObject *loads = PyObject_GetAttrString(picklemod, "loads");
+ Py_DECREF(picklemod);
+ if (loads == NULL) {
+ return -1;
+ }
+ PyObject *pickledobj = PyBytes_FromStringAndSize(pickled, size);
+ if (pickledobj == NULL) {
+ Py_DECREF(loads);
+ return -1;
+ }
+ PyObject *obj = PyObject_CallOneArg(loads, pickledobj);
+ Py_DECREF(loads);
+ Py_DECREF(pickledobj);
+ if (obj == NULL) {
+ return -1;
+ }
+ *p_obj = obj;
+ return 0;
+}
+
+
static int
_release_xid_data(_PyCrossInterpreterData *data, int rawfree)
{
if (info->msg != NULL) {
PyMem_RawFree((void *)info->msg);
}
+ if (info->pickled != NULL) {
+ PyMem_RawFree((void *)info->pickled);
+ }
*info = (_PyXI_excinfo){{NULL}};
}
}
}
+static int
+_convert_exc_to_TracebackException(PyObject *exc, PyObject **p_tbexc)
+{
+ PyObject *args = NULL;
+ PyObject *kwargs = NULL;
+ PyObject *create = NULL;
+
+ // This is inspired by _PyErr_Display().
+ PyObject *tbmod = PyImport_ImportModule("traceback");
+ if (tbmod == NULL) {
+ return -1;
+ }
+ PyObject *tbexc_type = PyObject_GetAttrString(tbmod, "TracebackException");
+ Py_DECREF(tbmod);
+ if (tbexc_type == NULL) {
+ return -1;
+ }
+ create = PyObject_GetAttrString(tbexc_type, "from_exception");
+ Py_DECREF(tbexc_type);
+ if (create == NULL) {
+ return -1;
+ }
+
+ args = PyTuple_Pack(1, exc);
+ if (args == NULL) {
+ goto error;
+ }
+
+ kwargs = PyDict_New();
+ if (kwargs == NULL) {
+ goto error;
+ }
+ if (PyDict_SetItemString(kwargs, "save_exc_type", Py_False) < 0) {
+ goto error;
+ }
+ if (PyDict_SetItemString(kwargs, "lookup_lines", Py_False) < 0) {
+ goto error;
+ }
+
+ PyObject *tbexc = PyObject_Call(create, args, kwargs);
+ Py_DECREF(args);
+ Py_DECREF(kwargs);
+ Py_DECREF(create);
+ if (tbexc == NULL) {
+ goto error;
+ }
+
+ *p_tbexc = tbexc;
+ return 0;
+
+error:
+ Py_XDECREF(args);
+ Py_XDECREF(kwargs);
+ Py_XDECREF(create);
+ return -1;
+}
+
static const char *
_PyXI_excinfo_InitFromException(_PyXI_excinfo *info, PyObject *exc)
{
goto error;
}
+ // Pickle a traceback.TracebackException.
+ PyObject *tbexc = NULL;
+ if (_convert_exc_to_TracebackException(exc, &tbexc) < 0) {
+#ifdef Py_DEBUG
+ PyErr_FormatUnraisable("Exception ignored while creating TracebackException");
+#endif
+ PyErr_Clear();
+ }
+ else {
+ if (_pickle_object(tbexc, &info->pickled, &info->pickled_len) < 0) {
+#ifdef Py_DEBUG
+ PyErr_FormatUnraisable("Exception ignored while pickling TracebackException");
+#endif
+ PyErr_Clear();
+ }
+ Py_DECREF(tbexc);
+ }
+
return NULL;
error:
static void
_PyXI_excinfo_Apply(_PyXI_excinfo *info, PyObject *exctype)
{
+ PyObject *tbexc = NULL;
+ if (info->pickled != NULL) {
+ if (_unpickle_object(info->pickled, info->pickled_len, &tbexc) < 0) {
+ PyErr_Clear();
+ }
+ }
+
PyObject *formatted = _PyXI_excinfo_format(info);
PyErr_SetObject(exctype, formatted);
Py_DECREF(formatted);
+
+ if (tbexc != NULL) {
+ PyObject *exc = PyErr_GetRaisedException();
+ if (PyObject_SetAttrString(exc, "_tbexc", tbexc) < 0) {
+#ifdef Py_DEBUG
+ PyErr_FormatUnraisable("Exception ignored when setting _tbexc");
+#endif
+ PyErr_Clear();
+ }
+ Py_DECREF(tbexc);
+ PyErr_SetRaisedException(exc);
+ }
}
static PyObject *
goto error;
}
+ if (info->pickled != NULL) {
+ PyObject *tbexc = NULL;
+ if (_unpickle_object(info->pickled, info->pickled_len, &tbexc) < 0) {
+ PyErr_Clear();
+ }
+ else {
+ res = PyObject_SetAttrString(ns, "tbexc", tbexc);
+ Py_DECREF(tbexc);
+ if (res < 0) {
+ goto error;
+ }
+ }
+ }
+
return ns;
error:
}
else {
failure = _PyXI_InitError(err, excval, _PyXI_ERR_UNCAUGHT_EXCEPTION);
+ Py_DECREF(excval);
if (failure == NULL && override != NULL) {
err->code = errcode;
}
err = NULL;
}
- // a temporary hack (famous last words)
- if (excval != NULL) {
- // XXX Store the traceback info (or rendered traceback) on
- // _PyXI_excinfo, attach it to the exception when applied,
- // and teach PyErr_Display() to print it.
-#ifdef Py_DEBUG
- // XXX Drop this once _Py_excinfo picks up the slack.
- PyErr_Display(NULL, excval, NULL);
-#endif
- Py_DECREF(excval);
- }
-
// Finished!
assert(!PyErr_Occurred());
session->error = err;
projects / thirdparty / Python / cpython.git / commitdiff
