diff --git a/Lib/test/test_repl.py b/Lib/test/test_repl.py index 54e69277282c30..042aa84b35dcf8 100644 --- a/Lib/test/test_repl.py +++ b/Lib/test/test_repl.py @@ -5,6 +5,7 @@ import subprocess import sys import unittest +from functools import partial from textwrap import dedent from test import support from test.support import ( @@ -27,7 +28,7 @@ raise unittest.SkipTest("test module requires subprocess") -def spawn_repl(*args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, **kw): +def spawn_repl(*args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, custom=False, **kw): """Run the Python REPL with the given arguments. kw is extra keyword args to pass to subprocess.Popen. Returns a Popen @@ -41,7 +42,11 @@ def spawn_repl(*args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, **kw): # path may be used by PyConfig_Get("module_search_paths") to build the # default module search path. stdin_fname = os.path.join(os.path.dirname(sys.executable), "") - cmd_line = [stdin_fname, '-I', '-i'] + cmd_line = [stdin_fname, '-I'] + # Don't re-run the built-in REPL from interactive mode + # if we're testing a custom REPL (such as the asyncio REPL). + if not custom: + cmd_line.append('-i') cmd_line.extend(args) # Set TERM=vt100, for the rationale see the comments in spawn_python() of @@ -55,6 +60,10 @@ def spawn_repl(*args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, **kw): stdout=stdout, stderr=stderr, **kw) + +spawn_asyncio_repl = partial(spawn_repl, "-m", "asyncio", custom=True) + + def run_on_interactive_mode(source): """Spawn a new Python interpreter, pass the given input source code from the stdin and return the @@ -359,7 +368,7 @@ def f(): class TestAsyncioREPL(unittest.TestCase): def test_multiple_statements_fail_early(self): user_input = "1 / 0; print(f'afterwards: {1+1}')" - p = spawn_repl("-m", "asyncio") + p = spawn_asyncio_repl() p.stdin.write(user_input) output = kill_python(p) self.assertIn("ZeroDivisionError", output) @@ -371,7 +380,7 @@ def test_toplevel_contextvars_sync(self): var = ContextVar("var", default="failed") var.set("ok") """) - p = spawn_repl("-m", "asyncio") + p = spawn_asyncio_repl() p.stdin.write(user_input) user_input2 = dedent(""" print(f"toplevel contextvar test: {var.get()}") @@ -387,7 +396,7 @@ def test_toplevel_contextvars_async(self): from contextvars import ContextVar var = ContextVar('var', default='failed') """) - p = spawn_repl("-m", "asyncio") + p = spawn_asyncio_repl() p.stdin.write(user_input) user_input2 = "async def set_var(): var.set('ok')\n" p.stdin.write(user_input2)