#!/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}", "Bolaget", "System", "Skattetabeller"], }, { "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", "Företagsuppgifter", "Fakturauppgifter", "E-post (SMTP)", "Anställda", "Kunder", "Momsregler"], "steps": [ { "keys": ["3"], "expect": ["E-post (SMTP)", "SMTP-server", "lämna tomt för oförändrat"], }, { "keys": ["esc"], "expect": ["Bolaget", "Företagsuppgifter"], }, { "keys": ["5"], "expect": ["Anställda", "Testanställd"], }, { "keys": ["esc"], "expect": ["Bolaget", "Kunder"], }, ], }, { "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:]))