a73x

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)")