mplex: port MplexCopyMode with differential test against tai.py and tmux 3.6
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 5.5
@@ -0,0 +1,318 @@
|
||||
#!/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 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, err, closed):
|
||||
last_copy = "-"
|
||||
if server.copied:
|
||||
text, command, append = server.copied[-1]
|
||||
last_copy = "%s/%s/%d" % (esc(text), esc(command), 1 if append else 0)
|
||||
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.copied)), last_copy, str(len(server.messages)),
|
||||
esc(server.messages[-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),
|
||||
])
|
||||
|
||||
|
||||
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 = test_tai.MplexStubServer()
|
||||
server.options.set("mode-keys", keys)
|
||||
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"", None)
|
||||
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, err, closed))
|
||||
if closed:
|
||||
break
|
||||
if expect is not None:
|
||||
rec.add("T", expect[0], esc(expect[1]))
|
||||
rec.add("X")
|
||||
|
||||
|
||||
# ---- 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)])
|
||||
|
||||
# ---- 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)))
|
||||
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")
|
||||
sys.stderr.write("scenarios=%d steps=%d\n" % (scenarios, steps_total))
|
||||
Executable
+27
@@ -0,0 +1,27 @@
|
||||
#!/bin/sh
|
||||
# Build and run the MplexCopyMode differential test. Writes only under taicpp/build/f4.
|
||||
# usage: tests/mplex_copy_run.sh [fuzz_count] [seed] (SAN=1 builds with ASan and UBSan)
|
||||
set -e
|
||||
here=$(cd "$(dirname "$0")" && pwd)
|
||||
tc=$(cd "$here/.." && pwd)
|
||||
repo=$(cd "$tc/.." && pwd)
|
||||
out="$tc/build/f4"
|
||||
mkdir -p "$out"
|
||||
srcs=""
|
||||
for f in "$tc"/src/mplex/*.cpp; do
|
||||
case "$f" in
|
||||
*/mplex_server.cpp|*/mplex_commands.cpp|*/mplex_client.cpp) continue ;;
|
||||
esac
|
||||
srcs="$srcs $f"
|
||||
done
|
||||
flags="-O2"
|
||||
bin="$out/mplex_copy_test"
|
||||
if [ -n "$SAN" ]; then
|
||||
flags="-g -O1 -fsanitize=address,undefined"
|
||||
bin="$out/mplex_copy_test_san"
|
||||
fi
|
||||
# The test supplies its own MplexServer stub and the MplexCommands destructor it needs.
|
||||
g++ -std=c++23 -Wall -Wextra $flags -DF4_STUB_COMMANDS -I"$tc/src/mplex" "$tc/tests/mplex_copy_test.cpp" $srcs \
|
||||
-Wl,--unresolved-symbols=ignore-in-object-files -o "$bin"
|
||||
python3 -I "$tc/tests/mplex_copy_gen.py" "$repo" "$out/scenarios.txt" "${1:-3600}" "${2:-20251025}"
|
||||
"$bin" "$out/scenarios.txt"
|
||||
@@ -0,0 +1,328 @@
|
||||
// Differential test of MplexCopyMode against tai.py's MplexCopyMode. Scenario file made by
|
||||
// mplex_copy_gen.py (tmux 3.6 fixtures from test_tai.py, every default copy-mode binding, seeded fuzz);
|
||||
// every step's full state line (cursor, scroll, selection, mark, search, copies, messages, render checksum)
|
||||
// must match the Python oracle byte for byte. Fixture scenarios also carry the value tmux 3.6 produced.
|
||||
//
|
||||
// The MplexServer here is the small stub the Python tests use (MplexStubServer): options, bindings, a copy sink.
|
||||
// Build and run: tests/mplex_copy_run.sh (writes only under taicpp/build/).
|
||||
#include <algorithm>
|
||||
#include <cstdio>
|
||||
#include <cstdlib>
|
||||
#include <fstream>
|
||||
#include <iostream>
|
||||
#include <sstream>
|
||||
#include <string>
|
||||
#include <tuple>
|
||||
#include <vector>
|
||||
|
||||
#include "mplex_args.h"
|
||||
#include "mplex_canvas.h"
|
||||
#include "mplex_commands.h"
|
||||
#include "mplex_config.h"
|
||||
#include "mplex_copymode.h"
|
||||
#include "mplex_renderer.h"
|
||||
#include "mplex_server.h"
|
||||
|
||||
using namespace detz::mplex;
|
||||
|
||||
namespace {
|
||||
std::vector<std::tuple<std::string, std::string, bool>> g_copied;
|
||||
std::vector<std::string> g_messages;
|
||||
} // namespace
|
||||
|
||||
#ifdef F4_STUB_COMMANDS
|
||||
struct detz::mplex::MplexCommands::Impl {};
|
||||
detz::mplex::MplexCommands::~MplexCommands() = default;
|
||||
#endif
|
||||
|
||||
namespace detz::mplex {
|
||||
|
||||
MplexServer::MplexServer(const std::string& home_in) : home(home_in) {
|
||||
config = std::make_shared<MplexConfig>();
|
||||
options = &config->options;
|
||||
started = 0.0;
|
||||
}
|
||||
MplexServer::~MplexServer() = default;
|
||||
|
||||
MplexOptions* MplexServer::pane_options(const std::shared_ptr<MplexPane>&) { return options; }
|
||||
void MplexServer::message(const std::shared_ptr<MplexClientConn>&, const std::string& text, std::optional<long>) {
|
||||
g_messages.push_back(text);
|
||||
}
|
||||
void MplexServer::mark_dirty(const std::shared_ptr<MplexWindow>&) {}
|
||||
void MplexServer::store_copy(const std::string& text, const std::string& command, bool append) {
|
||||
g_copied.emplace_back(text, command, append);
|
||||
}
|
||||
void MplexServer::run_binding(const std::shared_ptr<MplexClientConn>&, const MplexBinding& binding,
|
||||
const std::shared_ptr<MplexPane>& pane) {
|
||||
auto words = MplexCommandParser::parse(binding.command)[0];
|
||||
auto mode = std::dynamic_pointer_cast<MplexCopyMode>(pane->top_mode());
|
||||
mode->command(words[2], std::vector<std::string>(words.begin() + 3, words.end()), nullptr);
|
||||
}
|
||||
|
||||
} // namespace detz::mplex
|
||||
|
||||
namespace {
|
||||
|
||||
std::string esc(const std::string& text) {
|
||||
std::string out;
|
||||
for (char ch : text) {
|
||||
if (ch == '%') out += "%25";
|
||||
else if (ch == '\t') out += "%09";
|
||||
else if (ch == '\n') out += "%0A";
|
||||
else if (ch == '\r') out += "%0D";
|
||||
else out += ch;
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
std::string unesc(const std::string& text) {
|
||||
std::string out;
|
||||
for (size_t i = 0; i < text.size(); ++i) {
|
||||
if (text[i] == '%' && i + 2 < text.size()) {
|
||||
out += static_cast<char>(std::stoi(text.substr(i + 1, 2), nullptr, 16));
|
||||
i += 2;
|
||||
} else {
|
||||
out += text[i];
|
||||
}
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
std::vector<std::string> split_tabs(const std::string& line) {
|
||||
std::vector<std::string> out;
|
||||
size_t start = 0;
|
||||
while (true) {
|
||||
size_t at = line.find('\t', start);
|
||||
if (at == std::string::npos) {
|
||||
out.push_back(line.substr(start));
|
||||
break;
|
||||
}
|
||||
out.push_back(line.substr(start, at - start));
|
||||
start = at + 1;
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
uint32_t crc32(const std::string& data) {
|
||||
static uint32_t table[256];
|
||||
static bool ready = false;
|
||||
if (!ready) {
|
||||
for (uint32_t n = 0; n < 256; ++n) {
|
||||
uint32_t c = n;
|
||||
for (int k = 0; k < 8; ++k) c = (c & 1) ? 0xEDB88320u ^ (c >> 1) : c >> 1;
|
||||
table[n] = c;
|
||||
}
|
||||
ready = true;
|
||||
}
|
||||
uint32_t crc = 0xFFFFFFFFu;
|
||||
for (unsigned char byte : data) crc = table[(crc ^ byte) & 0xFF] ^ (crc >> 8);
|
||||
return crc ^ 0xFFFFFFFFu;
|
||||
}
|
||||
|
||||
std::string hex8(uint32_t value) {
|
||||
char buf[16];
|
||||
std::snprintf(buf, sizeof buf, "%08x", value);
|
||||
return buf;
|
||||
}
|
||||
|
||||
std::string digest(MplexCopyMode& mode) {
|
||||
alignas(16) static char scratch[sizeof(MplexClientConn)]; // render() ignores its client
|
||||
ModeBlock block = mode.render(*reinterpret_cast<MplexClientConn*>(scratch));
|
||||
std::string parts;
|
||||
for (const Row& row : block.grid) {
|
||||
for (const Cell& cell : row) {
|
||||
parts += to_utf8(cell.ch);
|
||||
parts += "|" + color_sgr(cell.fg) + "|" + color_sgr(cell.bg) + "|" + std::to_string(cell.attr) + ";";
|
||||
}
|
||||
parts += "\n";
|
||||
}
|
||||
if (block.cursor) parts += "(" + std::to_string(block.cursor->first) + ", " + std::to_string(block.cursor->second) + ")";
|
||||
else parts += "None";
|
||||
return hex8(crc32(parts));
|
||||
}
|
||||
|
||||
std::string pair_text(const std::optional<MplexCopyMode::Point>& value) {
|
||||
if (!value) return "-";
|
||||
return std::to_string(value->first) + "," + std::to_string(value->second);
|
||||
}
|
||||
|
||||
std::string search_text(const std::optional<MplexPane::SearchState>& value) {
|
||||
if (!value) return "-";
|
||||
return esc(value->text) + "," + (value->forward ? "1" : "0") + "," + (value->literal ? "1" : "0");
|
||||
}
|
||||
|
||||
std::string state_line(MplexCopyMode& mode, const std::shared_ptr<MplexPane>& pane, const std::string& err, bool closed) {
|
||||
std::string last_copy = "-";
|
||||
if (!g_copied.empty()) {
|
||||
const auto& [text, command, append] = g_copied.back();
|
||||
last_copy = esc(text) + "/" + esc(command) + "/" + (append ? "1" : "0");
|
||||
}
|
||||
std::vector<std::string> f = {
|
||||
"E", std::to_string(mode.cx), std::to_string(mode.cy), std::to_string(mode.oy), std::to_string(mode.hist()),
|
||||
pair_text(mode.anchor), pair_text(mode.mark), mode.rect ? "1" : "0", mode.line_select ? "1" : "0",
|
||||
search_text(mode.search), search_text(pane->last_search), mode.show_position ? "1" : "0", closed ? "1" : "0",
|
||||
std::to_string(g_copied.size()), last_copy, std::to_string(g_messages.size()),
|
||||
g_messages.empty() ? "-" : esc(g_messages.back()), std::to_string(mode.count), std::to_string(mode.lastcx),
|
||||
std::to_string(mode.lastsx), esc(err), digest(mode), std::to_string(pane->history_size()),
|
||||
std::to_string(pane->prefix_count)};
|
||||
std::string out;
|
||||
for (size_t i = 0; i < f.size(); ++i) out += (i ? "\t" : "") + f[i];
|
||||
return out;
|
||||
}
|
||||
|
||||
int failures = 0;
|
||||
int shown = 0;
|
||||
long checked_steps = 0;
|
||||
long checked_scenarios = 0;
|
||||
long tmux_checks = 0;
|
||||
long live_total = 0;
|
||||
long live_diff = 0;
|
||||
|
||||
void fail(const std::string& scenario, const std::string& what, const std::string& want, const std::string& got) {
|
||||
++failures;
|
||||
if (shown < 25) {
|
||||
++shown;
|
||||
std::cerr << "MISMATCH " << scenario << " " << what << "\n python: " << want << "\n c++ : " << got << "\n";
|
||||
}
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
int main(int argc, char** argv) {
|
||||
if (argc < 2) {
|
||||
std::cerr << "usage: mplex_copy_test SCENARIOS\n";
|
||||
return 2;
|
||||
}
|
||||
std::ifstream in(argv[1]);
|
||||
std::vector<std::string> lines;
|
||||
for (std::string line; std::getline(in, line);) lines.push_back(line);
|
||||
size_t at = 0;
|
||||
// command table
|
||||
{
|
||||
auto fields = split_tabs(lines.at(at++));
|
||||
std::vector<std::string> want(fields.begin() + 1, fields.end());
|
||||
if (want != MplexCopyMode::command_names()) {
|
||||
std::cerr << "command_names() differs from the Python c_* methods\n";
|
||||
++failures;
|
||||
}
|
||||
}
|
||||
while (at < lines.size()) {
|
||||
auto s = split_tabs(lines[at++]);
|
||||
if (s.empty() || s[0] != "S") {
|
||||
std::cerr << "scenario file out of step at line " << at << "\n";
|
||||
return 2;
|
||||
}
|
||||
std::string name = unesc(s[1]);
|
||||
int rows = std::stoi(s[2]), cols = std::stoi(s[3]), history = std::stoi(s[4]);
|
||||
std::string keys = s[5];
|
||||
std::string content = unesc(s[6]);
|
||||
bool view = s[7] == "1";
|
||||
bool exit_at_bottom = s[8] == "1";
|
||||
g_copied.clear();
|
||||
g_messages.clear();
|
||||
MplexServer server("/tmp");
|
||||
server.options->set("mode-keys", keys);
|
||||
auto pane = std::make_shared<MplexPane>(1, cols, rows, history, "/tmp");
|
||||
std::vector<std::string> view_lines;
|
||||
auto replace_nl = [](const std::string& text) {
|
||||
std::string out;
|
||||
for (char ch : text) {
|
||||
if (ch == '\n') out += "\r\n";
|
||||
else out += ch;
|
||||
}
|
||||
return out;
|
||||
};
|
||||
std::shared_ptr<MplexCopyMode> mode;
|
||||
if (view) {
|
||||
size_t start = 0;
|
||||
while (true) {
|
||||
size_t nl = content.find('\n', start);
|
||||
if (nl == std::string::npos) {
|
||||
view_lines.push_back(content.substr(start));
|
||||
break;
|
||||
}
|
||||
view_lines.push_back(content.substr(start, nl - start));
|
||||
start = nl + 1;
|
||||
}
|
||||
mode = std::make_shared<MplexCopyMode>(&server, pane, &view_lines);
|
||||
} else {
|
||||
pane->feed(replace_nl(content));
|
||||
mode = std::make_shared<MplexCopyMode>(&server, pane, nullptr, false, exit_at_bottom);
|
||||
}
|
||||
pane->push_mode(mode);
|
||||
++checked_scenarios;
|
||||
{
|
||||
auto e0 = split_tabs(lines[at++]);
|
||||
std::string got = std::to_string(mode->cx) + " " + std::to_string(mode->cy) + " " + std::to_string(mode->oy) + " " +
|
||||
std::to_string(mode->hist()) + " " + mode->name + " " + digest(*mode);
|
||||
std::string want = e0[1] + " " + e0[2] + " " + e0[3] + " " + e0[4] + " " + e0[5] + " " + e0[6];
|
||||
if (got != want) fail(name, "initial", want, got);
|
||||
}
|
||||
std::string last_state_cursor;
|
||||
std::vector<std::string> tmux_line;
|
||||
while (lines[at] != "X") {
|
||||
auto step = split_tabs(lines[at++]);
|
||||
std::string err;
|
||||
if (step[0] == "T") {
|
||||
tmux_line = step;
|
||||
continue;
|
||||
}
|
||||
if (step[0] == "C") {
|
||||
int count = std::stoi(step[1]);
|
||||
std::vector<std::string> words;
|
||||
for (size_t i = 2; i < step.size(); ++i) words.push_back(unesc(step[i]));
|
||||
pane->prefix_count = count;
|
||||
try {
|
||||
mode->command(words[0], std::vector<std::string>(words.begin() + 1, words.end()), nullptr);
|
||||
} catch (const MplexError& exc) {
|
||||
err = std::string("M:") + exc.what();
|
||||
} catch (const std::exception& exc) {
|
||||
err = std::string("X:") + exc.what();
|
||||
}
|
||||
} else if (step[0] == "K") {
|
||||
auto conn = std::make_shared<MplexClientConn>(&server, -1, 1);
|
||||
try {
|
||||
mode->key(unesc(step[1]), Bytes(), *conn);
|
||||
} catch (const MplexError& exc) {
|
||||
err = std::string("M:") + exc.what();
|
||||
} catch (const std::exception& exc) {
|
||||
err = std::string("X:") + exc.what();
|
||||
}
|
||||
} else if (step[0] == "F") {
|
||||
pane->feed(replace_nl(unesc(step[1])));
|
||||
} else if (step[0] == "R") {
|
||||
pane->place(0, 0, std::stoi(step[2]), std::stoi(step[1]));
|
||||
} else {
|
||||
std::cerr << "bad step line " << lines[at - 1] << "\n";
|
||||
return 2;
|
||||
}
|
||||
bool closed = std::find(pane->modes.begin(), pane->modes.end(), mode) == pane->modes.end();
|
||||
std::string got = state_line(*mode, pane, err, closed);
|
||||
std::string want = lines[at++];
|
||||
++checked_steps;
|
||||
if (got != want) fail(name, "after step [" + lines[at - 2] + "]", want, got);
|
||||
last_state_cursor = std::to_string(mode->cx) + " " + std::to_string(mode->cy) + " " + std::to_string(mode->oy);
|
||||
}
|
||||
++at; // "X"
|
||||
if (!tmux_line.empty() && tmux_line[1] == "livecursor") {
|
||||
++live_total;
|
||||
if (last_state_cursor != tmux_line[2]) {
|
||||
++live_diff;
|
||||
if (live_diff <= 12)
|
||||
std::cout << "tmux live differs: " << name << " tmux=" << tmux_line[2] << " mplex=" << last_state_cursor << "\n";
|
||||
}
|
||||
} else if (!tmux_line.empty()) {
|
||||
auto& t = tmux_line;
|
||||
++tmux_checks;
|
||||
if (t[1] == "cursor") {
|
||||
if (last_state_cursor != t[2]) fail(name, "tmux cursor", t[2], last_state_cursor);
|
||||
} else {
|
||||
std::string got = g_copied.empty() ? "" : std::get<0>(g_copied.back());
|
||||
if (got != unesc(t[2])) fail(name, "tmux copied text", unesc(t[2]), got);
|
||||
}
|
||||
}
|
||||
}
|
||||
std::cout << "scenarios=" << checked_scenarios << " steps=" << checked_steps << " tmux_fixture_checks=" << tmux_checks
|
||||
<< " failures=" << failures << " tmux_live=" << live_total << " tmux_live_differ=" << live_diff << "\n";
|
||||
return failures ? 1 : 0;
|
||||
}
|
||||
Reference in New Issue
Block a user