699 lines
30 KiB
Python
699 lines
30 KiB
Python
#!/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))
|