path: root/scripts/tui-golden.py

#!/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 base64
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",
    "ctrln": "\x0e",
    "ctrlenter": "\x1b[27;5;13~",
    "f5": "\x1b[15~",
}

# {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}",
                   "Företag", "Bolaget"],
    },
    {
        "name": "audit-ok",
        "screen": "audit",
        "expect": ["Revision", "Verifiera hashkedjan"],
        "steps": [
            {
                "keys": ["enter"],
                "expect": ["Kedjorna är intakta.", "verifikat"],
            },
        ],
    },
    {
        "name": "vouchers",
        "screen": "vouchers",
        "expect": ["Verifikat"],
    },
    {
        "name": "momsregler",
        "screen": "rules",
        "expect": [
            "Momsregler",
            " 1. 05     range    3000-3019",
            "10. 20     range    4510-4529",
        ],
    },
    {
        "name": "voucher-detail",
        "screen": "vouchers",
        "steps": [
            {
                "keys": ["enter"],
                "expect": ["Benämning", "Debet", "Kredit", "Underlag",
                           "kvitto.txt"],
            },
        ],
    },
    {
        "name": "kontolista-sticky",
        "screen": "reports",
        "expect": ["Rapporter"],
        "steps": [
            {
                "keys": ["7"],
                "expect": ["Konto", "Namn"],
            },
            {
                "keys": ["end"],
                "expect": ["Konto", "Namn", "konton, varav"],
            },
        ],
    },
    {
        "name": "bank-new-voucher",
        "screen": "bank",
        "expect": ["Bankavstämning", "GOLDEN INSÄTTNING", "CDON"],
        "steps": [
            {
                "keys": ["ctrln"],
                "expect": ["Nytt verifikat", "GOLDEN INSÄTTNING", "123,00",
                           "1930"],
            },
            {
                "keys": ["esc"],
                "expect": ["Bankavstämning", "GOLDEN INSÄTTNING"],
            },
            {
                "keys": ["enter"],
                "expect": ["Matcha transaktion", "Skapa nytt verifikat…"],
            },
            {
                "keys": ["esc"],
                "expect": ["Bankavstämning", "GOLDEN INSÄTTNING"],
            },
        ],
    },
    {
        "name": "bank",
        "screen": "bank",
        "expect": ["Bankavstämning", "GOLDEN INSÄTTNING", "CDON",
                   "förslag", "123,00"],
        "steps": [
            {
                "keys": ["enter", "enter"],
                "expect": ["Matchat."],
            },
            {
                "keys": ["enter"],
                "expect": ["→ A1", "1 omatchade"],
            },
        ],
    },
    {
        "name": "bank-new-voucher-post",
        "screen": "bank",
        "expect": ["Bankavstämning", "CDON", "1 omatchade"],
        "steps": [
            {
                "keys": ["ctrln"],
                "expect": ["Nytt verifikat", "CDON", "24,09", "1930"],
            },
            {
                "keys": ["\t\t\t\t\t\t\t", "3001", "\t", "24,09"],
            },
            {
                "keys": ["ctrlenter"],
                "expect": ["Bokfört"],
            },
            {
                "keys": ["enter"],
                "expect": ["Matchat med A3"],
            },
            {
                "keys": ["enter"],
                "expect": ["→ A3", "0 omatchade"],
            },
            {
                "keys": ["ctrln"],
                "expect": ["Transaktionen är redan matchad."],
            },
            {
                "keys": ["enter"],
                "expect": ["0 omatchade"],
            },
        ],
    },
    {
        "name": "invoices",
        "screen": "invoices",
        "expect": ["Fakturor", "Faktura", "Testkund AB", "12 500,00"],
    },
    {
        "name": "customers",
        "screen": "customers",
        "expect": ["Kunder", "Testkund AB"],
    },
    {
        "name": "invoice-form",
        "screen": "invoices",
        "steps": [
            {
                "keys": ["ctrln"],
                "expect": ["Ny faktura", "Kund"],
            },
        ],
    },
    {
        "name": "employees",
        "screen": "employees",
        "expect": ["Anställda", "Testanställd", "Ny anställd"],
    },
    {
        "name": "company",
        "screen": "company",
        "expect": ["Bolaget", "Fakturauppgifter", "E-post (SMTP)",
                   "Styrelseledamöter", "Test AB"],
        "steps": [
            {
                "keys": ["end", "up", "enter"],
                "expect": ["E-post (SMTP)", "SMTP-server",
                           "lämna tomt för oförändrat"],
            },
            {
                "keys": ["esc"],
                "expect": ["Bolaget", "Fakturauppgifter"],
            },
        ],
    },
    {
        "name": "system",
        "screen": "system",
        "expect": ["Systeminställningar", "Bilagornas mapp"],
    },
    {
        "name": "tax-tables",
        "screen": "tax_tables",
        "expect": ["Skattetabeller", "Status: skattetabeller för",
                   "Hämta från Skatteverket"],
    },
    {
        "name": "payroll",
        "screen": "payroll",
        "expect": ["Lönekörningar", "Ny lönekörning"],
    },
    {
        "name": "payroll-run",
        "screen": "payroll",
        "steps": [
            {
                "keys": ["ctrln"],
                "expect": ["Ny lönekörning", "Period", "Bokför körning"],
            },
            {
                "keys": ["f5"],
                "expect": ["Testanställd", "30 000,00", "Avgifter"],
            },
            {
                "keys": ["esc"],
                "expect": ["Ny lönekörning", "Bokför körning"],
            },
            {
                "keys": ["ctrlenter"],
                "expect": ["Bokför lönekörningen"],
            },
            {
                "keys": ["j"],
                "expect": ["Lönekörningen är bokförd"],
            },
            {
                "keys": ["enter"],
                "expect": ["Status: bokförd", "Lönebesked (PDF)"],
            },
            {
                "keys": ["down", "down", "down", "down", "enter"],
                "expect": ["Sparad", "Lönebesked"],
            },
        ],
    },
]

