import subprocess
import sys
import unittest
+from functools import partial
from textwrap import dedent
from test import support
from test.support import (
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
# path may be used by Py_GetPath() 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
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
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)
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()}")
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)