tai/taicpp/tests/mplex_copy_gen.py

#!/usr/bin/env python3
# retoor <retoor@molodetz.nl>
"""Scenario generator for taicpp/tests/mplex_copy_test.cpp.

Runs tai.py's real MplexCopyMode (the oracle) over recorded tmux fixtures from test_tai.py, every default
copy-mode binding, and a seeded random fuzz, and writes one scenario file. The C++ test replays each scenario
against the C++ MplexCopyMode and compares the state line after every step byte for byte.

usage: python3 -I mplex_copy_gen.py REPO_ROOT OUT_FILE [fuzz_count] [seed]
"""
import os
import random
import sys
import socket
import tempfile
import zlib

root = sys.argv[1]
out_path = sys.argv[2]
fuzz_count = int(sys.argv[3]) if len(sys.argv) > 3 else 3600
seed = int(sys.argv[4]) if len(sys.argv) > 4 else 20251025
sys.path.insert(0, root)
import tai  # noqa: E402
import test_tai  # noqa: E402


def esc(text):
    return text.replace("%", "%25").replace("\t", "%09").replace("\n", "%0A").replace("\r", "%0D")


def digest(mode):
    block, cursor = mode.render(None)
    parts = []
    for row in block:
        for cell in row:
            parts.append("%s|%s|%s|%d;" % (cell[0], cell[1] or "", cell[2] or "", cell[3]))
        parts.append("\n")
    parts.append("%r" % (cursor,))
    return zlib.crc32("".join(parts).encode("utf-8")) & 0xFFFFFFFF


def pair(value):
    return "-" if value is None else "%d,%d" % value


def search_text(value):
    if not value:
        return "-"
    return "%s,%d,%d" % (esc(value[0]), 1 if value[1] else 0, 1 if value[2] else 0)


class Recorder:
    def __init__(self):
        self.lines = []

    def add(self, *fields):
        self.lines.append("\t".join(str(f) for f in fields))


def state(mode, pane, server, client, err, closed):
    top = server.buffers[0]["data"] if server.buffers else ""
    shown = "-"
    if client.message is not None:
        shown = esc(client.message[0])
    prompt = "-" if client.prompt is None else esc(client.prompt.label + client.prompt.text)
    return "\t".join([
        "E", str(mode.cx), str(mode.cy), str(mode.oy), str(mode.hist()), pair(mode.anchor), pair(mode.mark),
        "1" if mode.rect else "0", "1" if mode.line_select else "0", search_text(mode.search),
        search_text(pane.last_search), "1" if mode.show_position else "0", "1" if closed else "0",
        str(len(server.buffers)), esc(top), str(len(server.messages)),
        esc(server.messages[-1][1]) if server.messages else "-", str(mode.count), str(mode.lastcx), str(mode.lastsx),
        esc(err), "%08x" % digest(mode), str(pane.history_size()), str(pane.prefix_count), shown, prompt,
    ])


HOME = tempfile.mkdtemp(prefix="home.", dir=os.path.dirname(os.path.abspath(out_path)))


def make_fixture(keys, pane):
    """A real MplexServer (not listening) with one session and window holding pane, and one attached client."""
    server = tai.MplexServer(HOME)
    server.options.set("mode-keys", keys)
    session = tai.MplexSession(server.next_id("session"), "0", server.options, "/", {})
    session.size = (pane.w, pane.h + 1)
    server.sessions[session.id] = session
    window = tai.MplexWindow(server.next_id("window"), "w", pane.w, pane.h, session.options)
    window.root = tai.MplexLayoutNode("pane", pane=pane, w=pane.w, h=pane.h)
    window.active = pane
    pane.window = window
    server.panes[pane.id] = pane
    server.register_window(session, window, 0, True)
    left, right = socket.socketpair()
    client = tai.MplexClientConn(server, left, server.next_id("client"))
    client.session = session
    client.attached = True
    server.clients.append(client)
    return server, client, (left, right)


