path: root/packaging/checks/split-button-live.py

#!/usr/bin/env python3
"""Live check: is the `+` split button DRAWN, and does clicking the drawn glyph
split the pane?

The unit tests can only prove the title line composes correctly. This drives a
real session with the live config's frame settings -- centred titles, rounded
corners, a pane gap, no default pane names -- because every one of those takes a
different path through the title builders, and the gap moves the drawn frame
away from the pane's allocation.

It then attaches a SECOND client, because that is what the live box always had
(the phantom client a headless `sl0ppi up` leaks) and it switches the frame onto
a third title builder again -- the one with a "MY FOCUS" middle, which is where
the button was being dropped for two releases.

Usage: split-button-live.py [path-to-sl0ppij] [path-to-config.kdl]

With no config it uses a minimal one carrying just the live frame settings.
Pass `config/config.kdl` to drive the real thing, layouts and plugins included.
Exits non-zero with a description of what it saw.
"""

import fcntl
import os
import pty
import re
import shutil
import struct
import subprocess
import sys
import tempfile
import termios
import time

BIN = os.path.abspath(
    sys.argv[1]
    if len(sys.argv) > 1
    else os.path.join(os.path.dirname(__file__), "../../sl0ppij/target/fastdev/sl0ppij")
)
ROWS, COLS = 30, 100
SESSION = f"sl0ptest-splitbutton-{os.getpid()}"

CONFIG = """
ui {
    pane_frames {
        rounded_corners true
        title_alignment "center"
        gap 1
        padding 0 0 0 0
        gap_aspect 2
        hide_default_pane_names true
    }
}
pane_frames true
pane_frame_style "full"
default_mode "locked"
show_startup_tips false
"""

ANSI = re.compile(r"\x1b\[[0-9;?]*[a-zA-Z]|\x1b[]P][^\x07\x1b]*(\x07|\x1b\\)?|\x1b[()][0-9A-B]")


def screen_rows(raw):
    """Rebuild the visible grid from the pty stream: cursor moves, erases and
    printable runs. Not a terminal emulator, but it must honour the erases --
    without them a pane that shrinks leaves its old frame behind in the model,
    and the check then clicks a `+` that is no longer on screen and calls the
    product broken."""
    grid = {}
    row = col = 0
    i = 0

    def erase(cells):
        for cell in cells:
            grid.pop(cell, None)

    while i < len(raw):
        ch = raw[i]
        if ch == "\x1b":
            m = re.match(r"\x1b\[([0-9;?]*)([a-zA-Z])", raw[i:])
            if m:
                args, verb = m.group(1), m.group(2)
                nums = [int(n) for n in args.replace("?", "").split(";") if n != ""]
                first = nums[0] if nums else 0
                if verb in "Hf":
                    row = (nums[0] - 1) if nums else 0
                    col = (nums[1] - 1) if len(nums) > 1 else 0
                elif verb == "A":
                    row = max(0, row - max(1, first))
                elif verb == "B":
                    row += max(1, first)
                elif verb == "C":
                    col += max(1, first)
                elif verb == "D":
                    col = max(0, col - max(1, first))
                elif verb == "G":
                    col = max(0, first - 1)
                elif verb == "J":
                    if first == 0:
                        erase(
                            [(row, c) for c in range(col, COLS)]
                            + [(r, c) for r in range(row + 1, ROWS) for c in range(COLS)]
                        )
                    elif first == 1:
                        erase(
                            [(row, c) for c in range(0, col + 1)]
                            + [(r, c) for r in range(0, row) for c in range(COLS)]
                        )
                    else:
                        grid.clear()
                elif verb == "K":
                    if first == 0:
                        erase([(row, c) for c in range(col, COLS)])
                    elif first == 1:
                        erase([(row, c) for c in range(0, col + 1)])
                    else:
                        erase([(row, c) for c in range(COLS)])
                i += m.end()
                continue
            m = re.match(r"\x1b][^\x07\x1b]*(\x07|\x1b\\)", raw[i:])
            if m:
                i += m.end()
                continue
            i += 1
            continue
        if ch == "\r":
            col = 0
        elif ch == "\n":
            row += 1
            col = 0
        elif ch >= " ":
            grid[(row, col)] = ch
            col += 1
        i += 1
    out = []
    for r in range(ROWS):
        line = "".join(grid.get((r, c), " ") for c in range(COLS))
        out.append(line.rstrip())
    return out


