class InteractiveColoredConsole(code.InteractiveConsole):
+ STATEMENT_FAILED = object()
+
def __init__(
self,
locals: dict[str, object] | None = None,
limit=traceback.BUILTIN_EXCEPTION_LIMIT)
self.write(''.join(lines))
+ def runcode(self, code):
+ try:
+ exec(code, self.locals)
+ except SystemExit:
+ raise
+ except BaseException:
+ self.showtraceback()
+ return self.STATEMENT_FAILED
+ return None
+
def runsource(self, source, filename="<input>", symbol="single"):
try:
tree = self.compile.compiler(
if code is None:
return True
- self.runcode(code)
+ result = self.runcode(code)
+ if result is self.STATEMENT_FAILED:
+ break
return False
self.write("\nKeyboardInterrupt\n")
else:
self.showtraceback()
-
+ return self.STATEMENT_FAILED
class REPLThread(threading.Thread):
self.assertFalse(more)
self.assertEqual(f.getvalue(), "1\n")
+ @force_not_colorized
+ def test_multiple_statements_fail_early(self):
+ console = InteractiveColoredConsole()
+ code = dedent("""\
+ raise Exception('foobar')
+ print('spam&eggs')
+ """)
+ f = io.StringIO()
+ with contextlib.redirect_stderr(f):
+ console.runsource(code)
+ self.assertIn('Exception: foobar', f.getvalue())
+ self.assertNotIn('spam&eggs', f.getvalue())
+
def test_empty(self):
namespace = {}
code = ""
self.assertEqual(traceback_lines, expected_lines)
-class TestAsyncioREPLContextVars(unittest.TestCase):
+class TestAsyncioREPL(unittest.TestCase):
+ def test_multiple_statements_fail_early(self):
+ user_input = "1 / 0; print('afterwards')"
+ p = spawn_repl("-m", "asyncio")
+ p.stdin.write(user_input)
+ output = kill_python(p)
+ self.assertIn("ZeroDivisionError", output)
+ self.assertNotIn("afterwards", output)
+
def test_toplevel_contextvars_sync(self):
user_input = dedent("""\
from contextvars import ContextVar
--- /dev/null
+Execution of multiple statements in the new REPL now stops immediately upon
+the first exception encountered. Patch by Bartosz Sławecki.