gh-140438: properly run the asyncio REPL tests (#140298)

Author: bswck

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), "<stdin>")
-    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)