rec = Recorder()
scenarios = 0
steps_total = 0


def run_scenario(name, rows, cols, history, keys, content, steps, lines=None, exit_at_bottom=False, expect=None):
    """steps: ("C", count, words) | ("F", text) | ("R", rows, cols)"""
    global scenarios, steps_total
    pane = tai.MplexPane(1, cols, rows, history, tempfile.gettempdir())
    if lines is None:
        pane.feed(content.replace("\n", "\r\n").encode("utf-8"))
    server, client, pair_sockets = make_fixture(keys, pane)
    if lines is not None:
        mode = tai.MplexCopyMode(server, pane, lines=lines)
    else:
        mode = tai.MplexCopyMode(server, pane, exit_at_bottom=exit_at_bottom)
    pane.push_mode(mode)
    rec.add("S", esc(name), rows, cols, history, keys, esc(content), "1" if lines is not None else "0",
            "1" if exit_at_bottom else "0", len(steps))
    rec.add("E0", str(mode.cx), str(mode.cy), str(mode.oy), str(mode.hist()), mode.name, "%08x" % digest(mode))
    scenarios += 1
    closed = False
    for step in steps:
        steps_total += 1
        if closed:
            break
        err = ""
        if step[0] == "C":
            _kind, count, words = step
            rec.add("C", count, *[esc(w) for w in words])
            pane.prefix_count = count
            try:
                mode.command(words[0], words[1:], None)
            except tai.MplexError as exc:
                err = "M:" + str(exc)
            except Exception as exc:  # noqa: BLE001
                err = "X:" + str(exc)
        elif step[0] == "K":
            rec.add("K", esc(step[1]))
            try:
                mode.key(step[1], b"", client)
            except tai.MplexError as exc:
                err = "M:" + str(exc)
            except Exception as exc:  # noqa: BLE001
                err = "X:" + str(exc)
        elif step[0] in ("Y", "Z"):
            rec.add(step[0], esc(step[1]) if step[0] == "Y" else "")
            try:
                if step[0] == "Z":
                    if client.prompt is not None:
                        client.prompt.key(server, client, "Escape", b"\x1b")
                elif client.prompt is not None:
                    for ch in step[1]:
                        if client.prompt is None:
                            break
                        client.prompt.key(server, client, "Space" if ch == " " else ch, ch.encode("utf-8"))
                    if client.prompt is not None:
                        client.prompt.key(server, client, "Enter", b"\r")
            except tai.MplexError as exc:
                err = "M:" + str(exc)
            except Exception as exc:  # noqa: BLE001
                err = "X:" + str(exc)
        elif step[0] == "F":
            rec.add("F", esc(step[1]))
            pane.feed(step[1].replace("\n", "\r\n").encode("utf-8"))
        else:
            rec.add("R", step[1], step[2])
            pane.place(0, 0, step[2], step[1])
        closed = mode not in pane.modes
        rec.lines.append(state(mode, pane, server, client, err, closed))
        if closed:
            break
    if expect is not None:
        rec.add("T", expect[0], esc(expect[1]))
    rec.add("X")
    for sock in pair_sockets:
        sock.close()


# ---- command table -----------------------------------------------------------------------------
names = [n[2:].replace("_", "-") for n in tai.MplexCopyMode.__dict__ if n.startswith("c_")]
rec.add("N", *names)

expand = test_tai.MplexCopyModeVersusTmuxTests.expand

# ---- recorded tmux cases (cursor, scroll) ---------------------------------------------------------
for index, (keys, commands, count, expected) in enumerate(test_tai.MPLEX_TMUX_CASES):
    pieces = expand(commands)
    steps = []
    for position, step in enumerate(pieces):
        words = step.split()
        steps.append(("C", count if position == len(pieces) - 1 and count > 1 else 1, words))
    run_scenario("tmux-case-%d" % index, 6, 30, 100, keys, "\n".join(test_tai.MPLEX_TMUX_CONTENT) + "\n", steps,
                 expect=("cursor", expected))

