diff options
Diffstat (limited to 'test/fs.py')
| -rw-r--r-- | test/fs.py | 47 |
1 files changed, 47 insertions, 0 deletions
@@ -3,6 +3,7 @@ import contextlib import fcntl import json import os +import shutil import signal from pathlib import Path import subprocess @@ -466,6 +467,51 @@ def unattended_messages_expire(binary): assert not shown(), 'an unattended message outlived MessageLinger' +def python_repl_runs_one_line(binary): + """A one-line exec sent to Python's own REPL (3.13's pyrepl, which asks + for bracketed paste) runs at once, with no second Enter: the Enter + reaches it after the paste, not inside it. A real pty, a real python3.""" + python = shutil.which('python3') + if python is None: + return + version = subprocess.run([python, '-c', 'import sys; print(sys.version_info >= (3, 13))'], + capture_output=True, text=True).stdout.strip() + if version != 'True': + return + with tempfile.TemporaryDirectory(prefix='pardes-repl-') as directory: + with session(binary, Path(directory), 'repl') as (client, _): + def panes(client): + return {int(row.split()[0]) for row in client.read('/index').splitlines()} + before = panes(client) + client.write('/pane/1/ctl', b'Tty\n') + deadline = time.monotonic() + 10 + while time.monotonic() < deadline and panes(client) == before: + time.sleep(.05) + tty = max(panes(client) - before) + def body(): + return client.read(f'/pane/{tty}/body').decode('utf-8', 'replace') + def wait_for(text, seconds=15): + deadline = time.monotonic() + seconds + while time.monotonic() < deadline: + if text in body(): + return True + time.sleep(.1) + return False + assert wait_for('$') or wait_for('>'), body() + client.write(f'/pane/{tty}/pty/data', python.encode() + b' -q\r') + assert wait_for('>>>'), body() + client.write(f'/pane/{tty}/ctl', b'Repl python\n') + # An exec on the REPL's own body goes to the REPL, as its event + # record written back does: an empty range acts on its text. + fid = client.open(f'/pane/{tty}/event', 2) + try: + record = b'MX0 0 0 10 print(6*7)\n' + client.rpc(118, struct.pack('<IQI', fid, 0, len(record)) + record) + assert wait_for('\n42'), body() + finally: + client.close(fid) + + def signal_cleans_up(binary): """A detached session told to stop by a signal unlinks its sockets.""" for sig in [signal.SIGTERM, signal.SIGINT, signal.SIGHUP]: @@ -487,6 +533,7 @@ def test(binary, quic=False): run_file(binary) exit_answers(binary) unattended_messages_expire(binary) + python_repl_runs_one_line(binary) signal_cleans_up(binary) for options, message in [ (['--fs'], b'no such option'), |