Add PTY terminal E2E harness
This commit is contained in:
slhx agent
Executable
+266
@@ -0,0 +1,266 @@
|
||||
#!/usr/bin/env python3
|
||||
"""PTY-backed terminal E2E harness for mim.
|
||||
|
||||
Runs a tiny high-signal scenario through a real pseudo-terminal and writes
|
||||
browser-viewable terminal artifacts. No third-party dependencies.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import html
|
||||
import os
|
||||
import pty
|
||||
import select
|
||||
import signal
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
ESC = 0x1B
|
||||
|
||||
|
||||
def visible_controls(data: bytes) -> str:
|
||||
out: list[str] = []
|
||||
text = data.decode("utf-8", errors="replace")
|
||||
for ch in text:
|
||||
code = ord(ch)
|
||||
if ch == "\r":
|
||||
out.append("␍")
|
||||
elif ch == "\n":
|
||||
out.append("␊\n")
|
||||
elif ch == "\x1b":
|
||||
out.append("␛")
|
||||
elif ch == "\t":
|
||||
out.append("⇥")
|
||||
elif code < 0x20:
|
||||
out.append("␀")
|
||||
else:
|
||||
out.append(ch)
|
||||
return "".join(out)
|
||||
|
||||
|
||||
class TerminalGrid:
|
||||
def __init__(self, width: int, height: int) -> None:
|
||||
self.width = width
|
||||
self.height = height
|
||||
self.rows = [[" " for _ in range(width)] for _ in range(height)]
|
||||
self.row = 0
|
||||
self.col = 0
|
||||
|
||||
def clear(self) -> None:
|
||||
self.rows = [[" " for _ in range(self.width)] for _ in range(self.height)]
|
||||
self.row = 0
|
||||
self.col = 0
|
||||
|
||||
def newline(self) -> None:
|
||||
self.row += 1
|
||||
if self.row >= self.height:
|
||||
self.rows.pop(0)
|
||||
self.rows.append([" " for _ in range(self.width)])
|
||||
self.row = self.height - 1
|
||||
|
||||
def put_char(self, ch: str) -> None:
|
||||
if ch == "\r":
|
||||
self.col = 0
|
||||
return
|
||||
if ch == "\n":
|
||||
self.newline()
|
||||
return
|
||||
if self.col >= self.width:
|
||||
self.col = 0
|
||||
self.newline()
|
||||
if 0 <= self.row < self.height and 0 <= self.col < self.width:
|
||||
self.rows[self.row][self.col] = ch
|
||||
self.col += 1
|
||||
|
||||
def feed(self, data: bytes) -> None:
|
||||
i = 0
|
||||
text = data.decode("utf-8", errors="replace")
|
||||
while i < len(text):
|
||||
ch = text[i]
|
||||
if ch == "\x1b" and i + 1 < len(text) and text[i + 1] == "[":
|
||||
j = i + 2
|
||||
while j < len(text) and not ("@" <= text[j] <= "~"):
|
||||
j += 1
|
||||
if j >= len(text):
|
||||
break
|
||||
command = text[j]
|
||||
params = text[i + 2 : j]
|
||||
if command == "H":
|
||||
self.row = 0
|
||||
self.col = 0
|
||||
elif command == "J" and params.endswith("2"):
|
||||
self.clear()
|
||||
# SGR and unknown CSI are intentionally ignored for semantic grid.
|
||||
i = j + 1
|
||||
continue
|
||||
self.put_char(ch)
|
||||
i += 1
|
||||
|
||||
def lines(self) -> list[str]:
|
||||
return ["".join(row).rstrip() for row in self.rows]
|
||||
|
||||
|
||||
def read_available(fd: int, timeout: float = 0.12) -> bytes:
|
||||
chunks: list[bytes] = []
|
||||
deadline = time.monotonic() + timeout
|
||||
while True:
|
||||
remaining = max(0.0, deadline - time.monotonic())
|
||||
ready, _, _ = select.select([fd], [], [], remaining)
|
||||
if not ready:
|
||||
break
|
||||
try:
|
||||
chunk = os.read(fd, 65536)
|
||||
except OSError:
|
||||
break
|
||||
if not chunk:
|
||||
break
|
||||
chunks.append(chunk)
|
||||
deadline = time.monotonic() + 0.04
|
||||
return b"".join(chunks)
|
||||
|
||||
|
||||
def spawn_under_pty(argv: list[str], width: int, height: int) -> tuple[int, int]:
|
||||
pid, fd = pty.fork()
|
||||
if pid == 0:
|
||||
os.environ["TERM"] = "xterm-256color"
|
||||
try:
|
||||
import fcntl
|
||||
import struct
|
||||
import termios
|
||||
|
||||
winsize = struct.pack("HHHH", height, width, 0, 0)
|
||||
fcntl.ioctl(0, termios.TIOCSWINSZ, winsize)
|
||||
except Exception:
|
||||
pass
|
||||
os.execv(argv[0], argv)
|
||||
return pid, fd
|
||||
|
||||
|
||||
def svg_for_lines(lines: list[str], width: int, height: int) -> str:
|
||||
cell_w = 9
|
||||
cell_h = 18
|
||||
pad = 12
|
||||
svg_w = pad * 2 + width * cell_w
|
||||
svg_h = pad * 2 + height * cell_h
|
||||
body = []
|
||||
for idx, line in enumerate(lines[:height]):
|
||||
body.append(
|
||||
f'<text x="{pad}" y="{pad + (idx + 1) * cell_h}" xml:space="preserve">{html.escape(line)}</text>'
|
||||
)
|
||||
return "\n".join(
|
||||
[
|
||||
f'<svg xmlns="http://www.w3.org/2000/svg" width="{svg_w}" height="{svg_h}" viewBox="0 0 {svg_w} {svg_h}">',
|
||||
'<rect width="100%" height="100%" fill="#12161e"/>',
|
||||
'<style>text{font-family:ui-monospace,SFMono-Regular,Menlo,Consolas,monospace;font-size:14px;fill:#d6deeb}</style>',
|
||||
*body,
|
||||
"</svg>",
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
def html_for_lines(lines: list[str], raw_svg_name: str) -> str:
|
||||
escaped = html.escape("\n".join(lines))
|
||||
return f"""<!doctype html>
|
||||
<meta charset="utf-8">
|
||||
<title>mim terminal-e2e</title>
|
||||
<style>
|
||||
body {{ margin: 0; background: #0f131a; color: #d6deeb; }}
|
||||
.terminal {{ white-space: pre; font: 14px/18px ui-monospace, SFMono-Regular, Menlo, Consolas, monospace; padding: 12px; }}
|
||||
a {{ color: #7ee787; }}
|
||||
</style>
|
||||
<div class="terminal">{escaped}</div>
|
||||
<p><a href="{html.escape(raw_svg_name)}">SVG screenshot artifact</a></p>
|
||||
"""
|
||||
|
||||
|
||||
def run_scenario(mim: Path, out_dir: Path, width: int, height: int) -> None:
|
||||
if out_dir.exists():
|
||||
shutil.rmtree(out_dir)
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
with tempfile.TemporaryDirectory(prefix="mim-terminal-e2e-") as tmp:
|
||||
target = Path(tmp) / "note.txt"
|
||||
target.write_text("", encoding="utf-8")
|
||||
pid, fd = spawn_under_pty([str(mim), str(target)], width, height)
|
||||
transcript = bytearray()
|
||||
grid = TerminalGrid(width, height)
|
||||
try:
|
||||
for payload in [b"", b"w", b"h", b"a", b"t", b" ", b"i", b"s", b" ", b"u", b"p", b" ", b"n", b" ", b"w", b" ", b"q"]:
|
||||
if payload:
|
||||
os.write(fd, payload)
|
||||
chunk = read_available(fd)
|
||||
transcript.extend(chunk)
|
||||
grid.feed(chunk)
|
||||
deadline = time.monotonic() + 2.0
|
||||
while time.monotonic() < deadline:
|
||||
done_pid, status = os.waitpid(pid, os.WNOHANG)
|
||||
chunk = read_available(fd, 0.03)
|
||||
transcript.extend(chunk)
|
||||
grid.feed(chunk)
|
||||
if done_pid == pid:
|
||||
if status != 0:
|
||||
raise SystemExit(f"mim exited non-zero: status={status}")
|
||||
break
|
||||
else:
|
||||
(out_dir).mkdir(parents=True, exist_ok=True)
|
||||
(out_dir / "raw-timeout.bin").write_bytes(bytes(transcript))
|
||||
(out_dir / "visible-timeout.txt").write_text(visible_controls(bytes(transcript)), encoding="utf-8")
|
||||
(out_dir / "transcript-timeout.txt").write_text("\n".join(grid.lines()) + "\n", encoding="utf-8")
|
||||
os.kill(pid, signal.SIGTERM)
|
||||
raise SystemExit(f"mim did not exit after save+quit script; artifacts={out_dir}")
|
||||
finally:
|
||||
try:
|
||||
os.close(fd)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
saved = target.read_text(encoding="utf-8")
|
||||
lines = grid.lines()
|
||||
(out_dir / "raw.bin").write_bytes(bytes(transcript))
|
||||
(out_dir / "visible-controls.txt").write_text(visible_controls(bytes(transcript)), encoding="utf-8")
|
||||
(out_dir / "transcript.txt").write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||||
(out_dir / "screenshot.svg").write_text(svg_for_lines(lines, width, height), encoding="utf-8")
|
||||
(out_dir / "terminal.html").write_text(html_for_lines(lines, "screenshot.svg"), encoding="utf-8")
|
||||
|
||||
failures: list[str] = []
|
||||
if saved != "what is up":
|
||||
failures.append(f"saved file mismatch: {saved!r}")
|
||||
joined = "\n".join(lines)
|
||||
raw_text = bytes(transcript).decode("utf-8", errors="ignore")
|
||||
if "1│" not in raw_text:
|
||||
failures.append("line-number gutter missing from terminal output")
|
||||
if "▌" not in raw_text:
|
||||
failures.append("cursor marker missing from terminal output")
|
||||
if "mode:normal" not in raw_text and "saved" not in raw_text:
|
||||
failures.append("expected editor status not observed")
|
||||
if "^[" in raw_text:
|
||||
failures.append("escape appears echoed as ^[")
|
||||
if "mode:normal" not in raw_text:
|
||||
failures.append("Space-n did not leave insert mode before save/quit")
|
||||
if any(len(line) > width for line in lines):
|
||||
failures.append("terminal line exceeds viewport width")
|
||||
if failures:
|
||||
print(f"artifacts\t{out_dir}", file=sys.stderr)
|
||||
for failure in failures:
|
||||
print(f"FAIL\t{failure}", file=sys.stderr)
|
||||
raise SystemExit(1)
|
||||
|
||||
print("scenario\tstatus\tartifacts")
|
||||
print(f"raw-key-chrome-crlf\tPASS\t{out_dir}")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description="Run mim through a PTY and write browser-viewable terminal E2E artifacts.")
|
||||
parser.add_argument("mim", type=Path, help="path to compiled mim binary")
|
||||
parser.add_argument("--out", type=Path, default=Path(".zig-cache/terminal-e2e/raw-key-chrome-crlf"))
|
||||
parser.add_argument("--width", type=int, default=48)
|
||||
parser.add_argument("--height", type=int, default=16)
|
||||
args = parser.parse_args()
|
||||
run_scenario(args.mim.resolve(), args.out, args.width, args.height)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user