# ---- recorded tmux selections (paste buffer text) --------------------------------------------------
for index, (keys, commands, expected) in enumerate(test_tai.MPLEX_TMUX_SELECTIONS):
    pieces = expand("history-top;" + commands)
    steps = [("C", 1, step.split()) for step in pieces]
    run_scenario("tmux-sel-%d" % index, 6, 30, 100, keys, "\n".join(test_tai.MPLEX_TMUX_CONTENT) + "\n", steps,
                 expect=("copied", expected))

# ---- every default copy-mode binding, as test_tai runs them ------------------------------------------
config = tai.MplexConfig()
for table in ("copy-mode", "copy-mode-vi"):
    for key, binding in config.tables[table].items():
        words = tai.MplexCommandParser.parse(binding.command)[0]
        if words[0] == "command-prompt":
            words = tai.MplexCommandParser.parse(words[-1].replace("%%", "3"))[0]
        if words[:2] != ["send-keys", "-X"]:
            continue
        words = [item for item in words[2:] if item != "--"]
        content = "\n".join("line %d (a b) {c} a,b,c" % n for n in range(40)) + "\n"
        run_scenario("binding-%s-%s" % (table, key), 6, 30, 100, "vi" if table == "copy-mode-vi" else "emacs", content,
                     [("C", 1, ["page-up"]), ("C", 1, ["begin-selection"]), ("C", 2, words)])
# every binding of both tables through key() as well (the server run_binding stub dispatches like test_tai's)
for table in ("copy-mode", "copy-mode-vi"):
    for key in list(config.tables[table]):
        content = "\n".join("line %d (a b) {c} a,b,c" % n for n in range(40)) + "\n"
        run_scenario("key-%s-%s" % (table, key), 6, 30, 100, "vi" if table == "copy-mode-vi" else "emacs", content,
                     [("C", 1, ["page-up"]), ("C", 1, ["begin-selection"]), ("K", key)])
# command-prompt bindings through the real server: open the prompt with the key, answer it (or cancel it)
PROMPT_ANSWERS = ["5", "0", "12", "x", "line 3", "(a b)", "9", "(", "line", "LINE", "{c}", "a,", " ", "-1"]
for table in ("copy-mode", "copy-mode-vi"):
    for key, binding in list(config.tables[table].items()):
        if not binding.command.startswith("command-prompt"):
            continue
        content = "\n".join("line %d (a b) {c} a,b,c" % n for n in range(40)) + "\n"
        for answer in PROMPT_ANSWERS:
            run_scenario("prompt-%s-%s-%s" % (table, key, answer), 6, 30, 100, "vi" if table == "copy-mode-vi" else "emacs",
                         content, [("C", 1, ["history-top"]), ("K", key), ("Y", answer), ("C", 1, ["search-again"]),
                                   ("C", 1, ["jump-again"])])
        run_scenario("promptesc-%s-%s" % (table, key), 6, 30, 100, "vi" if table == "copy-mode-vi" else "emacs",
                     content, [("K", key), ("Z", ""), ("C", 1, ["cursor-down"])])

# ---- seeded fuzz ----------------------------------------------------------------------------------------
rng = random.Random(seed)
WORDS = ["alpha", "beta", "gamma-delta", "foo(bar)", "baz[qux]", "{z}", "key=value;", "other_one,", "a-b", "c.d",
         "/path/to/f.py", "UPPER", "Mixed_Case", "x", "tab\tw", "é日本", "((", "))", "a,b,c", "...", "'q'",
         "(nested [mix {ok}] here)", "end", "z"]
NO_ARG = [n for n in names if not n.startswith(("jump-", "goto-line", "search-", "copy-pipe"))
          or n in ("jump-again", "jump-reverse", "jump-to-mark", "search-again", "search-reverse")]
