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

This commit is contained in:
Bartosz Sławecki 2025-10-23 17:23:23 +02:00 committed by GitHub
parent 574405c19e
commit 1a3da2c070
No known key found for this signature in database
GPG key ID: B5690EEEBB952194

View file

@ -5,6 +5,7 @@ import select
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 @@ if not has_subprocess_support:
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 @@ class TestInteractiveModeSyntaxErrors(unittest.TestCase):
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 @@ class TestAsyncioREPL(unittest.TestCase):
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 @@ class TestAsyncioREPL(unittest.TestCase):
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)