def spawn_client(env, args, rows=ROWS, cols=COLS):
    pid, fd = pty.fork()
    if pid == 0:
        os.execvpe(BIN, [BIN] + args, env)
    fcntl.ioctl(fd, termios.TIOCSWINSZ, struct.pack("HHHH", rows, cols, 0, 0))
    # non-blocking: a settled client produces nothing, and a blocking read on
    # its pty would hang this check instead of failing it
    os.set_blocking(fd, False)
    return pid, fd


def repaint(fd, cols=COLS, rows=ROWS):
    """Force a full redraw and return an EMPTY buffer to accumulate it into.

    The screen model here is deliberately simple, and the way a simple model
    goes wrong is stale cells: a pane that shrinks leaves its old frame behind,
    and the check then clicks a `+` that is not on screen any more. A resize
    makes the server repaint every cell, so reading from scratch afterwards
    cannot see anything that is no longer drawn.
    """
    # shrink, throw that paint away, then grow back: what the caller reads next
    # is one complete repaint of the final size and nothing else
    fcntl.ioctl(fd, termios.TIOCSWINSZ, struct.pack("HHHH", rows, cols - 1, 0, 0))
    time.sleep(0.8)
    try:
        while os.read(fd, 1 << 20):
            pass
    except (BlockingIOError, OSError):
        pass
    fcntl.ioctl(fd, termios.TIOCSWINSZ, struct.pack("HHHH", rows, cols, 0, 0))
    return ""


def read_screen(fd, buffered, seconds=2.5, until=None, quiet_for=0.7):
    """Accumulate output until the stream goes QUIET, not until a timer expires.

    A repaint arrives in many chunks, and stopping mid-repaint yields a screen
    that is half old and half new -- which reads exactly like a button drawn
    where it is not. Waiting for silence is the difference between a check that
    finds bugs and one that invents them.
    """
    deadline = time.time() + seconds
    last_data = time.time()
    while time.time() < deadline:
        time.sleep(0.15)
        try:
            chunk = os.read(fd, 1 << 20).decode("utf8", "replace")
        except BlockingIOError:
            chunk = ""
        except OSError:
            break
        if chunk:
            buffered += chunk
            last_data = time.time()
            continue
        if until and until(screen_rows(buffered)):
            break
        if buffered and time.time() - last_data >= quiet_for:
            break
    return buffered, screen_rows(buffered)


def find_button(rows):
    """(row, 0-based column of the `+`) on a pane's top frame row."""
    found = find_buttons(rows)
    return found[0] if found else None


def find_buttons(rows):
    """Every drawn split button: (row, 0-based column of the `+`).

    A frame row can hold more than one, once panes sit side by side.

    Identified by the CORNER it sits against: the button is the three columns
    immediately left of a pane's top-right corner (`split_button_columns`). A
    bare " + " search also matches the stack list's "+N more" and the short
    multi-client indication, and those share a screen row with a real frame as
    soon as two panes sit side by side -- which then reads as a button that does
    nothing when clicked, i.e. as this feature's own bug.
    """
    found = []
    for y, line in enumerate(rows):
        start = 0
        while True:
            at = line.find(" + ", start)
            if at < 0:
                break
            corner = line[at + 3] if at + 3 < len(line) else ""
            if corner in ("╮", "┐"):  # rounded / square top-right
                found.append((y, at + 1))
            start = at + 1
    return found