PIPE = ["copy-pipe", "copy-pipe-and-cancel", "copy-pipe-end-of-line", "copy-pipe-end-of-line-and-cancel",
        "copy-pipe-line", "copy-pipe-line-and-cancel"]
JUMPS = ["jump-forward", "jump-backward", "jump-to-forward", "jump-to-backward"]
SEARCHES = ["search-forward", "search-backward", "search-forward-text", "search-backward-text"]
INCR = ["search-forward-incremental", "search-backward-incremental"]
PATTERNS = ["a", "e", "A", "foo", "(", "[a-c]+", "\\bkey", "x*", "", "line", "Mixed", "b.", "(a|b)c", "l.ne", "\\d+", "end$",
            "^ *a", "[", "a{2}", "o+", "\\w+", "é", "\\.", "t", "(?i)ALPHA"]
JARGS = ["a", "e", "(", " ", "x", "é", "-", "="]


def pick_content():
    lines = []
    for _n in range(rng.choice([0, 2, 5, 8, 14, 25, 45, 60])):
        indent = " " * rng.choice([0, 0, 0, 2, 4])
        count = rng.choice([0, 1, 2, 3, 4, 6, 9])
        sep = rng.choice([" ", " ", "  ", ""])
        lines.append(indent + sep.join(rng.choice(WORDS) for _k in range(count)))
    return "\n".join(lines) + ("\n" if lines else "")


def pick_step():
    roll = rng.random()
    if roll < 0.07:
        return ("F", rng.choice(["\n", "new line\n", "more text (x) y\nand\n\n", "a" * rng.choice([5, 25, 70]) + "\n"]))
    if roll < 0.10:
        return ("R", rng.choice([3, 4, 6, 8, 10]), rng.choice([8, 12, 20, 30, 40]))
    count = rng.choice([1, 1, 1, 1, 2, 3, 5, 12])
    if rng.random() < 0.08:
        table = config.tables["copy-mode-vi" if rng.random() < 0.5 else "copy-mode"]
        return ("K", rng.choice(list(table)))
    if rng.random() < 0.06:
        return ("Y", rng.choice(PROMPT_ANSWERS))
    if rng.random() < 0.02:
        return ("Z", "")
    pick = rng.random()
    if pick < 0.08:
        return ("C", count, [rng.choice(JUMPS), rng.choice(JARGS)])
    if pick < 0.14:
        return ("C", count, [rng.choice(SEARCHES), rng.choice(PATTERNS)])
    if pick < 0.17:
        return ("C", count, [rng.choice(INCR), rng.choice(["=", "+", "-", ""]) + rng.choice(PATTERNS)])
    if pick < 0.19:
        return ("C", count, ["goto-line", rng.choice(["0", "3", "9", "40", "-2", "x", " 7 ", "1_0"])])
    if pick < 0.21:
        return ("C", count, [rng.choice(PIPE), rng.choice(["cat", "tr a b"])])
    if pick < 0.215:
        return ("C", count, ["bogus-command"])
    if pick < 0.22:
        return ("C", count, [rng.choice(JUMPS)])
    return ("C", count, [rng.choice(NO_ARG)])


for number in range(fuzz_count):
    keys = rng.choice(["emacs", "vi"])
    rows = rng.choice([2, 3, 4, 5, 6, 8, 10])
    cols = rng.choice([6, 10, 14, 20, 30, 40])
    history = rng.choice([0, 3, 10, 50, 200])
    view = rng.random() < 0.06
    content = pick_content()
    steps = [pick_step() for _k in range(rng.randint(1, 14))]
    if rng.random() < 0.3:
        steps.insert(0, ("C", 1, ["history-top"]))
    if view:
        text_lines = content.split("\n")[:40] or [""]
        run_scenario("fuzz-view-%d" % number, rows, cols, history, keys, "\n".join(text_lines), steps, lines=text_lines)
    else:
        run_scenario("fuzz-%d" % number, rows, cols, history, keys, content, steps, exit_at_bottom=rng.random() < 0.1)


