Phase 5: Ausführung und Output implementieren und archivieren
This commit is contained in:
96
tests/support/ide-execution-pty.py
Normal file
96
tests/support/ide-execution-pty.py
Normal file
@@ -0,0 +1,96 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Headless Unix PTY smoke: real terminal handoff, child Ctrl+C and IDE resume.
|
||||
Run after cargo build -p tb-ide: python3 tests/support/ide-execution-pty.py
|
||||
"""
|
||||
import fcntl
|
||||
import os
|
||||
from pathlib import Path
|
||||
import select
|
||||
import signal
|
||||
import struct
|
||||
import subprocess
|
||||
import tempfile
|
||||
import termios
|
||||
import time
|
||||
|
||||
binary = Path(__file__).resolve().parents[2] / 'target/debug/tb'
|
||||
|
||||
|
||||
def exercise(abort=False, file_shell=False):
|
||||
with tempfile.TemporaryDirectory(prefix='tb-pty-') as directory:
|
||||
source = Path(directory) / 'main.bas'
|
||||
command = "printf '\\nWAITING_CHILD\\n'; exec sleep 30" if abort else "printf '\\nSHELL_PROOF\\n'; exit 7"
|
||||
source.write_text(f'PRINT SHELL("{command}")\nSTOP\nPRINT "CONTINUED"\nEND\n')
|
||||
master, slave = os.openpty()
|
||||
original = termios.tcgetattr(slave)
|
||||
fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack('HHHH', 30, 100, 0, 0))
|
||||
pid = os.fork()
|
||||
if pid == 0:
|
||||
os.setsid()
|
||||
fcntl.ioctl(slave, termios.TIOCSCTTY, 0)
|
||||
for fd in (0, 1, 2):
|
||||
os.dup2(slave, fd)
|
||||
os.close(master)
|
||||
os.close(slave)
|
||||
os.environ['TERM'] = 'xterm-256color'
|
||||
os.environ['XDG_CONFIG_HOME'] = directory
|
||||
# Keep the controlling terminal alive while checking restoration.
|
||||
signal.signal(signal.SIGINT, lambda *_: None)
|
||||
result = subprocess.run([str(binary), str(source)], check=False)
|
||||
restored = termios.tcgetattr(0) == original
|
||||
os._exit(result.returncode if restored else 99)
|
||||
transcript = bytearray()
|
||||
reaped = False
|
||||
|
||||
def until(marker, start=0):
|
||||
deadline = time.monotonic() + 10
|
||||
while marker not in transcript[start:]:
|
||||
assert time.monotonic() < deadline, (marker, bytes(transcript[-2000:]))
|
||||
if select.select([master], [], [], .1)[0]:
|
||||
transcript.extend(os.read(master, 65536))
|
||||
|
||||
try:
|
||||
until(b'?1049h')
|
||||
start = len(transcript)
|
||||
if file_shell:
|
||||
os.write(master, b'\x1bfh') # File -> Shell
|
||||
until(b'?1049l', start)
|
||||
os.write(master, b"printf '\\nFILE_SHELL_PROOF\\n'; exit\n")
|
||||
until(b'\r\nFILE_SHELL_PROOF\r\n', start)
|
||||
until(b'?1049h', start)
|
||||
else:
|
||||
os.write(master, b'\x1b[15~') # F5
|
||||
until(b'\r\nWAITING_CHILD\r\n' if abort else b'\r\nSHELL_PROOF\r\n', start)
|
||||
if abort:
|
||||
os.write(master, b'\x03') # cooked terminal SIGINT to child + IDE
|
||||
until(b'Paused', start)
|
||||
start = len(transcript)
|
||||
os.write(master, b'\x1b[15~')
|
||||
until(b'Ended', start)
|
||||
start = len(transcript)
|
||||
os.write(master, b'\x1b[1;3S') # Alt+F4
|
||||
until(b'Ungespeicherte', start) # standalone BAS has an unsaved project manifest
|
||||
os.write(master, b'\x1b[C\x1b[C\r') # discard temporary project
|
||||
deadline = time.monotonic() + 10
|
||||
while True:
|
||||
child, status = os.waitpid(pid, os.WNOHANG)
|
||||
if child:
|
||||
reaped = True
|
||||
assert os.waitstatus_to_exitcode(status) == 0
|
||||
break
|
||||
assert time.monotonic() < deadline, bytes(transcript[-1000:])
|
||||
if select.select([master], [], [], .05)[0]:
|
||||
transcript.extend(os.read(master, 65536))
|
||||
assert transcript.count(b'?1049h') >= 2
|
||||
assert transcript.count(b'?1049l') >= 2
|
||||
finally:
|
||||
if not reaped:
|
||||
os.killpg(pid, signal.SIGKILL)
|
||||
os.waitpid(pid, 0)
|
||||
os.close(master)
|
||||
os.close(slave)
|
||||
|
||||
|
||||
for kwargs in ({}, {'abort': True}, {'file_shell': True}):
|
||||
exercise(**kwargs)
|
||||
print('PASS', kwargs or {'shell_exit': 7})
|
||||
Reference in New Issue
Block a user