hardening and bugfixing prior to stable release
This commit is contained in:
Benjamin Diedrichsen
Executable
+57
@@ -0,0 +1,57 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Drive an interactive CLI through a pty on a fixed schedule.
|
||||
|
||||
scripts/drive.py <script.json> -- <cmd> [args...]
|
||||
|
||||
`script.json` is a list of `[seconds_since_start, "text to send"]` pairs.
|
||||
Everything the child prints is echoed to this process's stdout.
|
||||
|
||||
Timing-based, so it is the blunt one — good for a quick manual poke at the TUI,
|
||||
bad for anything that has to be reliable. Prefer `expect.py`, which waits for the
|
||||
prompt instead of guessing when it will appear.
|
||||
|
||||
Environment: `PTY_ROWS` / `PTY_COLS` (default 50x200), `DRIVE_TIMEOUT` seconds.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import select
|
||||
import signal
|
||||
import sys
|
||||
import time
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
from ptysize import size_from_env, spawn # noqa: E402
|
||||
|
||||
sep = sys.argv.index("--")
|
||||
steps = json.loads(open(sys.argv[1]).read())
|
||||
cmd = sys.argv[sep + 1 :]
|
||||
|
||||
rows, cols = size_from_env()
|
||||
pid, fd = spawn(cmd, rows, cols)
|
||||
|
||||
start = time.time()
|
||||
pending = list(steps)
|
||||
deadline = start + float(os.environ.get("DRIVE_TIMEOUT", "3600"))
|
||||
|
||||
while True:
|
||||
if pending and time.time() - start >= pending[0][0]:
|
||||
_, text = pending.pop(0)
|
||||
os.write(fd, text.encode())
|
||||
r, _, _ = select.select([fd], [], [], 0.2)
|
||||
if r:
|
||||
try:
|
||||
data = os.read(fd, 65536)
|
||||
except OSError:
|
||||
break
|
||||
if not data:
|
||||
break
|
||||
sys.stdout.buffer.write(data)
|
||||
sys.stdout.buffer.flush()
|
||||
if time.time() > deadline:
|
||||
os.kill(pid, signal.SIGKILL)
|
||||
break
|
||||
|
||||
_, status = os.waitpid(pid, 0)
|
||||
sys.stderr.write("\n[drive.py] exit status: %d\n" % (status >> 8))
|
||||
sys.exit(status >> 8)
|
||||
Executable
+110
@@ -0,0 +1,110 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Expect-style pty driver — waits for each prompt before answering it.
|
||||
|
||||
scripts/expect.py <script.json> -- <cmd> [args...]
|
||||
|
||||
`script.json` is a list of steps:
|
||||
|
||||
{"expect": "<regex>", "send": "<text>", "settle": 0.4}
|
||||
{"send": "<text>"} -- send immediately
|
||||
|
||||
Each regex is matched against the ANSI-stripped output accumulated *since the
|
||||
previous step completed*, not against everything seen so far, so the same prompt
|
||||
text can be awaited twice in one run — which the nopy variable form does, once
|
||||
per cube.
|
||||
|
||||
Environment: `PTY_ROWS` / `PTY_COLS` (default 50x200), `EXPECT_TIMEOUT` seconds,
|
||||
`EXPECT_LOG` for the transcript path (default `expect.log`).
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import select
|
||||
import signal
|
||||
import sys
|
||||
import time
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
from ptysize import size_from_env, spawn # noqa: E402
|
||||
|
||||
ANSI = re.compile(rb"\x1b\[[0-9;?]*[a-zA-Z]|\x1b\][^\x07]*\x07|\r")
|
||||
|
||||
sep = sys.argv.index("--")
|
||||
steps = json.loads(open(sys.argv[1]).read())
|
||||
cmd = sys.argv[sep + 1 :]
|
||||
TIMEOUT = float(os.environ.get("EXPECT_TIMEOUT", "900"))
|
||||
|
||||
rows, cols = size_from_env()
|
||||
pid, fd = spawn(cmd, rows, cols)
|
||||
|
||||
log = open(os.environ.get("EXPECT_LOG", "expect.log"), "wb")
|
||||
start = time.time()
|
||||
window = b"" # output since the last completed step
|
||||
alive = True
|
||||
|
||||
|
||||
def pump(seconds):
|
||||
"""Read child output for `seconds`, appending to `window`."""
|
||||
global window, alive
|
||||
end = time.time() + seconds
|
||||
while time.time() < end:
|
||||
r, _, _ = select.select([fd], [], [], min(0.2, max(0.01, end - time.time())))
|
||||
if not r:
|
||||
continue
|
||||
try:
|
||||
data = os.read(fd, 65536)
|
||||
except OSError:
|
||||
alive = False
|
||||
return
|
||||
if not data:
|
||||
alive = False
|
||||
return
|
||||
window += data
|
||||
log.write(data)
|
||||
log.flush()
|
||||
|
||||
|
||||
failed = False
|
||||
|
||||
for i, step in enumerate(steps):
|
||||
pattern = step.get("expect")
|
||||
if pattern:
|
||||
rx = re.compile(pattern.encode())
|
||||
found = False
|
||||
while time.time() - start < TIMEOUT:
|
||||
if rx.search(ANSI.sub(b"", window)):
|
||||
found = True
|
||||
break
|
||||
if not alive:
|
||||
break
|
||||
pump(0.3)
|
||||
if not found:
|
||||
sys.stderr.write(
|
||||
"\n[expect] step %d timed out waiting for %r\n" % (i, pattern)
|
||||
)
|
||||
failed = True
|
||||
os.kill(pid, signal.SIGKILL)
|
||||
break
|
||||
sys.stderr.write("[expect] step %d matched %r\n" % (i, pattern))
|
||||
pump(step.get("settle", 0.5))
|
||||
window = b""
|
||||
text = step.get("send")
|
||||
if text:
|
||||
os.write(fd, text.encode())
|
||||
|
||||
# drain until the child exits
|
||||
while alive and time.time() - start < TIMEOUT:
|
||||
pump(1.0)
|
||||
if alive:
|
||||
sys.stderr.write("\n[expect] overall timeout, killing child\n")
|
||||
failed = True
|
||||
os.kill(pid, signal.SIGKILL)
|
||||
|
||||
_, status = os.waitpid(pid, 0)
|
||||
log.close()
|
||||
code = status >> 8
|
||||
sys.stderr.write("\n[expect] exit status: %d\n" % code)
|
||||
# A driver that gave up must not report the child's exit code as its own — a
|
||||
# SIGKILLed child can still look like a clean 0 to a caller reading $?.
|
||||
sys.exit(1 if failed else code)
|
||||
@@ -0,0 +1,50 @@
|
||||
"""Spawn a child on a pty whose window size is what you asked for.
|
||||
|
||||
`pty.fork()` leaves the new terminal at **0 rows by 0 columns**, and nothing
|
||||
fixes that afterwards: `COLUMNS`/`LINES` in the environment are a shell
|
||||
convention that `ioctl(TIOCGWINSZ)` has never heard of, so `process.stdout.rows`
|
||||
in the child stays 0 however they are set.
|
||||
|
||||
That is not a detail. A driver that forgets the ioctl is testing a terminal no
|
||||
user has, and it lied to us once already — the enquirer `Form` under it returned
|
||||
`{}` for reasons that had nothing to do with the code under test. See the
|
||||
`terminalSize` comment in `packages/nopy/src/nopy.prompts.ts`.
|
||||
|
||||
So the size is always set explicitly here, including when it is set to 0: the
|
||||
degenerate terminal is worth testing, but only on purpose.
|
||||
"""
|
||||
|
||||
import fcntl
|
||||
import os
|
||||
import pty
|
||||
import struct
|
||||
import termios
|
||||
|
||||
|
||||
def set_winsize(fd, rows, cols):
|
||||
fcntl.ioctl(fd, termios.TIOCSWINSZ, struct.pack("HHHH", rows, cols, 0, 0))
|
||||
|
||||
|
||||
def spawn(cmd, rows, cols, env=None):
|
||||
"""Fork `cmd` onto a pty sized `rows` x `cols`. Returns (pid, fd)."""
|
||||
pid, fd = pty.fork()
|
||||
if pid == 0:
|
||||
os.environ["TERM"] = "xterm-256color"
|
||||
# Kept in step with the ioctl so that a program reading either one gets
|
||||
# the same answer. The ioctl is what actually matters.
|
||||
os.environ["COLUMNS"] = str(cols)
|
||||
os.environ["LINES"] = str(rows)
|
||||
for key, value in (env or {}).items():
|
||||
os.environ[key] = value
|
||||
os.execvp(cmd[0], cmd)
|
||||
|
||||
set_winsize(fd, rows, cols)
|
||||
return pid, fd
|
||||
|
||||
|
||||
def size_from_env(default_rows=50, default_cols=200):
|
||||
"""`PTY_ROWS` / `PTY_COLS`, so a caller can ask for the 0x0 case."""
|
||||
return (
|
||||
int(os.environ.get("PTY_ROWS", default_rows)),
|
||||
int(os.environ.get("PTY_COLS", default_cols)),
|
||||
)
|
||||
Reference in New Issue
Block a user