# ---- live tmux 3.6 oracle -----------------------------------------------------------------------------------
# Fresh random movement sequences run in the installed tmux (an attached-size 30x6 window with the same text as
# the recorded fixtures); the C++ test reports how many final cursor/scroll states agree with tmux.
TMUX_MOVES = ["cursor-up", "cursor-down", "cursor-left", "cursor-right", "start-of-line", "end-of-line",
              "back-to-indentation", "top-line", "middle-line", "bottom-line", "history-top", "history-bottom",
              "page-up", "page-down", "halfpage-up", "halfpage-down", "scroll-up", "scroll-down", "next-word",
              "next-word-end", "previous-word", "next-space", "next-space-end", "previous-space", "next-paragraph",
              "previous-paragraph", "next-matching-bracket", "previous-matching-bracket", "jump-forward",
              "jump-backward", "jump-to-forward", "jump-to-backward", "jump-again", "jump-reverse"]


def tmux_live(count):
    import shutil
    import subprocess
    import time
    if shutil.which("tmux") is None or count <= 0:
        return
    sock = "f4gen%d" % os.getpid()
    base = ["tmux", "-L", sock, "-f", "/dev/null"]
    content = "\n".join(test_tai.MPLEX_TMUX_CONTENT) + "\n"
    with tempfile.NamedTemporaryFile("w", suffix=".txt", delete=False) as handle:
        handle.write(content)
        path = handle.name
    live_rng = random.Random(seed + 1)
    try:
        for number in range(count):
            keys = "vi" if number % 2 else "emacs"
            steps = []
            for _k in range(live_rng.randint(1, 10)):
                name = live_rng.choice(TMUX_MOVES)
                words = [name]
                if name.startswith("jump-") and name not in ("jump-again", "jump-reverse"):
                    words.append(live_rng.choice(["a", "e", "o", "t", " ", "l"]))
                steps.append((1, words))
            if live_rng.random() < 0.3:
                steps[-1] = (live_rng.choice([2, 3, 5]), steps[-1][1])
            session = "s%d" % number
            subprocess.run(base + ["new-session", "-d", "-x", "30", "-y", "6", "-s", session,
                                   "cat %s; exec sleep 600" % path], check=True, capture_output=True)
            try:
                for _t in range(100):
                    shot = subprocess.run(base + ["capture-pane", "-p", "-t", session], capture_output=True, text=True).stdout
                    if "last line of text here ok" in shot:
                        break
                    time.sleep(0.02)
                subprocess.run(base + ["set-option", "-t", session, "mode-keys", keys], check=True, capture_output=True)
                subprocess.run(base + ["copy-mode", "-t", session], check=True, capture_output=True)
                for amount, words in steps:
                    argv = base + ["send-keys", "-t", session, "-X"]
                    if amount > 1:
                        argv = base + ["send-keys", "-t", session, "-N", str(amount), "-X"]
                    subprocess.run(argv + words, capture_output=True)
                shown = subprocess.run(base + ["display-message", "-p", "-t", session,
                                               "#{copy_cursor_x} #{copy_cursor_y} #{scroll_position}"],
                                       capture_output=True, text=True).stdout.strip()
            finally:
                subprocess.run(base + ["kill-session", "-t", session], capture_output=True)
            run_scenario("tmuxlive-%d" % number, 6, 30, 100, keys, content,
                         [("C", amount, words) for amount, words in steps], expect=("livecursor", shown))
    finally:
        subprocess.run(base + ["kill-server"], capture_output=True)
        os.unlink(path)


tmux_live(int(os.environ.get("F4_TMUX_LIVE", "240")))

with open(out_path, "w", encoding="utf-8") as handle:
    handle.write("\n".join(rec.lines) + "\n")
import shutil  # noqa: E402

shutil.rmtree(HOME, ignore_errors=True)
sys.stderr.write("scenarios=%d steps=%d\n" % (scenarios, steps_total))