test/native_tiling.py
Ref: Size: 14.9 KiB History
#!/usr/bin/env python3
"""Shared real-SDL fixture and focused unslept output-flood acceptance.
The state hook describes geometry; daemon grids, PTY dimensions, and framebuffer
pixels independently establish that input and rendering use that geometry.
"""
import json
import os
from pathlib import Path
import re
import shlex
import shutil
import signal
import socket
import subprocess
import sys
import tempfile
import time
DEADLINE = 5.0
SPLIT_BINDINGS = {"beside": "b", "stacked": "v"}
def require(condition, message):
if not condition:
raise AssertionError(message)
def eventually(probe, message, seconds=DEADLINE):
deadline = time.monotonic() + seconds
while time.monotonic() < deadline:
result = probe()
if result:
return result
time.sleep(0.04)
raise AssertionError(message)
class Rig:
float_size = (960, 600)
float_scale = None
def __init__(self, mux, muxg):
self.mux, self.muxg = str(Path(mux).resolve()), str(Path(muxg).resolve())
self.root = Path(tempfile.mkdtemp(prefix="muxg-tiling-"))
self._owned_root = self.root
self.env = os.environ.copy()
# Preserve the compositor address before isolating runtime state.
self.env["WAYLAND_DISPLAY"] = str(Path(self.env.get("XDG_RUNTIME_DIR", "/tmp")) /
self.env.get("WAYLAND_DISPLAY", "wayland-0"))
for key in ("XDG_STATE_HOME", "XDG_RUNTIME_DIR", "XDG_CONFIG_HOME", "XDG_CACHE_HOME"):
path = self.root / key
path.mkdir(mode=0o700)
self.env[key] = str(path)
self.env["SHELL"] = "/bin/sh"
self.env["SDL_VIDEO_DRIVER"] = os.environ.get("MUXG_VIDEODRIVER", "offscreen")
self.env.pop("SDL_VIDEODRIVER", None)
self.procs, self.logs, self.daemons = [], [], []
self.targets = {}
self.gui = None
self.gui_cwd = None
self.fd = None
self.serial = 0
self.checkpoints = 0
self.test_label = "tiling"
def spawn(self, argv, label, env=None, cwd=None):
log = (self.root / (label + ".log")).open("wb")
self.logs.append(log)
proc = subprocess.Popen(argv, env=self.env if env is None else env,
cwd=cwd, stdout=log, stderr=subprocess.STDOUT)
self.procs.append(proc)
return proc
def command(self, *args, check=True):
return subprocess.run([self.mux, *args], env=self.env, capture_output=True,
text=True, timeout=3, check=check)
def stop_daemon(self, sock, proc):
# `mux d stop` waits for its peer PID to disappear. Reap our child while
# that command waits, otherwise its zombie keeps the PID alive forever.
stop = self.spawn([self.mux, "d", "stop", "--sock", sock], f"stop-{proc.pid}")
def reaped():
daemon_done = proc.poll() is not None
stop_done = stop.poll() is not None
return daemon_done and stop_done
eventually(reaped, "daemon did not stop and get reaped")
require(stop.returncode == 0, "daemon stop command failed")
def daemon(self, label, quic=False):
sock = str(self.root / (label + ".sock"))
env = self.env.copy()
home = self.root / (label + "-state")
home.mkdir()
for key in ("XDG_STATE_HOME", "XDG_RUNTIME_DIR", "XDG_CONFIG_HOME", "XDG_CACHE_HOME"):
path = home / key
path.mkdir(mode=0o700)
env[key] = str(path)
args = [self.mux, "d", "start", "--sock", sock]
if quic:
self.command("d", "keygen")
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as reserve:
reserve.bind(("127.0.0.1", 0))
port = reserve.getsockname()[1]
args += ["--quic", f"127.0.0.1:{port}", "--key",
str(Path(self.env["XDG_CONFIG_HOME"]) / "mux/key")]
self.targets[sock] = f"quic://127.0.0.1:{port}"
proc = self.spawn(args, label, env)
self.daemons.append((sock, proc))
eventually(lambda: Path(sock).exists(), label + " daemon did not start")
return sock, proc
def start_gui(self, first, second, label):
self.catalogue(["--sock " + first, self.targets.get(second, "--sock " + second)])
return self.launch_gui(["--sock", first, "--session", "left"], label)
def launch_gui(self, args, label, attached=True):
require(self.gui is None, "close the previous fixture window before relaunch")
fifo = self.root / (label + ".fifo")
os.mkfifo(fifo)
self.fd = os.open(fifo, os.O_RDWR | os.O_NONBLOCK)
self.env["MUXG_TEST_FIFO"] = str(fifo)
self.gui_log = self.root / (label + ".log")
self.gui = self.spawn([self.muxg, *args], label, cwd=self.gui_cwd)
state = self.state()
if os.environ.get('MUXG_TEST_SWAY_FLOAT') == '1' and (
self.env['SDL_VIDEO_DRIVER'] == 'wayland' or self.float_scale is not None):
require(self.env['SDL_VIDEO_DRIVER'] == 'wayland', 'Sway floating requires Wayland')
def mapped(node):
return node.get('pid') == self.gui.pid or any(
mapped(child) for child in node.get('nodes', []) + node.get('floating_nodes', []))
eventually(lambda: mapped(json.loads(subprocess.check_output(
['swaymsg', '-r', '-t', 'get_tree'], env=self.env, timeout=3))),
'owned test window did not map before floating')
width, height = self.float_size
command = (f'[pid={self.gui.pid}] floating enable, border none, '
f'resize set width {width} px height {height} px')
result = subprocess.run(['swaymsg', '-r', command], env=self.env,
capture_output=True, text=True, check=True, timeout=3)
require(all(item['success'] for item in json.loads(result.stdout)),
'could not float this test window')
state = self.wait_state(lambda s: (s['logical_width'], s['logical_height']) == self.float_size)
if self.float_scale is not None:
require(state['width'] == state['logical_width'] * self.float_scale and
state['height'] == state['logical_height'] * self.float_scale,
'Wayland resize acceptance requires the expected density')
if attached:
state = self.wait_state(lambda s: bool(s["panes"]) and
all(p["phase"] == "attached" for p in s["panes"]))
return state
def send(self, *lines):
require(self.gui.poll() is None, "GUI exited unexpectedly")
data = ("\n".join(lines) + "\n").encode()
require(os.write(self.fd, data) == len(data), "short FIFO write")
def key(self, name):
self.send("key:" + name)
def chord(self, key):
self.send("key:prefix", "key:" + key)
def split(self, direction):
key = SPLIT_BINDINGS.get(direction)
require(key is not None, "unknown split direction: " + str(direction))
self.chord(key)
def open_picker(self, direction=None):
self.chord("enter")
self.picker("hosts")
if direction:
self.key(SPLIT_BINDINGS[direction])
def shell(self, command):
self.send("text:" + command, "key:enter")
def artifact(self, command, suffix):
self.serial += 1
path = self.root / (str(self.serial) + suffix)
self.send(command + ":" + str(path))
eventually(lambda: path.exists() or self.gui.poll() is not None,
"GUI did not produce " + command)
require(path.exists(), "GUI exited while producing " + command)
return path
def state(self):
return json.loads(self.artifact("state", ".json").read_text())
def wait_state(self, predicate, seconds=DEADLINE):
def probe():
state = self.state()
return state if predicate(state) else None
return eventually(probe, "GUI state did not converge", seconds=seconds)
def catalogue(self, targets):
path = Path(self.env["XDG_STATE_HOME"]) / "mux" / "hosts"
path.parent.mkdir(exist_ok=True)
path.write_text("\n".join(targets) + "\n")
def picker(self, level=None):
return self.wait_state(lambda s: s.get("picker") and
(level is None or s["picker"]["level"] == level))
def choose(self, label):
def present():
state = self.state()
picker = state.get("picker")
if not picker:
return None
labels = [row["label"] if isinstance(row, dict) else row for row in picker["rows"]]
matches = [i for i, text in enumerate(labels) if text == label]
return (matches[0], picker["selected"]) if len(matches) == 1 else None
chosen, selected = eventually(present, f"picker row {label!r} not found uniquely", seconds=20)
delta = chosen - selected
for _ in range(abs(delta)):
self.key("down" if delta > 0 else "up")
self.key("enter")
def host(self, spelling, direction=None):
self.open_picker(direction)
self.choose(spelling)
self.picker("sessions")
def new_session(self, name):
self.choose("New session...")
self.picker("session_name")
self.send("text:" + name, "key:enter")
def pixels(self):
raw = self.artifact("capture", ".ppm").read_bytes()
magic, dims, maximum, pixels = raw.split(b"\n", 3)
require(magic == b"P6" and maximum == b"255", "invalid framebuffer capture")
width, height = map(int, dims.split())
require(len(pixels) == width * height * 3, "truncated framebuffer capture")
return width, height, pixels
def dump(self, sock, session):
return self.command("d", "dump", "--sock", sock, "--session", session,
check=False).stdout
def wait_marker(self, sock, session, marker):
eventually(lambda: marker in self.dump(sock, session), marker + " did not reach grid")
def status(self, sock, session):
return json.loads(self.command("a", "status", "--sock", sock,
"--session", session, "--timeout", "2000").stdout)
def frames(self):
before = self.gui_log.read_text().count("\ntotal ")
self.gui.send_signal(signal.SIGUSR1)
def fresh():
text = self.gui_log.read_text()
return text if text.count("\ntotal ") > before else None
report = eventually(fresh, "no fresh completed frame report")
count = int(re.findall(r"timing \((\d+) frames\)", report)[-1])
p99 = int(re.findall(r"^total\s+\d+\s+\d+\s+(\d+)", report, re.M)[-1])
return count, p99
def quit(self):
self.send("quit")
require(self.gui.wait(timeout=3) == 0, "GUI failed to detach cleanly")
os.close(self.fd)
self.fd = None
self.gui = None
def ok(self, message):
self.checkpoints += 1
print(f"{self.test_label} OK ({self.checkpoints}): {message}", flush=True)
def close(self, success=None):
if success is None:
success = sys.exc_info()[0] is None
if self.gui is not None and self.gui.poll() is None:
self.gui.terminate()
for sock, proc in self.daemons:
if proc.poll() is None:
try:
self.stop_daemon(sock, proc)
except (subprocess.TimeoutExpired, AssertionError):
proc.terminate()
for proc in self.procs:
if proc.poll() is None:
proc.terminate()
try:
proc.wait(timeout=3)
except subprocess.TimeoutExpired:
proc.kill()
proc.wait(timeout=3)
if self.fd is not None:
os.close(self.fd)
for log in self.logs:
log.close()
if success and os.environ.get("MUXG_TEST_KEEP_ARTIFACTS") != "1":
require(self.root == self._owned_root and not self.root.is_symlink(),
"refusing to remove a foreign artifact root")
shutil.rmtree(self._owned_root)
else:
print("Native test artifacts retained:", self.root, flush=True)
def rect_contains(rect, x, y):
return rect["x"] <= x < rect["x"] + rect["w"] and rect["y"] <= y < rect["y"] + rect["h"]
def colour_counts(capture, rect, colour):
width, height, pixels = capture
inside = outside = 0
for offset in range(0, len(pixels), 3):
r, g, b = pixels[offset:offset + 3]
if colour == "red":
hit = r > 80 and r > g * 2 and r > b * 2
elif colour == "green":
hit = g > 80 and g > r * 2 and g > b * 2
else:
hit = b > 80 and b > r * 2 and b > g * 2
if hit:
y, x = divmod(offset // 3, width)
if rect_contains(rect, x, y):
inside += 1
else:
outside += 1
return inside, outside
def output_flood(rig, refs):
"""Unslept 256 KiB writes stress fairness independently of the paced journey."""
left, right, _ = refs
# A bounded flood with an external stop signal. During the frame-count
# interval only nonforcing state observations are allowed, never captures.
stop = rig.root / "stop-flood"
progress = rig.root / "flood-progress"
rig.focus(left)
flood = (f"i=0; while [ $i -lt 4096 ] && [ ! -e {shlex.quote(str(stop))} ]; do "
"head -c 262144 /dev/zero | tr '\\000' X; "
f"i=$((i+1)); echo $i > {shlex.quote(str(progress))}; done; echo FLOOD-DONE")
rig.shell(flood)
eventually(lambda: progress.exists() and progress.stat().st_size > 0, "flood did not start")
count1, _ = rig.frames()
rig.focus(right)
started = time.monotonic()
rig.shell("printf '\\033[38;2;0;0;255mRESPONSIVE-%s\\033[0m\\n' RIGHT")
rig.wait_marker(*refs[right], "RESPONSIVE-RIGHT")
# State observes the last completed frame and must not itself request a
# repaint. This detects B staying visually stale while A keeps painting.
rig.wait_state(lambda s: "RESPONSIVE-RIGHT" in next(p for p in s["panes"] if p["id"] == right)["painted_text"])
latency = time.monotonic() - started
time.sleep(0.3)
require("FLOOD-DONE" not in rig.dump(*refs[left]), "flood ended before concurrent check")
count2, p99 = rig.frames()
require(colour_counts(rig.pixels(), next(p for p in rig.state()["panes"] if p["id"] == right)["content"], "blue")[0] > 30,
"new responsive marker was not rendered in the other pane")
stop.touch()
rig.wait_marker(*refs[left], "FLOOD-DONE")
require(count2 > count1, "no frames painted while output flooded")
require(latency < DEADLINE and p99 < 20000, f"flood isolation exceeded budget: {latency}s / {p99}us")
rig.ok(f"flood leaves other pane responsive ({latency*1000:.0f} ms input-to-frame, {p99} us frame p99)")