# --------------------------------------------------------------------------
# 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"]
    from datetime import date, timedelta
    start = date.fromisoformat(fy.get("start_date", "2026-01-01"))
    start_iso = start.isoformat()
    out = run_checked([str(bind / "bokfctl"), "--socket", str(sock),
                       "--user", "admin", "--password", env["BOKFD_PASSWORD"],
                       "--org", str(org_id), "voucher.post",
                       json.dumps({
                           "date": start_iso,
                           "description": "Golden verifikat",
                           "rows": [{"account": "1930", "debit_ore": 12300},
                                    {"account": "3001", "credit_ore": 12300}],
                       })], env)
    vid = json.loads(out)["result"]["id"]
    run_checked([str(bind / "bokfctl"), "--socket", str(sock),
                  "--user", "admin", "--password", env["BOKFD_PASSWORD"],
                  "--org", str(org_id), "attachment.put",
                  json.dumps({"filename": "kvitto.txt",
                              "content_base64": "a3ZpdHRv",
                              "voucher_id": vid})], env)
    golden_date = (start + timedelta(days=1)).isoformat()
    csv = (
        "Bokförd;Valutadatum;Text;Typ;Insättningar;Uttag;Bokfört saldo\n"
        f"{golden_date};{golden_date};GOLDEN INSÄTTNING;Insättning;"
        "123,00;;1000,00\n"
        f"{start_iso};{start_iso};CDON        /26-09-17;"
        "Kortköp;;24,09;975,91\n"
    )
    run_checked([str(bind / "bokfctl"), "--socket", str(sock),
                  "--user", "admin", "--password", env["BOKFD_PASSWORD"],
                  "--org", str(org_id), "bank.import",
                  json.dumps({"format": "seb",
                              "account": "1930",
                              "content_base64": base64.b64encode(
                                  csv.encode("utf-8")).decode()})], env)
    out = run_checked([str(bind / "bokfctl"), "--socket", str(sock),
                       "--user", "admin", "--password", env["BOKFD_PASSWORD"],
                       "--org", str(org_id), "customer.create",
                       json.dumps({"name": "Testkund AB",
                                   "address": "Testgatan 1",
                                   "postal_code": "111 22",
                                   "city": "Stockholm",
                                   "email": "test@example.com",
                                   "payment_days": 30})], env)
    customer_id = json.loads(out)["result"]["id"]
    due_date = (start + timedelta(days=30)).isoformat()
    run_checked([str(bind / "bokfctl"), "--socket", str(sock),
                  "--user", "admin", "--password", env["BOKFD_PASSWORD"],
                  "--org", str(org_id), "invoice.issue",
                  json.dumps({
                      "customer_id": customer_id,
                      "invoice_date": start_iso,
                      "due_date": due_date,
                      "rows": [{"description": "Konsulttjänster",
                                "quantity": "10",
                                "unit": "tim",
                                "unit_price_ore": 100000,
                                "vat_code": "25"}],
                  })], env)
    # Payroll rig: one B record covering the employee's salary, and one
    # employee, so the Lön screens have data to show.
    table_line = "30B30      1  80000 7000 7000 7000 7000 7000 7000\n"
    run_checked([str(bind / "bokfctl"), "--socket", str(sock),
                  "--user", "admin", "--password", env["BOKFD_PASSWORD"],
                  "--org", str(org_id), "payroll.tax_tables_import",
                  json.dumps({"year": start.year,
                              "content_base64": base64.b64encode(
                                  table_line.encode("utf-8")).decode()})], env)
    run_checked([str(bind / "bokfctl"), "--socket", str(sock),
                  "--user", "admin", "--password", env["BOKFD_PASSWORD"],
                  "--org", str(org_id), "employee.create",
                  json.dumps({"name": "Testanställd",
                              "personal_no": "19800101-1234",
                              "monthly_salary_ore": 3000000,
                              "tax_table": 30,
                              "tax_column": 1,
                              "email": "test@example.com"})], env)
    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,
                "BOKFD_SECRET_KEY": os.urandom(32).hex()})
    env.pop("BOKFD_SESSION", None)
    env.pop("BOKFD_TOKEN", None)
    # keep generated PDFs from opening a window: the TUI then reports where
    # it saved the file, which is what the scenarios assert on
    env.pop("DISPLAY", None)
    env.pop("WAYLAND_DISPLAY", 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:]))