def main():
    tmp = tempfile.mkdtemp(prefix="sl0p-splitbutton-")
    if len(sys.argv) > 2:
        config_file = os.path.abspath(sys.argv[2])
        config_dir = os.path.dirname(config_file)
        print(f"config: {config_file}")
    else:
        config_file = os.path.join(tmp, "config.kdl")
        config_dir = tmp
        with open(config_file, "w") as f:
            f.write(CONFIG)
    env = dict(os.environ)
    env.update(
        {
            "SL0PPIJ_SOCKET_DIR": os.path.join(tmp, "sock"),  # isolate from live sessions
            "SL0PPIJ_CONFIG_FILE": config_file,
            "SL0PPIJ_CONFIG_DIR": config_dir,
            "SL0PPIJ_CACHE_DIR": os.path.join(tmp, "cache"),
            "TERM": "xterm-256color",
            "SHELL": "/bin/sh",
            "PS1": "$ ",
        }
    )

    pid, fd = spawn_client(env, ["-s", SESSION])
    second_pid = second_fd = None

    raw = ""
    failures = []
    try:
        raw, rows = read_screen(fd, raw, seconds=20, until=find_button)

        # 1. one client: is it drawn at all, on the pane's top frame row?
        hit = find_button(rows)
        if hit is None:
            failures.append("one client: the split button was not drawn on any frame row")
            print("\n".join(f"{i:02d}|{r}" for i, r in enumerate(rows)))
            return failures
        y, x = hit
        print(f"one client:  drawn at row {y}, column {x}")
        print(f"   {rows[y]}")

        # 2. two clients: the frame grows a "MY FOCUS" middle, which is a
        #    different title builder -- and the one the live box always used,
        #    because a headless `sl0ppi up` leaves a phantom client behind.
        second_pid, second_fd = spawn_client(env, ["attach", SESSION])
        _, second_rows = read_screen(second_fd, "", seconds=10, until=find_button)
        raw, rows = read_screen(fd, repaint(fd), seconds=5)
        focus_middle = any("FOCUS" in r or "| ME |" in r for r in rows + second_rows)
        if not focus_middle:
            failures.append("two clients: no focus indication -- the second client did not attach")
        hit2 = find_button(rows)
        print(f"two clients: drawn at {hit2}")
        if hit2 is None:
            failures.append("two clients: the split button was dropped from the title line")
            print("\n".join(f"{i:02d}|{r}" for i, r in enumerate(rows)))
        elif hit2 != hit:
            failures.append(f"two clients: the button moved, {hit} -> {hit2}")
        if hit2:
            print(f"   {rows[hit2[0]]}")

        def pane_count():
            proc = subprocess.run(
                [BIN, "-s", SESSION, "action", "list-panes", "--json"],
                env=env,
                capture_output=True,
                text=True,
            )
            if os.environ.get("DEBUG"):
                print(f"list-panes: {proc.stdout[:2000]!r} {proc.stderr[:500]!r}")
            return proc.stdout.count('"pane_x"')

        def pane_geometries():
            out = subprocess.run(
                [BIN, "-s", SESSION, "action", "list-panes", "--json"],
                env=env,
                capture_output=True,
                text=True,
            ).stdout
            try:
                import json

                return [
                    {
                        k: p[k]
                        for k in (
                            "title",
                            "is_plugin",
                            "is_focused",
                            "pane_x",
                            "pane_y",
                            "pane_columns",
                            "pane_rows",
                            "pane_content_x",
                            "pane_content_y",
                        )
                    }
                    for p in json.loads(out)
                    if not p["is_plugin"]
                ]
            except Exception as e:  # noqa: BLE001 - diagnostics only
                return [f"could not parse list-panes: {e}"]

        before = pane_count()

        # 2. clicking two columns right of the glyph -- where the OLD hit-test
        #    looked, out in the pane gap -- must do nothing
        os.write(fd, f"\x1b[<0;{x + 3};{y + 1}M".encode())
        os.write(fd, f"\x1b[<0;{x + 3};{y + 1}m".encode())
        time.sleep(1.5)
        if pane_count() != before:
            failures.append("a click in the gap beside the button split the pane")

        # 3. clicking the drawn glyph splits
        def click(col, row):
            os.write(fd, f"\x1b[<0;{col + 1};{row + 1}M".encode())
            os.write(fd, f"\x1b[<0;{col + 1};{row + 1}m".encode())
            time.sleep(1.5)

        def settled():
            """A freshly repainted, fully arrived screen."""
            return read_screen(fd, repaint(fd), seconds=8)

        click(x, y)
        after = pane_count()
        print(f"panes: {before} -> {after}")
        if after != before + 1:
            failures.append(f"clicking the drawn button did not split ({before} -> {after})")

        # 4. EVERY drawn button must split the pane it belongs to, not just the
        #    first one -- panes away from the screen edge are inset differently
        #    (the gap is only applied at the edges), an unfocused pane is a
        #    different code path in the handler, and a stacked pane's frame is
        #    one row somewhere else entirely.
        for phase, setup in (
            ("tiled, side by side", None),
            ("stacked", ["action", "new-pane", "--", "/bin/sh"]),
        ):
            if setup:
                subprocess.run([BIN, "-s", SESSION] + setup, env=env, capture_output=True)
                time.sleep(1)
                ids = subprocess.run(
                    [BIN, "-s", SESSION, "action", "list-panes", "--json"],
                    env=env,
                    capture_output=True,
                    text=True,
                ).stdout
                terminals = [
                    str(m)
                    for m in re.findall(r'"id": (\d+),\n    "is_plugin": false', ids)
                ]
                if len(terminals) >= 2:
                    subprocess.run(
                        [BIN, "-s", SESSION, "action", "stack-panes", "--"] + terminals[:3],
                        env=env,
                        capture_output=True,
                    )
                time.sleep(1.5)
            raw, rows = settled()
            print(f"{phase}: {len(find_buttons(rows))} button(s) drawn at {find_buttons(rows)}")
            if not find_buttons(rows):
                failures.append(f"{phase}: no split button drawn at all")
            # Re-scan between clicks: every split moves the panes, so a list of
            # buttons collected before the first click is stale by the second.
            clicked = set()
            for _ in range(4):
                raw, rows = settled()
                todo = [b for b in find_buttons(rows) if b not in clicked]
                if not todo:
                    break
                by, bx = todo[0]
                clicked.add((by, bx))
                n0 = pane_count()
                click(bx, by)
                n1 = pane_count()
                print(f"   click ({by},{bx}): {n0} -> {n1}")
                if n1 != n0 + 1:
                    failures.append(
                        f"{phase}: the button drawn at row {by} column {bx} did not split"
                        f" ({n0} -> {n1})"
                    )
                    # the screen the button was found on, so the next reader can
                    # see WHICH pane's button was ignored, plus the geometry the
                    # server thinks those panes have
                    print(f"   screen when ({by},{bx}) was clicked:")
                    print("\n".join(f"   {i:02d}|{r}" for i, r in enumerate(rows)))
                    print("   panes (x, y, cols, rows) per the server:")
                    for p in pane_geometries():
                        print(f"     {p}")

        if os.environ.get("DEBUG"):
            raw, rows = read_screen(fd, raw, seconds=1)
            print("\n".join(f"{i:02d}|{r}" for i, r in enumerate(rows)))
        return failures
    finally:
        subprocess.run(
            [BIN, "kill-session", SESSION], env=env, capture_output=True, text=True
        )
        # `kill-session NAME`, not `-s NAME kill-session`: the latter parses
        # but leaves the session running. Then DELETE it -- a killed session
        # stays in `list-sessions` as "EXITED - attach to resurrect", so a day
        # of check runs silently fills the user's session picker.
        subprocess.run([BIN, "delete-session", SESSION], env=env, capture_output=True, text=True)
        # kill-session does not always take the server with it, and a leftover
        # test session is how a "MY FOCUS" bug report gets filed against nobody.
        #
        # Matched on `comm` plus THIS run's socket dir, never `pkill -f`: that
        # matches the checking process's own command line, and has killed real
        # clients here before.
        time.sleep(1)
        ps = subprocess.run(
            ["ps", "-eo", "pid,comm,args"], capture_output=True, text=True
        ).stdout
        for line in ps.splitlines():
            fields = line.split(None, 2)
            if len(fields) < 3:
                continue
            pid_str, comm, args = fields
            if comm == os.path.basename(BIN) and tmp in args:
                try:
                    os.kill(int(pid_str), 15)
                    print(f"cleaned up leftover server {pid_str}")
                except (ProcessLookupError, ValueError, PermissionError):
                    pass
        shutil.rmtree(tmp, ignore_errors=True)
        for p, f in ((pid, fd), (second_pid, second_fd)):
            if f is None:
                continue
            try:
                os.close(f)
            except OSError:
                pass
            os.waitpid(p, os.WNOHANG)


if __name__ == "__main__":
    problems = main()
    for p in problems:
        print(f"FAIL: {p}")
    print("ok: the split button is drawn where it is clicked" if not problems else "")
    sys.exit(1 if problems else 0)