diff options
| -rwxr-xr-x | scripts/tui-golden.py | 680 |
1 files changed, 680 insertions, 0 deletions
diff --git a/scripts/tui-golden.py b/scripts/tui-golden.py new file mode 100755 index 0000000..2a60d7f --- /dev/null +++ b/scripts/tui-golden.py @@ -0,0 +1,680 @@ +#!/usr/bin/env python3 +# Automated pty golden tests for bokftui. +# +# python3 scripts/tui-golden.py [--keep] [--verbose] [--only NAME] +# +# Starts a throwaway bokfd in a mktemp -d under /tmp, creates a test org, then +# drives bokftui through scripts/tui-sandbox.sh (isolated XDG config/cache) on +# a pty and asserts Swedish screen text. Prints PASS/FAIL per scenario and a +# screen diff on failure; exits non-zero if anything failed. +# +# No third-party modules are needed. The built-in renderer below replays the +# raw pty bytes through a small ANSI screen emulator that handles the +# sequences a full-screen curses app emits: CSI H/f, A/B/C/D/E/F/G/d, J, K, +# X, P, @, L/M, S/T, r (scrolling region), ignored SGR/private modes, DEC +# special graphics (ESC(0, SO/SI), CR/LF/BS/TAB and ESC 7/8/D/M/E. +# Limitations: no wide-character cell widths (each decoded codepoint takes one +# cell), no scrollback, writes at the last column clamp instead of autowrap, +# and colour/attributes are ignored. That is enough for stable text asserts, +# not for pixel-exact comparison. + +import difflib +import fcntl +import json +import os +import pty +import select +import shutil +import signal +import stat +import struct +import subprocess +import sys +import tempfile +import termios +import time +from pathlib import Path + +# -------------------------------------------------------------------------- +# Expected strings (data, not code) -- extend scenarios here. + +ORG_NAME = "Test AB" +ORG_NR = "5560123456" + +KEYS = { + "enter": "\r", + "esc": "\x1b", + "up": "\x1bOA", # application-mode arrows: curses enables keypad + "down": "\x1bOB", + "right": "\x1bOC", + "left": "\x1bOD", + "home": "\x1bOH", + "end": "\x1bOF", + "pgup": "\x1b[5~", + "pgdn": "\x1b[6~", + "ctrlc": "\x03", +} + +# {org_name} {org_nr} {fy_label} {fy_start} {fy_end} are substituted at run +# time. "keys" names come from KEYS (raw strings/literals work too). +SCENARIOS = [ + { + "name": "dashboard", + "screen": "dashboard", + "expect": ["{org_name}", "{fy_label}", "{fy_start}", "{fy_end}"], + }, + { + "name": "audit-ok", + "screen": "audit", + "expect": ["Revision", "Verifiera hashkedjan"], + "steps": [ + { + "keys": ["enter"], + "expect": ["Kedjorna är intakta.", "verifikat"], + }, + ], + }, + { + "name": "vouchers", + "screen": "vouchers", + "expect": ["Verifikat"], + }, +] + +# -------------------------------------------------------------------------- +# Tunables. + +TERM_COLS, TERM_ROWS = 120, 36 +QUIET = 0.25 # pty considered stable after this many quiet seconds +STABLE_TIMEOUT = 8.0 # max time to wait for output to settle +READY_TIMEOUT = 15.0 # max time to wait for the initial screen +KEY_ATTEMPTS = 2 # bounded retry per key step (startup races) +EXIT_TIMEOUT = 6.0 # max time to wait for the TUI to quit on ^C +DAEMON_TIMEOUT = 20.0 + +_ACS = { + "`": "◆", "a": "▒", "f": "°", "g": "±", "j": "┘", "k": "┐", "l": "┌", + "m": "└", "n": "┼", "o": "⎺", "p": "⎻", "q": "─", "r": "⎼", "s": "⎽", + "t": "├", "u": "┤", "v": "┴", "w": "┬", "x": "│", "y": "≤", "z": "≥", + "{": "π", "|": "≠", "}": "£", "~": "·", +} + + +def _params(arg): + if not arg: + return [] + out = [] + for part in arg.split(";"): + try: + out.append(int(part) if part else None) + except ValueError: + out.append(None) + return out + + +class Terminal: + def __init__(self, cols, rows): + self.cols, self.rows = cols, rows + self.reset() + + def reset(self): + self.cells = [[" "] * self.cols for _ in range(self.rows)] + self.cx = self.cy = 0 + self.top, self.bottom = 0, self.rows - 1 + self.saved = (0, 0) + self.acs = [False, False] + self.shifted = 0 + self.state = "text" + self.buf = "" + + def feed(self, text): + for ch in text: + self._feed(ch) + + def _feed(self, ch): + st = self.state + if st == "text": + if ch == "\x1b": + self.state = "esc" + elif ch == "\r": + self.cx = 0 + elif ch == "\n": + self._lf() + elif ch == "\b": + self.cx = max(0, self.cx - 1) + elif ch == "\t": + self.cx = min(self.cols - 1, (self.cx // 8 + 1) * 8) + elif ch == "\x0e": + self.shifted = 1 + elif ch == "\x0f": + self.shifted = 0 + elif ord(ch) >= 32 and ord(ch) != 127: + self._put(ch) + return + if st == "esc": + if ch == "[": + self.state, self.buf = "csi", "" + elif ch == "]": + self.state, self.buf = "osc", "" + elif ch == "(": + self.state = "g0" + elif ch == ")": + self.state = "g1" + elif ch == "7": + self.saved = (self.cy, self.cx) + self.state = "text" + elif ch == "8": + self.cy, self.cx = self.saved + self.state = "text" + elif ch == "D": + self._lf() + self.state = "text" + elif ch == "M": + self._ri() + self.state = "text" + elif ch == "E": + self.cx = 0 + self._lf() + self.state = "text" + else: + self.state = "text" + return + if st in ("g0", "g1"): + self.acs[0 if st == "g0" else 1] = ch == "0" + self.state = "text" + return + if st == "csi": + if 0x40 <= ord(ch) <= 0x7e: + self._csi(self.buf, ch) + self.state = "text" + else: + self.buf += ch + return + if st == "osc": + if ch == "\x07": + self.state = "text" + elif ch == "\x1b": + self.state = "osc_esc" + return + if st == "osc_esc": + self.state = "text" if ch == "\\" else "osc" + + def _csi(self, buf, final): + private = "" + i = 0 + while i < len(buf) and buf[i] in "?><=!": + private += buf[i] + i += 1 + p = _params(buf[i:]) + + def num(idx, default=1): + return p[idx] if idx < len(p) and p[idx] is not None else default + + if private: + return # modes: cursor visibility, alt screen, keypad ... ignored + if final in "Hf": + self._move(num(0) - 1, num(1) - 1) + elif final == "A": + self._move(self.cy - num(0), self.cx) + elif final == "B": + self._move(self.cy + num(0), self.cx) + elif final == "C": + self._move(self.cy, self.cx + num(0)) + elif final == "D": + self._move(self.cy, self.cx - num(0)) + elif final == "E": + self._move(self.cy + num(0), 0) + elif final == "F": + self._move(self.cy - num(0), 0) + elif final in "G`": + self._move(self.cy, num(0) - 1) + elif final == "d": + self._move(num(0) - 1, self.cx) + elif final == "J": + self._erase_display(num(0, 0)) + elif final == "K": + self._erase_line(num(0, 0)) + elif final == "X": + self._erase_chars(num(0)) + elif final == "P": + self._delete_chars(num(0)) + elif final == "@": + self._insert_chars(num(0)) + elif final == "r": + top = num(0) - 1 + bottom = num(1) - 1 if len(p) > 1 and p[1] is not None else self.rows - 1 + if 0 <= top < bottom < self.rows: + self.top, self.bottom = top, bottom + self._move(0, 0) + elif final == "L": + self._insert_lines(num(0)) + elif final == "M": + self._delete_lines(num(0)) + elif final == "S": + self._scroll_up(num(0)) + elif final == "T": + self._scroll_down(num(0)) + elif final == "s": + self.saved = (self.cy, self.cx) + elif final == "u": + self.cy, self.cx = self.saved + # "m" (SGR) and anything unknown: ignored + + def _move(self, cy, cx): + self.cy = max(0, min(self.rows - 1, cy)) + self.cx = max(0, min(self.cols - 1, cx)) + + def _put(self, ch): + if self.acs[self.shifted] and ch in _ACS: + ch = _ACS[ch] + if 0 <= self.cy < self.rows: + self.cells[self.cy][self.cx] = ch + if self.cx < self.cols - 1: + self.cx += 1 + + def _lf(self): + if self.cy == self.bottom: + self._scroll_up(1) + else: + self.cy = min(self.rows - 1, self.cy + 1) + + def _ri(self): + if self.cy == self.top: + self._scroll_down(1) + else: + self.cy = max(0, self.cy - 1) + + def _scroll_up(self, n): + for _ in range(max(1, n)): + for y in range(self.top, self.bottom): + self.cells[y] = self.cells[y + 1][:] + self.cells[self.bottom] = [" "] * self.cols + + def _scroll_down(self, n): + for _ in range(max(1, n)): + for y in range(self.bottom, self.top, -1): + self.cells[y] = self.cells[y - 1][:] + self.cells[self.top] = [" "] * self.cols + + def _erase_line(self, mode): + if mode == 0: + for x in range(self.cx, self.cols): + self.cells[self.cy][x] = " " + elif mode == 1: + for x in range(0, min(self.cx + 1, self.cols)): + self.cells[self.cy][x] = " " + else: + self.cells[self.cy] = [" "] * self.cols + + def _erase_display(self, mode): + if mode == 0: + self._erase_line(0) + for y in range(self.cy + 1, self.rows): + self.cells[y] = [" "] * self.cols + elif mode == 1: + self._erase_line(1) + for y in range(0, self.cy): + self.cells[y] = [" "] * self.cols + else: + self.cells = [[" "] * self.cols for _ in range(self.rows)] + + def _erase_chars(self, n): + for x in range(self.cx, min(self.cols, self.cx + max(1, n))): + self.cells[self.cy][x] = " " + + def _delete_chars(self, n): + row = self.cells[self.cy] + del row[self.cx:self.cx + max(1, n)] + self.cells[self.cy] = row + [" "] * (self.cols - len(row)) + + def _insert_chars(self, n): + row = self.cells[self.cy] + self.cells[self.cy] = row[:self.cx] + [" "] * max(1, n) + row[self.cx:] + self.cells[self.cy] = self.cells[self.cy][:self.cols] + + def _insert_lines(self, n): + if not (self.top <= self.cy <= self.bottom): + return + for _ in range(max(1, n)): + self.cells.insert(self.cy, [" "] * self.cols) + del self.cells[self.bottom + 1] + + def _delete_lines(self, n): + if not (self.top <= self.cy <= self.bottom): + return + for _ in range(max(1, n)): + del self.cells[self.cy] + self.cells.insert(self.bottom, [" "] * self.cols) + + def text(self): + return "\n".join("".join(row).rstrip() for row in self.cells) + + +class HarnessError(Exception): + pass + + +class PtyApp: + def __init__(self, argv, env, cols, rows): + self.argv, self.env = argv, env + self.cols, self.rows = cols, rows + self.raw = bytearray() + self.pid = self.fd = None + self.exited = False + self.status = None + + def start(self): + pid, fd = pty.fork() + if pid == 0: + try: + os.execvpe(self.argv[0], self.argv, self.env) + except OSError: + os._exit(127) + self.pid, self.fd = pid, fd + fcntl.ioctl(fd, termios.TIOCSWINSZ, + struct.pack("HHHH", self.rows, self.cols, 0, 0)) + return self + + def pump(self, quiet=QUIET, timeout=STABLE_TIMEOUT): + last = time.time() + while time.time() - last < timeout: + ready, _, _ = select.select([self.fd], [], [], 0.05) + if ready: + try: + data = os.read(self.fd, 65536) + except OSError: + data = b"" + if not data: + self.exited = True + return + self.raw += data + last = time.time() + elif time.time() - last >= quiet: + return + + def render(self): + term = Terminal(self.cols, self.rows) + term.feed(bytes(self.raw).decode("utf-8", "replace")) + return term.text() + + def send(self, data): + try: + os.write(self.fd, data) + except OSError as e: + raise HarnessError(f"could not write to pty: {e}") + + def _reap(self): + try: + pid, status = os.waitpid(self.pid, os.WNOHANG) + except ChildProcessError: + self.exited = True + return True + if pid == self.pid: + self.exited, self.status = True, status + return True + return False + + def close(self): + if self.fd is None: + return + try: + os.write(self.fd, KEYS["ctrlc"].encode()) + except OSError: + pass + deadline = time.time() + EXIT_TIMEOUT + while time.time() < deadline and not self._reap(): + ready, _, _ = select.select([self.fd], [], [], 0.05) + if ready: + try: + os.read(self.fd, 65536) + except OSError: + pass + if not self.exited: + try: + os.killpg(self.pid, signal.SIGKILL) + except (ProcessLookupError, PermissionError): + try: + os.kill(self.pid, signal.SIGKILL) + except ProcessLookupError: + pass + try: + os.waitpid(self.pid, 0) + except ChildProcessError: + pass + try: + os.close(self.fd) + except OSError: + pass + self.fd = None + + +def run_checked(cmd, env): + p = subprocess.run(cmd, env=env, stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + text=True) + if p.returncode != 0: + raise HarnessError(f"{' '.join(cmd[:2])} failed " + f"({p.returncode}):\n{p.stdout.strip()}") + return p.stdout + + +def ensure_build(repo): + need = [n for n in ("bokfd", "bokftui", "bokfctl") + if not os.access(repo / "build" / n, os.X_OK)] + if not need: + return + print(f"building missing binaries: {', '.join(need)}", flush=True) + p = subprocess.run(["make", f"-j{os.cpu_count() or 1}"], cwd=str(repo)) + if p.returncode != 0: + raise HarnessError("make failed") + for n in need: + if not os.access(repo / "build" / n, os.X_OK): + raise HarnessError(f"make did not produce build/{n}") + + +def wait_for(app, needles, timeout): + deadline = time.time() + timeout + screen = "" + while True: + app.pump() + screen = app.render() + missing = [n for n in needles if n not in screen] + if not missing or app.exited or time.time() >= deadline: + return missing, screen + time.sleep(0.05) + + +def format_expect(strings, ctx): + return [s.format(**ctx) for s in strings] + + +def resolve_keys(spec): + if isinstance(spec, str): + spec = [spec] + return "".join(KEYS.get(k, k) for k in spec).encode("utf-8") + + +def report_failure(name, missing, before, current): + print(f"FAIL {name}") + for m in missing: + print(f" missing: {m!r}") + diff = list(difflib.unified_diff( + before.splitlines(), current.splitlines(), + fromfile="screen before", tofile="screen after", lineterm="", n=2)) + if diff: + print(" --- screen diff ---") + for line in diff: + print(" " + line) + else: + print(" (screen unchanged)") + print(" --- current screen ---") + for i, line in enumerate(current.splitlines(), 1): + print(f" {i:>3}|{line}") + print(flush=True) + + +def setup_rig(repo, tmp, env): + bind, db = repo / "build", tmp / "t.db" + sock = tmp / "sock" + run_checked([str(bind / "bokfd"), "init", "--db", str(db), + "--user", "admin"], env) + log = tmp / "daemon.log" + daemon = subprocess.Popen( + [str(bind / "bokfd"), "--db", str(db), "--socket", str(sock), + "--log-level", "info"], + env=env, stdin=subprocess.DEVNULL, stdout=open(log, "wb"), + stderr=subprocess.STDOUT, start_new_session=True) + deadline = time.time() + DAEMON_TIMEOUT + while time.time() < deadline: + if daemon.poll() is not None: + raise HarnessError("daemon exited during startup:\n" + + log.read_text(errors="replace").strip()) + try: + if stat.S_ISSOCK(os.stat(sock).st_mode): + break + except OSError: + pass + time.sleep(0.05) + else: + raise HarnessError(f"daemon socket {sock} did not appear") + + out = run_checked([str(bind / "bokfctl"), "--socket", str(sock), + "--user", "admin", "--password", env["BOKFD_PASSWORD"], + "org.create", + json.dumps({"name": ORG_NAME, "org_nr": ORG_NR})], env) + org = json.loads(out)["result"] + org_id, fy_id = org["id"], org["fiscal_year_id"] + out = run_checked([str(bind / "bokfctl"), "--socket", str(sock), + "--user", "admin", "--password", env["BOKFD_PASSWORD"], + "--org", str(org_id), "fiscal_year.get", + json.dumps({"id": fy_id})], env) + fy = json.loads(out)["result"] + return daemon, sock, org_id, fy + + +def main(argv): + opts = set(argv) + if "--help" in opts or "-h" in opts: + print("usage: tui-golden.py [--keep] [--verbose] [--only NAME]") + return 0 + verbose = "--verbose" in opts + keep = "--keep" in opts + only = None + rest = list(argv) + while rest: + a = rest.pop(0) + if a == "--only" and rest: + only = rest.pop(0) + elif a not in ("--keep", "--verbose", "--only"): + print(f"unknown option: {a}", file=sys.stderr) + return 2 + + repo = Path(__file__).resolve().parent.parent + tmp = Path(tempfile.mkdtemp(prefix="bokf-golden-", dir="/tmp")) + password = "golden-" + os.urandom(6).hex() + env = dict(os.environ) + env.update({"BOKFD_PASSWORD": password}) + env.pop("BOKFD_SESSION", None) + env.pop("BOKFD_TOKEN", None) + + daemon = None + failures = 0 + keep_note = keep + try: + ensure_build(repo) + daemon, sock, org_id, fy = setup_rig(repo, tmp, env) + ctx = { + "org_name": ORG_NAME, + "org_nr": ORG_NR, + "fy_label": fy.get("label", ""), + "fy_start": fy.get("start_date", ""), + "fy_end": fy.get("end_date", ""), + } + print(f"bokf TUI golden tests: org {org_id} \"{ORG_NAME}\", " + f"fy {fy.get('label')} (id {fy.get('id')}), socket {sock}") + env.update({ + "TERM": "xterm-256color", + "LC_ALL": "C.UTF-8", + "BOKFD_USER": "admin", + }) + sandbox = repo / "scripts" / "tui-sandbox.sh" + bokftui = repo / "build" / "bokftui" + selected = [sc for sc in SCENARIOS if not only or sc["name"] == only] + if only and not selected: + raise HarnessError(f"unknown scenario: {only}") + for sc in selected: + argv_tui = [str(sandbox), "--", str(bokftui), + "--socket", str(sock), "--user", "admin", + "--org", str(org_id), "--fy", str(fy["id"]), + "--screen", sc["screen"]] + app = PtyApp(argv_tui, env, TERM_COLS, TERM_ROWS) + name = sc["name"] + missing = None + before = current = "" + try: + app.start() + missing, before = wait_for( + app, format_expect(sc.get("expect", []), ctx), READY_TIMEOUT) + if missing: + report_failure(name, missing, "", before) + failures += 1 + continue + current = before + for step in sc.get("steps", []): + keys = resolve_keys(step["keys"]) + want = format_expect(step.get("expect", []), ctx) + for attempt in range(1, KEY_ATTEMPTS + 1): + before = app.render() + app.send(keys) + app.pump() + current = app.render() + missing = [n for n in want if n not in current] + if not missing: + break + # Retry only when the key left the screen untouched + # (startup race); a visible change means it arrived. + if current == before and attempt < KEY_ATTEMPTS: + print(f" {name}: retrying key step " + f"{step['keys']!r} (screen unchanged)") + continue + break + if missing: + report_failure(name, missing, before, current) + failures += 1 + break + if not missing: + print(f"PASS {name}") + if verbose: + for line in current.splitlines(): + print(f" |{line}") + finally: + app.close() + except HarnessError as e: + print(f"HARNESS ERROR: {e}", file=sys.stderr) + if daemon and daemon.poll() is not None: + print((tmp / "daemon.log").read_text(errors="replace"), + file=sys.stderr) + failures = max(failures, 1) + finally: + if daemon and daemon.poll() is None: + try: + os.killpg(daemon.pid, signal.SIGTERM) + except ProcessLookupError: + pass + try: + daemon.wait(timeout=5) + except subprocess.TimeoutExpired: + try: + os.killpg(daemon.pid, signal.SIGKILL) + except ProcessLookupError: + pass + daemon.wait() + if keep_note: + print(f"kept temp dir: {tmp}") + else: + shutil.rmtree(tmp, ignore_errors=True) + return 1 if failures else 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) |