#!/usr/bin/env python3 """Fixture generator for mplex_model_test.cpp (fork F3 classes). Loads ../../tai.py with importlib, drives the real Python classes (MplexCanvas, MplexStatusText, MplexFont, MplexPrompt, MplexChooser, MplexFlowScreen, MplexRenderer, MplexTargets, MplexCtx, MplexSession, MplexWindow, MplexClientConn) over seeded random scenarios and writes one JSON object per line: the scenario input plus what Python produced. The C++ test replays the input through the C++ classes and compares. Usage: mplex_model_gen.py OUT.jsonl SCRATCH_DIR """ import collections import importlib.util import json import os import random import struct import sys import time HERE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.path.abspath(os.path.join(HERE, "..", "..")) spec = importlib.util.spec_from_file_location("tai_mod", os.path.join(ROOT, "tai.py")) tai = importlib.util.module_from_spec(spec) sys.modules["tai_mod"] = tai spec.loader.exec_module(tai) OUT = sys.argv[1] SCRATCH = sys.argv[2] os.makedirs(SCRATCH, exist_ok=True) rnd = random.Random(20260515) cases = [] FGS = [None, "31", "42", "38;5;12", "38;2;1;2;3", "94", "107", "48;5;200", "48;2;255;0;9"] CHARS = list("ab Zz-é世#|") + ["x", " "] def cell_json(cell): if cell is None: return [0, None, None, 0] return [ord(cell[0]), cell[1], cell[2], cell[3]] def cell_from(item): return (chr(item[0]), item[1], item[2], item[3]) def rand_cell(): return (rnd.choice(CHARS), rnd.choice(FGS), rnd.choice(FGS), rnd.choice([0, 0, 1, 2, 4, 5, 33, 64, 128, 255])) def canvas_json(canvas): return {"rows": canvas.rows, "cols": canvas.cols, "cells": [[cell_json(c) for c in row] for row in canvas.cells], "cursor": list(canvas.cursor) if canvas.cursor is not None else None} # ---- A: border_char ----------------------------------------------------------------------- for lines in ("single", "double", "heavy", "simple", "bogus"): for mask in range(16): flags = [bool(mask & 1), bool(mask & 2), bool(mask & 4), bool(mask & 8)] cases.append({"t": "border", "lines": lines, "flags": flags, "want": ord(tai.MplexCanvas.border_char(*flags, lines))}) # ---- B: encode ----------------------------------------------------------------------------- for _ in range(80): rows, cols = rnd.randint(1, 6), rnd.randint(1, 9) canvas = tai.MplexCanvas(rows, cols) for r in range(rows): for c in range(cols): canvas.cells[r][c] = rand_cell() if rnd.random() < 0.7 else canvas.BLANK canvas.cursor = (rnd.randrange(cols), rnd.randrange(rows)) if rnd.random() < 0.6 else None mode = rnd.choice(["none", "same", "mutate", "resize"]) prev = None if mode == "same": prev = canvas.encode(None)[1] elif mode == "mutate": other = tai.MplexCanvas(rows, cols) other.cells = [list(row) for row in canvas.cells] for _k in range(rnd.randint(1, 4)): other.cells[rnd.randrange(rows)][rnd.randrange(cols)] = rand_cell() prev = other.encode(None)[1] elif mode == "resize": other = tai.MplexCanvas(rows + 1, cols) prev = other.encode(None)[1] data, snap = canvas.encode(prev) cases.append({"t": "encode", "canvas": canvas_json(canvas), "prev": None if prev is None else [[cell_json(c) for c in row] for row in prev], "want": data.hex()}) # ---- C: status text ----------------------------------------------------------------------------- PIECES = ["abc", "#[fg=red]", "#[bg=blue,bold]", "#[default]", "#[nobold]", "#[bogus]", "#[", "x#[fg=green", "é", "]", "#[fg=colour200,reverse]", "#[bg=#102030]", "#[none]", "#[fg=brightred]", "q", " "] for _ in range(80): text = "".join(rnd.choice(PIECES) for _k in range(rnd.randint(0, 7))) base = (rnd.choice(FGS[:6]), rnd.choice(FGS[:6]), rnd.choice([0, 1, 4])) reset = None if rnd.random() < 0.5 else (rnd.choice(FGS[:6]), rnd.choice(FGS[:6]), rnd.choice([0, 2])) cells = tai.MplexStatusText.cells(text, base, reset) cases.append({"t": "statustext", "text": text, "base": list(base), "reset": None if reset is None else list(reset), "want": [cell_json(c) for c in cells]}) # ---- D: font ------------------------------------------------------------------------------------------ for text in ["", "0", "12:34", "09:59PM", "AM", "x", "7:", "0123456789:AMP"]: for bg in (None, "44", "48;5;9"): rows = tai.MplexFont.draw(text, bg) cases.append({"t": "font", "text": text, "bg": bg, "width": tai.MplexFont.width(text), "want": [[cell_json(c) for c in row] for row in rows]}) # ---- E: prompt --------------------------------------------------------------------------------------------- class PServer: def __init__(self, buf): self.buf = buf def top_buffer_text(self): return self.buf class PClient: def __init__(self): self.prompt = None self.redispatched = [] def redispatch(self, name, raw): self.redispatched.append(name) EDIT_KEYS = ["Enter", "C-j", "Escape", "C-c", "C-g", "C-a", "Home", "C-e", "End", "C-b", "Left", "C-f", "Right", "M-b", "M-f", "BSpace", "C-h", "DC", "C-d", "C-k", "C-u", "C-w", "C-y", "C-t", "Up", "C-p", "Down", "C-n", "C-s", "C-r", "F5", "Tab", "M-x", "S-Up"] def key_raw(name): if name == "Space": return b" " if len(name) == 1 or (name and name[0].isalpha() and len(name) == 1): return name.encode("utf-8") return tai.MplexKeys.to_bytes(name) if name not in ("M-x",) else b"\x1bx" def rand_key(digits_bias): pick = rnd.random() if pick < 0.45: return rnd.choice(list("abyYn0123456789 -_é世#")) if pick < 0.5: return "Space" if digits_bias and pick < 0.7: return rnd.choice(list("0123456789")) return rnd.choice(EDIT_KEYS) def run_prompt_case(): mode = rnd.choice(["plain"] * 6 + ["single", "numeric", "key_name", "confirm"]) label = rnd.choice(["", "cmd: ", "é> "]) text = rnd.choice(["", "hello", "two words here", "é世x"]) history = rnd.choice([[], ["one"], ["a b", "c", "d"]]) buf = rnd.choice(["", "PASTE", "é"]) with_step = rnd.random() < 0.5 keys = [rand_key(mode == "numeric") for _i in range(rnd.randint(1, 30))] events = [] srv = PServer(buf) client = PClient() hist_list = list(history) kwargs = {"history": hist_list} if mode == "single": kwargs["single"] = True elif mode == "numeric": kwargs["numeric"] = True elif mode == "key_name": kwargs["key_name"] = True elif mode == "confirm": kwargs["confirm"] = True kwargs["on_change"] = lambda t: events.append(["change", t]) kwargs["on_cancel"] = lambda: events.append(["cancel"]) if with_step: kwargs["on_step"] = lambda n, t: events.append(["step", n, t]) prompt = tai.MplexPrompt(label, text, lambda v: events.append(["done", v]), **kwargs) client.prompt = prompt trace = [] for name in keys: if client.prompt is None: break prompt.key(srv, client, name, key_raw(name)) trace.append([prompt.text, prompt.pos, prompt.hist_index, client.prompt is None]) cases.append({"t": "prompt", "mode": mode, "label": label, "text": text, "history": history, "buf": buf, "step": with_step, "keys": [[k, key_raw(k).hex()] for k in keys[:len(trace)]], "view": [prompt.view()[0], prompt.view()[1]], "want": {"trace": trace, "events": events, "history_after": hist_list}}) for _ in range(300): run_prompt_case() # explicit printable() checks for name, raw in [("a", b"a"), ("Space", b" "), ("é", "é".encode()), ("C-a", b"\x01"), ("Enter", b"\r"), ("x", b"\xff"), ("é", b"\xc3"), ("", b"\xff"), ("ab", b"ab"), ("世", "世".encode()), ("\x01", b"\x01")]: cases.append({"t": "printable", "name": name, "raw": raw.hex(), "want": tai.MplexPrompt.printable(name, raw)}) # ---- F: chooser ---------------------------------------------------------------------------------------------- class CServer: def __init__(self): self.options = tai.MplexOptions() self.panes = {} self.dirty = 0 def pane_options(self, pane): return pane.window.options if pane.window is not None else self.options def mark_dirty(self, window=None): self.dirty += 1 def ask_confirm(self, client, text, callback): client.prompt = tai.MplexPrompt(text + " ", "", lambda value: callback(), confirm=True) client.dirty = True def refresh_choosers(self): for pane in list(self.panes.values()): for mode in list(pane.modes): if isinstance(mode, tai.MplexChooser): mode.rebuild() class CClient: def __init__(self): self.prompt = None self.dirty = False self.overlay = None def tree_spec(depth=0): count = rnd.randint(1, 4) nodes = [] for _i in range(count): label = rnd.choice(["sess:one", "win: two", "plain", "é:x", "a long label that is very long indeed 0123456789"]) kids = tree_spec(depth + 1) if depth < 2 and rnd.random() < 0.45 else [] nodes.append([label, rnd.randint(0, 999), kids]) return nodes def build_nodes(spec_nodes): return [tai.MplexChoice(label, payload=payload, children=build_nodes(kids), expanded=(rnd.random() < 0.0)) for label, payload, kids in spec_nodes] CKEYS = ["Up", "Down", "k", "j", "PPage", "NPage", "Home", "End", "g", "G", "Right", "Left", "l", "h", "+", "-", "t", "v", "x", "y", "n", "Enter", "q", "Escape", "C-b", "C-f", "C-p", "C-n", "0", "1", "2", "5", "9", "z"] def remove_payload(nodes, payload): keep = [] for n in nodes: if n[1] == payload: continue n[2] = remove_payload(n[2], payload) keep.append(n) return keep def chooser_state(mode, pane, client, extra): def walk(nodes): return [[n.label, n.payload, n.depth if hasattr(n, "depth") else 0, n.expanded, n.tag, walk(n.children)] for n in nodes] state = {"cursor": mode.cursor, "visible": [[n.label, n.depth, n.expanded, n.tag] for n in mode.visible], "closed": mode not in pane.modes, "prompt": None if client.prompt is None else client.prompt.view()[0], "chosen": list(extra["chosen"]), "deleted": list(extra["deleted"]), "show_preview": mode.show_preview} if mode in pane.modes: grid, cur = mode.render(client) state["grid"] = [[cell_json(c) for c in row] for row in grid] return state def run_chooser_case(): w, h = rnd.randint(8, 40), rnd.randint(2, 20) spec_nodes = tree_spec() with_delete = rnd.random() < 0.6 with_preview = rnd.random() < 0.6 selected = None flat = [] def collect(nodes): for n in nodes: flat.append(n[1]) collect(n[2]) collect(spec_nodes) if flat and rnd.random() < 0.4: selected = rnd.choice(flat) srv = CServer() pane = tai.MplexPane(0, w, h, 100, SCRATCH, "chooser") srv.panes[0] = pane client = CClient() extra = {"chosen": [], "deleted": []} data = [list(n) for n in spec_nodes] expand_first = rnd.random() < 0.5 def builder(): built = build_nodes(data) if expand_first: for node in built: node.expanded = True return built def on_choose(node, cl): extra["chosen"].append(node.payload) def on_delete(node, cl): extra["deleted"].append(node.payload) data[:] = remove_payload(data, node.payload) def preview(item, pw, ph): return [[(c, None, None, 0) for c in ("pv:" + item.label)[:pw]] for _k in range(ph)] mode = tai.MplexChooser(srv, pane, rnd.choice(["tree-mode", "buffer-mode"]), builder, on_choose, on_delete=on_delete if with_delete else None, preview=preview if with_preview else None, selected=selected) pane.push_mode(mode) keys = [rnd.choice(CKEYS) for _i in range(rnd.randint(1, 25))] states = [chooser_state(mode, pane, client, extra)] used = [] for name in keys: if mode not in pane.modes and client.prompt is None: break used.append(name) if client.prompt is not None: client.prompt.key(srv, client, name, name.encode()) else: mode.key(name, name.encode(), client) states.append(chooser_state(mode, pane, client, extra)) cases.append({"t": "chooser", "w": w, "h": h, "tree": spec_nodes, "delete": with_delete, "preview": with_preview, "selected": selected, "expand_first": expand_first, "name": mode.name, "keys": used, "want": states}) for _ in range(120): run_chooser_case() # ---- G: flow screen ------------------------------------------------------------------------------------------------ class FServer: def __init__(self): self.flows = {} class FClient: def __init__(self): self.overlay = None self.dirty = False WORDS = ["alpha", "beta", "gamma", "delta", "epsilon", "a-very-long-hyphenated-word-that-needs-breaking", "x", "é世", "zeta", "supercalifragilisticexpialidocious_supercalifragilisticexpialidocious", "tab\there"] def rand_text(): paras = [] for _p in range(rnd.randint(0, 4)): paras.append(" ".join(rnd.choice(WORDS) for _w in range(rnd.randint(0, 25)))) return "\n".join(paras) FKEYS = ["Up", "Down", "k", "j", "C-p", "C-n", "Home", "End", "g", "G", "Enter", "q", "Escape", "C-c", "1", "2", "9", "0", "z", "x"] for _ in range(160): n_opt = rnd.choice([0, 0, 1, 2, 3, 5, 9, 10, 12]) options = ["opt %d %s" % (i, rnd.choice(WORDS)) for i in range(n_opt)] title = rnd.choice(["", "Deploy", "A long title " * 8, "é"]) text = rand_text() cursor = rnd.randint(-2, n_opt + 2) rows, cols = rnd.randint(6, 40), rnd.randint(8, 100) area_y = rnd.choice([0, 0, 1, 2]) srv = FServer() client = FClient() srv.flows[7] = {"answer": None, "settled": None, "until": 0} screen = tai.MplexFlowScreen(srv, 7, title, text, options, cursor, 123.5) client.overlay = screen canvas = tai.MplexCanvas(rows, cols) screen.draw(canvas, area_y) initial_cursor = screen.cursor lines = screen.rows(max(20, min(cols - 4, 68))) keys = [rnd.choice(FKEYS) for _i in range(rnd.randint(0, 8))] drop_at = rnd.choice([None, None, None, 2]) states = [] used = [] for i, name in enumerate(keys): if drop_at is not None and i == drop_at: srv.flows.pop(7, None) used.append(name) screen.key(name, name.encode(), client) ans = srv.flows.get(7, {}).get("answer") states.append([screen.cursor, ans, client.overlay is None, client.dirty]) if client.overlay is None: break cases.append({"t": "flowscreen", "title": title, "text": text, "options": options, "cursor": cursor, "rows": rows, "cols": cols, "area_y": area_y, "canvas": canvas_json(canvas), "initial_cursor": initial_cursor, "lines": lines, "keys": used, "drop_at": drop_at, "want": states}) # ---- H: model fixtures: renderer, targets, session, window --------------------------------------------------------------------- class Model: """The same op list is replayed by the C++ test; every op mirrors one Python statement group.""" def __init__(self, home): self.server = tai.MplexServer(home) self.sessions = [] self.windows = [] self.cur_session = None self.cur_window = None self.errors = [] def op(self, item): kind = item[0] server = self.server try: if kind == "session": _k, name, cols, rows = item session = tai.MplexSession(server.next_id("session"), name, server.options, "/", {}) session.size = (cols, rows) server.sessions[session.id] = session self.sessions.append(session) self.cur_session = session elif kind == "window": _k, name, index, select = item session = self.cur_session cols, rows = session.size window = tai.MplexWindow(server.next_id("window"), name, max(1, cols), max(1, rows - (1 if server.options.get("status") else 0)), session.options) window.session = session pane = tai.MplexPane(server.next_id("pane"), window.width, window.height, 50, server.dir, "mdl") pane.window = window server.panes[pane.id] = pane window.root = tai.MplexLayoutNode("pane", pane=pane, w=window.width, h=window.height) window.active = pane pane.last_active = 1.0 server.register_window(session, window, index, select) window.relayout() self.windows.append(window) self.cur_window = window elif kind == "split": _k, pos, direction, before, size = item window = self.cur_window target = window.panes()[pos % len(window.panes())] leaf = window.leaf_of(target) span = leaf.w if direction == "lr" else leaf.h if span < 3: raise tai.MplexError("no space for new pane") pane = tai.MplexPane(server.next_id("pane"), 1, 1, 50, server.dir, "mdl") pane.window = window server.panes[pane.id] = pane window.root, _fresh = tai.MplexLayout.split(window.root, leaf, direction, before, pane, size) window.layout_index = -1 window.relayout() pane.last_active = float(pane.id) window.set_active(pane) window.active.last_active = float(pane.id) elif kind == "feed": _k, pos, text = item panes = self.cur_window.panes() panes[pos % len(panes)].feed(text.encode("utf-8")) elif kind == "active": _k, pos = item panes = self.cur_window.panes() pane = panes[pos % len(panes)] self.cur_window.set_active(pane) pane.last_active = float(pane.id) elif kind == "zoom": _k, pos = item panes = self.cur_window.panes() self.cur_window.zoomed = panes[pos % len(panes)] self.cur_window.relayout() elif kind == "opt": _k, scope, name, value = item target = {"server": lambda: server.options, "session": lambda: self.cur_session.options, "window": lambda: self.cur_window.options}[scope]() target.set(name, value) elif kind == "select": _k, index = item self.cur_session.select(index) elif kind == "sessopt": _k, which, name, value = item self.sessions[which % len(self.sessions)].options.set(name, value) elif kind == "mark": _k, pos = item panes = self.cur_window.panes() server.marked = panes[pos % len(panes)] elif kind == "curwin": _k, widx = item self.cur_window = self.windows[widx % len(self.windows)] self.cur_session = self.cur_window.session except tai.MplexError as exc: self.errors.append([kind, str(exc)]) def dump(self): out = {"errors": self.errors, "sessions": []} for s in self.server.sessions.values(): wins = [] for index in sorted(s.windows): w = s.windows[index] wins.append({"index": index, "id": w.id, "name": w.name, "w": w.width, "h": w.height, "active": w.active.id, "last_pane": w.last_pane.id if w.last_pane else None, "zoomed": w.zoomed.id if w.zoomed else None, "panes": [[p.id, p.x, p.y, p.w, p.h, p.last_active] for p in w.panes()]}) out["sessions"].append({"id": s.id, "name": s.name, "current": s.current, "last": s.last, "history": [x.id for x in s.history], "indexes": s.indexes(), "free0": s.free_index(0), "free3": s.free_index(3), "tag": s.tag, "windows": wins}) return out NAMES = ["bash", "vim", "logs", "a:b", "web", "w1", "w2", "é"] def rand_model_ops(deep): ops = [] for si in range(rnd.randint(1, 3)): ops.append(["session", rnd.choice(["main", "work", "s%d" % si, "other"]), rnd.randint(20, 90), rnd.randint(8, 30)]) used = set() for wi in range(rnd.randint(1, 4)): index = rnd.choice([i for i in range(0, 7) if i not in used]) used.add(index) ops.append(["window", rnd.choice(NAMES), index, rnd.random() < 0.5]) for _s in range(rnd.randint(0, 3 if deep else 2)): ops.append(["split", rnd.randint(0, 5), rnd.choice(["lr", "tb"]), rnd.random() < 0.3, rnd.choice([None, None, 5, 10])]) for pi in range(rnd.randint(0, 3)): ops.append(["feed", rnd.randint(0, 5), rnd.choice(["hello world\r\n", "\x1b[31mred\x1b[0m text", "$ ls\r\nfoo bar\r\n", "é世\r\n"])]) if rnd.random() < 0.5: ops.append(["active", rnd.randint(0, 5)]) if rnd.random() < 0.2: ops.append(["zoom", rnd.randint(0, 5)]) if rnd.random() < 0.2: ops.append(["mark", rnd.randint(0, 5)]) for _k in range(rnd.randint(0, 3)): ops.append(["select", rnd.randint(0, 6)]) return ops def common_opts(): return [["opt", "server", "status-right", "R-FIXED"], ["opt", "server", "status-left", "[#{session_name}] "]] RENDER_OPTS = [["status-position", "top"], ["status-justify", "centre"], ["status-justify", "right"], ["status", "off"], ["status-left-length", "4"], ["status-right-length", "3"], ["pane-border-lines", "double"], ["pane-border-lines", "heavy"], ["pane-border-lines", "simple"], ["window-status-format", "#I:#W#{?window_flags,#{window_flags}, }"], ["status-style", "bg=blue,fg=white"], ["message-style", "bg=red,fg=white"], ["window-status-separator", " | "], ["status-left", "#[fg=red]#S#[default] "], ["status-right", "#[bold]#{window_panes} p"], ["window-status-current-style", "bold,bg=cyan"], ["window-status-last-style", "underscore"], ["pane-active-border-style", "fg=blue"]] for n in range(70): ops = common_opts() + rand_model_ops(True) # window-scoped options go after windows exist; server-scoped ones can come any time for _k in range(rnd.randint(0, 3)): name, value = rnd.choice(RENDER_OPTS) ops.append(["opt", "server", name, value]) home = os.path.join(SCRATCH, "home%d" % n) os.makedirs(home, exist_ok=True) model = Model(home) for item in ops: model.op(item) # a few clients on random sessions sess_list = list(model.server.sessions.values()) clients = [] for _c in range(3): sess = rnd.choice(sess_list) conn = tai.MplexClientConn(model.server, None, 0) conn.session = sess conn.attached = True conn.rows, conn.cols = rnd.randint(4, 32), rnd.randint(6, 100) conn.table = "root" spec_c = {"session": sess.id, "rows": conn.rows, "cols": conn.cols} pick = rnd.random() if pick < 0.2: conn.prompt = tai.MplexPrompt("cmd: ", "ls é", lambda v: None) conn.prompt.pos = rnd.randint(0, 4) spec_c["prompt"] = [conn.prompt.label, conn.prompt.text, conn.prompt.pos] elif pick < 0.35: conn.message = ("multi\nline msg " + "x" * rnd.randint(0, 120), time.monotonic() + 1000) spec_c["message"] = conn.message[0] elif pick < 0.45: conn.message = ("expired", time.monotonic() - 1) spec_c["message_expired"] = "expired" elif pick < 0.6 and sess.window() is not None: conn.overlay = tai.MplexPaneNumbers(model.server, sess.window(), time.monotonic() + 1000) spec_c["overlay"] = "numbers" elif pick < 0.7: model.server.flows[1] = {"answer": None, "settled": None, "until": 1} conn.overlay = tai.MplexFlowScreen(model.server, 1, "Title", "Some body text to wrap " * 3, ["yes", "no", "maybe"], 1, 1e12) spec_c["overlay"] = "flow" canvas = model.server.renderer.frame(conn) spec_c["want"] = canvas_json(canvas) clients.append(spec_c) case = {"t": "render", "ops": ops, "clients": clients, "dump": model.dump()} cases.append(case) # ---- I: targets / ctx ----------------------------------------------------------------------------------------------------------- SPECS_S = ["", "main", "work", "ma", "w*", "=main", "$0", "$1", "$9", "nope", "s0", "*", "zz*", "=="] SPECS_W = ["", "0", "1", "2", "3", "7", "bash", "vim", "ba", "v", "w*", "=1", "=0", "==2", "=bash", "!", "{last}", "^", "{start}", "$", "{end}", "{next}", "{previous}", "+", "-", "+2", "-3", "main:", "main:1", "work:bash", "main:+", "work:-", "x:1", ":2", "@0", "@1", "@5", "@99", "main", "work", "a:b", "a", "é", ":", "main:nope", "1.1", "web"] SPECS_P = ["", "0", "1", "2", "5", "%0", "%1", "%3", "%99", "!", "{last}", "+", "-", "+1", "-2", "{next}", "{previous}", "{top}", "{bottom}", "{left}", "{right}", "{top-left}", "{top-right}", "{bottom-left}", "{bottom-right}", "{bogus}", ":.1", ":1.0", ":0.2", "main:0.1", "main:1.+", ".1", ".+", "0.1", "bash", "bash.1", "nope", "@0.1", "@1.0", "@9.0", ":.!"] for n in range(70): ops = common_opts() + rand_model_ops(True) home = os.path.join(SCRATCH, "tgt%d" % n) os.makedirs(home, exist_ok=True) model = Model(home) for item in ops: model.op(item) if rnd.random() < 0.4: extra_op = ["sessopt", rnd.randint(0, 5), "pane-base-index", "1"] ops.append(extra_op) model.op(extra_op) sess_list = list(model.server.sessions.values()) ctx_specs = [] for kind in ("default", "client", "pane", "envpane", "nosess_client"): ctx_specs.append(kind) queries = [] for ck in ctx_specs: conn = None pane = None if ck == "client": conn = tai.MplexClientConn(model.server, None, 0) conn.session = rnd.choice(sess_list) cinfo = {"session": conn.session.id} elif ck == "pane": pane = rnd.choice(list(model.server.panes.values())) cinfo = {"pane": pane.id} elif ck == "envpane": conn = tai.MplexClientConn(model.server, None, 0) conn.env_pane = rnd.choice(list(model.server.panes.keys()) + [999]) cinfo = {"env_pane": conn.env_pane} elif ck == "nosess_client": conn = tai.MplexClientConn(model.server, None, 0) cinfo = {"nosess": True} else: cinfo = {} ctx = tai.MplexCtx(model.server, client=conn, pane=pane) ctx_ids = [ctx.session.id if ctx.session else None, ctx.window.id if ctx.window else None, ctx.pane.id if ctx.pane else None] results = [] for which, pool in (("session", SPECS_S), ("window", SPECS_W), ("pane", SPECS_P)): for spec_text in pool: if rnd.random() < 0.55: continue try: if which == "session": res = ["ok", tai.MplexTargets.session(ctx, spec_text).id] elif which == "window": s, i, w = tai.MplexTargets.window(ctx, spec_text) res = ["ok", s.id, i, w.id] else: s, w, p = tai.MplexTargets.pane(ctx, spec_text) res = ["ok", s.id if s else None, w.id if w else None, p.id if p else None] except tai.MplexError as exc: res = ["err", str(exc)] results.append([which, spec_text, res]) for s in sess_list: for text in ["", "0", "1", "!", "bash", "+", "=1", "{next}", "-2", "9"]: if rnd.random() < 0.7: continue results.append(["windex", [s.id, text], tai.MplexTargets.window_index(s, text)]) queries.append({"ctx": cinfo, "ctx_ids": ctx_ids, "results": results}) cases.append({"t": "targets", "ops": ops, "queries": queries, "dump": model.dump()}) # ---- J: client conn framing --------------------------------------------------------------------------------------------------------- for _ in range(60): conn = tai.MplexClientConn(None, None, 1) frames_in = [] payloads = [] stream = b"" for _f in range(rnd.randint(0, 6)): kind = bytes([rnd.choice(b"HIRAO")]) payload = bytes(rnd.randrange(256) for _b in range(rnd.choice([0, 1, 5, 300]))) stream += kind + struct.pack(">I", len(payload)) + payload if rnd.random() < 0.15: stream += b"X" + struct.pack(">I", tai.MPLEX_FRAME_MAX + 1) + b"abc" chunks = [] pos = 0 while pos < len(stream): step = rnd.randint(1, 120) chunks.append(stream[pos:pos + step]) pos += step results = [] for chunk in chunks: res = conn.frames(chunk) results.append(None if res is None else [[k.hex(), p.hex()] for k, p in res]) if res is None: break sent = tai.MplexClientConn(None, None, 2) sends = [] for _s in range(rnd.randint(0, 4)): kind = bytes([rnd.choice(b"AOMEDXZ")]) payload = bytes(rnd.randrange(256) for _b in range(rnd.choice([0, 3, 100]))) sends.append([kind.hex(), payload.hex(), sent.send(kind, payload)]) cases.append({"t": "frames", "chunks": [c.hex() for c in chunks], "want": results, "sends": sends, "out": bytes(sent.out).hex()}) # oversize out queue conn = tai.MplexClientConn(None, None, 3) conn.out += b"x" * (tai.MPLEX_CLIENT_OUT_MAX + 1) conn.prev = [1] r = conn.send(b"O", b"abc") cases.append({"t": "send_full", "want": [r, conn.prev is None, len(conn.out)]}) with open(OUT, "w", encoding="utf-8") as handle: for case in cases: handle.write(json.dumps(case) + "\n") print("cases", len(cases), collections.Counter(c["t"] for c in cases))