254 lines
8.3 KiB
Python
254 lines
8.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Differential pty test for the commands that need an attached client (F6).
|
|
|
|
A real client runs under a real pty against the C++ server and against tai.py's server, the same
|
|
key bytes (prefix, prompts, choosers, confirmations) are typed into both, and after every step the
|
|
server state read back through `--mplex <command>` must be identical (times, homes normalised).
|
|
|
|
python3 tests/mplex_commands_pty.py <cli binary> <repo root holding tai.py> [workdir]
|
|
"""
|
|
import fcntl
|
|
import os
|
|
import pty
|
|
import re
|
|
import select
|
|
import shlex
|
|
import shutil
|
|
import signal
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import termios
|
|
import time
|
|
|
|
PREFIX = b"\x02"
|
|
ENTER = b"\r"
|
|
ESC = b"\x1b"
|
|
UP = b"\x1b[A"
|
|
DOWN = b"\x1b[B"
|
|
|
|
STEPS = [
|
|
("cli", "new-session -d -s other -x 80 -y 24 'sleep 600'"),
|
|
("attach", None),
|
|
("keys", PREFIX + b"c"),
|
|
("keys", PREFIX + b","),
|
|
("keys", b"\x15renamed" + ENTER),
|
|
("keys", PREFIX + b"%"),
|
|
("keys", PREFIX + b'"'),
|
|
("keys", PREFIX + b"o"),
|
|
("keys", PREFIX + b"z"),
|
|
("keys", PREFIX + b"z"),
|
|
("keys", PREFIX + b"x"),
|
|
("keys", b"n"),
|
|
("keys", PREFIX + b"x"),
|
|
("keys", b"y"),
|
|
("keys", PREFIX + b":"),
|
|
("keys", b"split-window -h" + ENTER),
|
|
("keys", PREFIX + b":"),
|
|
("keys", b"nosuchcommand" + ENTER),
|
|
("keys", PREFIX + b":"),
|
|
("keys", b"set-option -g status-left 'X ' ; rename-window viaprompt" + ENTER),
|
|
("keys", PREFIX + b"$"),
|
|
("keys", b"\x15sess2" + ENTER),
|
|
("keys", PREFIX + b"&"),
|
|
("keys", b"y"),
|
|
("keys", PREFIX + b"'"),
|
|
("keys", b"0" + ENTER),
|
|
("keys", PREFIX + b"("),
|
|
("keys", PREFIX + b")"),
|
|
("keys", PREFIX + b"L"),
|
|
("keys", PREFIX + b"l"),
|
|
("keys", PREFIX + b"n"),
|
|
("keys", PREFIX + b"p"),
|
|
("keys", PREFIX + b"q"),
|
|
("keys", b"1"),
|
|
("keys", PREFIX + b"q"),
|
|
("keys", b"x"),
|
|
("keys", PREFIX + b"w"),
|
|
("keys", DOWN + DOWN + b"j"),
|
|
("keys", ENTER),
|
|
("keys", PREFIX + b"s"),
|
|
("keys", b"j" + ENTER),
|
|
("keys", PREFIX + b"w"),
|
|
("keys", b"q"),
|
|
("keys", PREFIX + b"="),
|
|
("keys", PREFIX + b"i"),
|
|
("keys", PREFIX + b"t"),
|
|
("keys", b"x"),
|
|
("cli", "set-buffer clip"),
|
|
("keys", PREFIX + b"="),
|
|
("keys", ENTER),
|
|
("keys", PREFIX + b"]"),
|
|
("keys", PREFIX + b"["),
|
|
("keys", b"k"),
|
|
("keys", ESC),
|
|
("keys", PREFIX + b"["),
|
|
("keys", PREFIX + b"f"),
|
|
("keys", b"renamed" + ENTER),
|
|
("keys", PREFIX + b"r"),
|
|
("keys", PREFIX + b"~"),
|
|
("keys", b"q"),
|
|
("keys", PREFIX + b"?"),
|
|
("keys", b"q"),
|
|
("keys", PREFIX + b"/"),
|
|
("keys", b"c"),
|
|
("keys", PREFIX + b"D"),
|
|
("keys", b"q"),
|
|
("keys", PREFIX + b"{"),
|
|
("keys", PREFIX + b"}"),
|
|
("keys", PREFIX + b"\x0f"),
|
|
("keys", PREFIX + b"\x1bo" if False else PREFIX + b"!"),
|
|
("keys", PREFIX + b" "),
|
|
("keys", PREFIX + b"E"),
|
|
("keys", PREFIX + b"\x1b1"),
|
|
("keys", PREFIX + b"\x1b5"),
|
|
("keys", PREFIX + b"m"),
|
|
("keys", PREFIX + b"M"),
|
|
("keys", PREFIX + b"\x1b[1;5A"),
|
|
("keys", PREFIX + b"\x1b[A"),
|
|
("keys", PREFIX + b"\x02"),
|
|
("cli", "display-panes"),
|
|
("cli", "command-prompt"),
|
|
("cli", "switch-client -n"),
|
|
("keys", PREFIX + b"d"),
|
|
("cli", "list-clients"),
|
|
]
|
|
|
|
SNAPSHOT = [
|
|
"list-sessions -F '#{session_name}:#{session_windows}:#{session_attached}'",
|
|
"list-windows -a -F '#{session_name}:#{window_index}:#{window_id}#{window_flags}:#{window_panes}:#{window_zoomed_flag}'",
|
|
"list-panes -a -F '#{session_name}:#{window_index}.#{pane_index} #{pane_width}x#{pane_height} #{pane_active} #{pane_mode} #{pane_marked}'",
|
|
"list-clients -F '#{client_key_table} #{session_name} #{client_prefix}'",
|
|
"list-buffers -F '#{buffer_name}:#{buffer_size}'",
|
|
"show-options -g status-left",
|
|
]
|
|
|
|
|
|
class Impl:
|
|
def __init__(self, name, cli, base):
|
|
self.name = name
|
|
self.cli = cli
|
|
self.home = os.path.join(base, name + "-home")
|
|
self.tmp = os.path.join(base, name + "-tmp")
|
|
self.pid = None
|
|
self.fd = None
|
|
|
|
def env(self):
|
|
env = {k: v for k, v in os.environ.items() if k not in ("TMUX", "TAI_PLEXER_ACTIVE", "TAI_PLEXER_SOCKET", "TAI_PLEXER_PANE")}
|
|
env.update(TAI_HOME=self.home, HOME=self.tmp, SHELL="/bin/sh", TERM="xterm", LANG="C.UTF-8")
|
|
return env
|
|
|
|
def run(self, line):
|
|
done = subprocess.run(self.cli + " " + line, shell=True, env=self.env(), stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=60)
|
|
return done.stdout.decode("utf-8", "replace"), done.returncode
|
|
|
|
def reset(self):
|
|
try:
|
|
self.run("kill-server")
|
|
except subprocess.TimeoutExpired:
|
|
pass
|
|
time.sleep(0.2)
|
|
shutil.rmtree(self.home, ignore_errors=True)
|
|
shutil.rmtree(self.tmp, ignore_errors=True)
|
|
os.makedirs(self.home)
|
|
os.makedirs(self.tmp)
|
|
|
|
def attach(self):
|
|
pid, fd = pty.fork()
|
|
if pid == 0:
|
|
fcntl.ioctl(0, termios.TIOCSWINSZ, struct.pack("HHHH", 24, 80, 0, 0))
|
|
os.chdir(self.tmp)
|
|
os.execvpe("/bin/sh", ["sh", "-c", self.cli + " new-session -s main 'sleep 600'"], self.env())
|
|
self.pid, self.fd = pid, fd
|
|
fcntl.ioctl(fd, termios.TIOCSWINSZ, struct.pack("HHHH", 24, 80, 0, 0))
|
|
|
|
def drain(self, seconds):
|
|
end = time.time() + seconds
|
|
while time.time() < end:
|
|
ready, _w, _x = select.select([self.fd], [], [], 0.05) if self.fd is not None else ([], [], [])
|
|
if ready:
|
|
try:
|
|
if not os.read(self.fd, 65536):
|
|
self.fd = None
|
|
except OSError:
|
|
self.fd = None
|
|
|
|
def type(self, data):
|
|
if self.fd is not None:
|
|
os.write(self.fd, data)
|
|
|
|
def close(self):
|
|
if self.pid:
|
|
try:
|
|
os.kill(self.pid, signal.SIGKILL)
|
|
os.waitpid(self.pid, 0)
|
|
except OSError:
|
|
pass
|
|
self.pid = None
|
|
if self.fd is not None:
|
|
try:
|
|
os.close(self.fd)
|
|
except OSError:
|
|
pass
|
|
self.fd = None
|
|
|
|
|
|
def normalise(text, impl):
|
|
text = text.replace(impl.home, "<HOME>").replace(impl.tmp, "<TMP>")
|
|
text = re.sub(r"\d\d:\d\d:\d\d", "HH:MM:SS", text)
|
|
text = re.sub(r"/dev/pts/\d+", "<TTY>", text)
|
|
return text
|
|
|
|
|
|
def main():
|
|
cli_bin = os.path.abspath(sys.argv[1])
|
|
root = os.path.abspath(sys.argv[2])
|
|
base = os.path.abspath(sys.argv[3] if len(sys.argv) > 3 else "/tmp/mplex_commands_pty_work")
|
|
os.makedirs(base, exist_ok=True)
|
|
boot = "import sys; sys.path.insert(0, %r); import tai; sys.exit(tai.main())" % root
|
|
cpp = Impl("cpp", shlex.quote(cli_bin) + " --mplex", base)
|
|
py = Impl("py", "python3 -I -X pycache_prefix=%s -c %s --mplex" % (shlex.quote(os.path.join(base, "pyc")), shlex.quote(boot)), base)
|
|
impls = [cpp, py]
|
|
for impl in impls:
|
|
impl.reset()
|
|
# the server and the client must start from the same minimal state: the client creates "main"
|
|
checks = 0
|
|
failures = 0
|
|
for index, (kind, arg) in enumerate(STEPS, 1):
|
|
outcomes = []
|
|
for impl in impls:
|
|
if kind == "attach":
|
|
impl.attach()
|
|
impl.drain(1.0)
|
|
elif kind == "keys":
|
|
impl.type(arg)
|
|
impl.drain(0.5)
|
|
elif kind == "cli":
|
|
text, rc = impl.run(arg)
|
|
outcomes.append(("cli", normalise(text, impl), rc))
|
|
impl.drain(0.2)
|
|
if kind in ("keys", "attach", "cli"):
|
|
snap = []
|
|
for line in SNAPSHOT:
|
|
text, rc = impl.run(line)
|
|
snap.append(normalise(text, impl) + "|rc=%d" % rc)
|
|
outcomes.append(("snap", "\n".join(snap), 0))
|
|
checks += 1
|
|
if outcomes[: len(outcomes) // 2] != outcomes[len(outcomes) // 2:]:
|
|
failures += 1
|
|
half = len(outcomes) // 2
|
|
print("MISMATCH step %d: %s %r" % (index, kind, arg))
|
|
for a, b in zip(outcomes[:half], outcomes[half:]):
|
|
if a != b:
|
|
print("--- cpp\n%s\n--- python\n%s" % (a, b))
|
|
for impl in impls:
|
|
impl.close()
|
|
impl.run("kill-server")
|
|
print("mplex_commands_pty: %d steps, %d mismatches" % (checks, failures))
|
|
return 1 if failures else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|