tests: pty golden harness for the TUI

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:]))