#!/usr/bin/env python3

"""Run swm's unit and headless integration tests."""

from __future__ import annotations

import glob

import os

import re

import shlex

import shutil

import signal

import subprocess

import sys

import time

from pathlib import Path

SCRIPT = Path(__file__).resolve()

TESTDIR = SCRIPT.parent

ROOT = TESTDIR.parent

BUILD = Path(os.environ.get("SWM_TEST_BUILD", TESTDIR / ".build")).resolve()

SOURCE = BUILD / "source"

BIN = BUILD / "bin"

PROFILES = BUILD / "profiles"

LOGS = BUILD / "logs"

CHILDREN: list[subprocess.Popen] = []

CURRENT_COMMAND: list[str] = []

CURRENT_LOG: Path | None = None

class Failure(RuntimeError):

    """Report a test failure without an internal traceback."""

def command_text(command: list[str]) -> str:

    """Format a command for diagnostics."""

    return shlex.join(map(str, command))

def run(*args: object, env: dict[str, str] | None = None) -> str:

    """Run a command and return its standard output."""

    global CURRENT_COMMAND

    command = [str(arg) for arg in args]

    CURRENT_COMMAND = command

    result = subprocess.run(command, text=True, capture_output=True, env=env)

    if result.returncode:

        output = result.stdout + result.stderr

        raise Failure(

            f"command: {command_text(command)}\nstatus: {result.returncode}\n{output}"

        )

    return result.stdout

def spawn(log: Path, *args: object, env: dict[str, str] | None = None) -> subprocess.Popen:

    """Start and track a background command."""

    global CURRENT_COMMAND, CURRENT_LOG

    command = [str(arg) for arg in args]

    CURRENT_COMMAND = command

    CURRENT_LOG = log

    log.parent.mkdir(parents=True, exist_ok=True)

    stream = log.open("w")

    process = subprocess.Popen(command, stdout=stream, stderr=subprocess.STDOUT, env=env)

    stream.close()

    CHILDREN.append(process)

    return process

def terminate(process: subprocess.Popen) -> None:

    """Terminate one tracked process."""

    if process.poll() is None:

        process.terminate()

        try:

            process.wait(timeout=1)

        except subprocess.TimeoutExpired:

            process.kill()

            process.wait()

    if process in CHILDREN:

        CHILDREN.remove(process)

def cleanup_processes() -> None:

    """Terminate all tracked processes in reverse order."""

    for process in reversed(CHILDREN[:]):

        terminate(process)

def tail(path: Path, lines: int = 40) -> str:

    """Return the tail of a diagnostic file."""

    if not path.exists():

        return ""

    return "\n".join(path.read_text(errors="replace").splitlines()[-lines:])

def eventually(description: str, predicate, timeout: float = 2.0):

    """Wait until a predicate returns a truthy value."""

    deadline = time.monotonic() + timeout

    last: Exception | None = None

    while time.monotonic() < deadline:

        try:

            value = predicate()

            if value:

                return value

        except (OSError, ValueError, IndexError) as error:

            last = error

        time.sleep(0.02)

    detail = f": {last}" if last else ""

    raise Failure(f"timeout waiting for {description}{detail}")

def consistently(description: str, predicate, duration: float = 0.5):

    """Fail if a predicate stops holding within a duration."""

    deadline = time.monotonic() + duration

    while time.monotonic() < deadline:

        if not predicate():

            raise Failure(f"{description} did not hold")

        time.sleep(0.02)

def test(name: str, body) -> None:

    """Run and report one named test."""

    global CURRENT_COMMAND, CURRENT_LOG

    CURRENT_COMMAND = []

    CURRENT_LOG = None

    progress = os.environ.get("SWM_TEST_PROGRESS")

    print(f"{name} ... ", end="", flush=True)

    if progress:

        with open(progress, "a") as stream:

            stream.write(f"{name} ... ")

    try:

        body()

    except Exception:

        print("FAIL", flush=True)

        if progress:

            with open(progress, "a") as stream:

                stream.write("FAIL\n")

        raise

    print("ok", flush=True)

    if progress:

        with open(progress, "a") as stream:

            stream.write("ok\n")

def generate_python_protocols(pkgconfig_executable: str) -> None:

    """Generate Python client protocol bindings."""

    wlproto = Path(

        run(pkgconfig_executable, "--variable=pkgdatadir", "wayland-protocols").strip()

    )

    xmls = [

        Path("/usr/share/wayland/wayland.xml"),

        wlproto / "stable/xdg-shell/xdg-shell.xml",

        wlproto / "staging/ext-workspace/ext-workspace-v1.xml",

        wlproto / "staging/ext-session-lock/ext-session-lock-v1.xml",

        wlproto / "unstable/idle-inhibit/idle-inhibit-unstable-v1.xml",

        wlproto

        / "unstable/keyboard-shortcuts-inhibit/keyboard-shortcuts-inhibit-unstable-v1.xml",

        wlproto / "unstable/pointer-constraints/pointer-constraints-unstable-v1.xml",

        wlproto / "unstable/relative-pointer/relative-pointer-unstable-v1.xml",

        wlproto / "unstable/text-input/text-input-unstable-v3.xml",

        ROOT / "protocols/input-method-unstable-v2.xml",

        ROOT / "protocols/virtual-keyboard-unstable-v1.xml",

        ROOT / "protocols/wlr-foreign-toplevel-management-unstable-v1.xml",

        ROOT / "protocols/wlr-layer-shell-unstable-v1.xml",

        ROOT / "protocols/wlr-output-power-management-unstable-v1.xml",

        ROOT / "protocols/wlr-virtual-pointer-unstable-v1.xml",

    ]

    package = BUILD / "python" / "protocols"

    shutil.rmtree(package, ignore_errors=True)

    package.mkdir(parents=True)

    (package / "__init__.py").touch()

    run("pywayland-scanner", "-o", package, "-i", *xmls)

    # pywayland 0.4 generates a circular import for ext-workspace-v1. The

    # group events refer to objects that already exist, so they do not need an

    # interface constructor. Dropping that annotation breaks the cycle while

    # preserving the wire signature.

    workspace_group = package / "ext_workspace_v1/ext_workspace_group_handle_v1.py"

    generated = workspace_group.read_text()

    generated = generated.replace(

        "from .ext_workspace_handle_v1 import ExtWorkspaceHandleV1\n", ""

    ).replace(

        "Argument(ArgumentType.Object, interface=ExtWorkspaceHandleV1)",

        "Argument(ArgumentType.Object)",

    )

    workspace_group.write_text(generated)

def prepare_suite(unit: bool, integration: bool) -> None:

    """Prepare runtime files for the selected tests."""

    binaries = [

        BIN / name

        for name, selected in (("unit", unit), ("swm", integration), ("swmctl", integration))

        if selected

    ]

    missing = [binary for binary in binaries if not binary.exists()]

    if missing:

        raise Failure(f"test binary not built: {missing[0]}; run tests through make")

    for directory in (PROFILES, LOGS):

        shutil.rmtree(directory, ignore_errors=True)

        directory.mkdir(parents=True)

    if unit:

        stubs = BUILD / "stubs"

        shutil.rmtree(stubs, ignore_errors=True)

        stubs.mkdir()

        stub_commands = (

            "dbus-update-activation-environment",

            "foot",

            "waybar",

            "swaybg",

            "sh",

            "wl-paste",

        )

        for name in stub_commands:

            (stubs / name).symlink_to("/bin/true")

        notification = stubs / "notify-send"

        notification.write_text(

            "#!/bin/sh\nprintf '%s\\n' \"$@\" > \"$SWM_TEST_DIR/notification\"\n"

        )

        notification.chmod(0o755)

    if integration:

        for executable in ("pywayland-scanner", "timeout"):

            if not shutil.which(executable):

                raise Failure(f"required tool not found: {executable}")

        generate_python_protocols(os.environ.get("PKG_CONFIG", "pkg-config"))

        source_config = (SOURCE / "config.h").read_text()

        if re.search(r"#define\s+(?:MODKEY|MOD)\s+WLR_MODIFIER_ALT", source_config):

            modkey = 8

        elif re.search(r"#define\s+(?:MODKEY|MOD)\s+WLR_MODIFIER_LOGO", source_config):

            modkey = 64

        else:

            raise Failure("unsupported modifier in config.def.h")

        (BUILD / "modkey").write_text(str(modkey))

def run_unit() -> None:

    """Run every compositor unit test in an isolated process."""

    runtime = BUILD / "unit-runtime"

    shutil.rmtree(runtime, ignore_errors=True)

    runtime.mkdir()

    env = os.environ.copy()

    env["LLVM_PROFILE_FILE"] = str(PROFILES / "unit-%m-%p.profraw")

    env["SWM_TEST_STUB_PATH"] = str(BUILD / "stubs")

    env["SWM_TEST_DIR"] = str(runtime)

    for name in run(BIN / "unit", "--list", env=env).splitlines():

        test(name, lambda name=name: run(BIN / "unit", name, env=env))

    shutil.rmtree(runtime)

def current_title() -> str:

    """Return the compositor's published title."""

    path = Path(os.environ["XDG_RUNTIME_DIR"]) / "swm-title"

    return path.read_text().rstrip("\n") if path.exists() else ""

def wait_title(title: str) -> None:

    """Wait for the compositor to publish a title."""

    eventually(f"title {title!r}", lambda: current_title() == title)

def client_command(role: str, *args: object) -> list[str]:

    """Return a Python protocol-client command."""

    return [sys.executable, str(TESTDIR / "clients.py"), role, *map(str, args)]

def client_log(name: str) -> Path:

    """Return a fixture client's log path."""

    return Path(os.environ["SWM_TEST_LOG_DIR"]) / f"{name}.log"

def start_client(name: str, color: str, fullscreen: bool = False) -> subprocess.Popen:

    """Start an XDG client and wait for its geometry report."""

    report = Path(os.environ["SWM_TEST_DIR"]) / f"{name}.geometry"

    env = os.environ.copy()

    env["SWM_CLIENT_REPORT"] = str(report)

    args: list[object] = [color, name]

    if fullscreen:

        args.append("fullscreen")

    process = spawn(client_log(name), *client_command("xdg", *args), env=env)

    wait_title(name)

    eventually(f"report {report}", lambda: report.exists() and report.stat().st_size)

    return process

def geometry(name: str) -> list[int]:

    """Return a client's reported geometry."""

    path = Path(os.environ["SWM_TEST_DIR"]) / f"{name}.geometry"

    return list(map(int, path.read_text().split()))

def published_geometry(title: str) -> list[int]:

    """Return a client's compositor-published outer geometry."""

    path = Path(os.environ["XDG_RUNTIME_DIR"]) / "swm-windows"

    match = re.search(

        rf"^(\d+),(\d+) (\d+)x(\d+) {re.escape(title)}$", path.read_text(), re.MULTILINE

    )

    if not match:

        raise Failure(f"missing published geometry for {title}")

    return list(map(int, match.groups()))

def protocol(role: str, *args: object) -> str:

    """Run one native-protocol action implemented in Python."""

    return run(*client_command(role, *args), env=os.environ.copy())

def virtual_keyboard(key: object, modifiers: int = 0) -> None:

    """Send one configured key combination."""

    configured = int((BUILD / "modkey").read_text()) | modifiers

    protocol("keyboard", key, configured)

def held_pointer_click(name: str, x: int, button: str | None = None) -> None:

    """Click while the compositor modifier is held by another client."""

    ready = Path(os.environ["SWM_TEST_DIR"]) / f"{name}.ready"

    release = Path(os.environ["SWM_TEST_DIR"]) / f"{name}.release"

    env = os.environ.copy()

    env["SWM_KEYBOARD_READY"] = str(ready)

    env["SWM_KEYBOARD_RELEASE"] = str(release)

    configured = int((BUILD / "modkey").read_text())

    keyboard = spawn(

        client_log(name),

        *client_command("keyboard", "none", configured),

        env=env,

    )

    eventually("held keyboard readiness", lambda: ready.exists())

    args: list[object] = [x]

    if button:

        args.append(button)

    protocol("pointer", *args)

    release.write_text("release\n")

    eventually("held keyboard exit", lambda: keyboard.poll() is not None)

    if keyboard in CHILDREN:

        CHILDREN.remove(keyboard)

def integration_session() -> None:

    """Exercise compositor behavior in a one-output session."""

    one = start_client("one", "ff5588dd")

    two = start_client("two", "ff55aa88")

    def publication() -> None:

        windows = (Path(os.environ["XDG_RUNTIME_DIR"]) / "swm-windows").read_text()

        if not re.search(r"^\d+,\d+ \d+x\d+ one$", windows, re.MULTILINE):

            raise Failure("missing published rectangle for one")

        if not re.search(r"^\d+,\d+ \d+x\d+ two$", windows, re.MULTILINE):

            raise Failure("missing published rectangle for two")

        before = current_title()

        virtual_keyboard(36)

        wait_title("two" if before == "one" else "one")

    test("window publication and focus", publication)

    def client_resize() -> None:

        report = Path(os.environ["SWM_TEST_DIR"]) / "client-resize.geometry"

        trigger = Path(os.environ["SWM_TEST_DIR"]) / "client-resize.trigger"

        committed = Path(os.environ["SWM_TEST_DIR"]) / "client-resize.committed"

        env = os.environ.copy()

        env["SWM_CLIENT_REPORT"] = str(report)

        env["SWM_RESIZE_TRIGGER"] = str(trigger)

        env["SWM_RESIZE_COMMITTED"] = str(committed)

        targets = ([200, 120], [520, 320])

        client = spawn(

            client_log("client-resize"),

            *client_command("xdg-resize", *targets[0], *targets[1]),

            env=env,

        )

        wait_title("client-resize")

        eventually("client resize report", lambda: report.exists() and report.stat().st_size)

        def border_size():

            configured = geometry("client-resize")

            outer = published_geometry("client-resize")

            delta = [outer[2] - configured[0], outer[3] - configured[1]]

            return delta if delta[0] == delta[1] and delta[0] >= 0 else None

        border = eventually("settled floating geometry", border_size)

        for index, target in enumerate(targets):

            trigger.write_text(f"{index}\n")

            eventually(

                f"client resize commit {index}",

                lambda index=index: committed.exists()

                and committed.read_text().strip() == str(index),

            )

            expected = [target[0] + border[0], target[1] + border[1]]

            eventually(

                f"client-selected floating size {index}",

                lambda expected=expected: published_geometry("client-resize")[2:] == expected,

            )

            consistently(

                f"client-selected floating size {index}",

                lambda expected=expected: published_geometry("client-resize")[2:] == expected,

            )

        terminate(client)

    test("client-selected floating resize", client_resize)

    def window_control() -> None:

        target = current_title()

        virtual_keyboard(3, 1)

        eventually("focus after moving switch target", lambda: current_title() != target)

        listed = run(BIN / "swmctl", "window", "list").splitlines()

        matches = [line for line in listed if line.split("\t", 1)[0] == target]

        if len(matches) != 1 or len(matches[0].split("\t")) != 3:

            raise Failure(f"unexpected toplevel list: {listed!r}")

        identifier = matches[0].rsplit("\t", 1)[1]

        duplicate = start_client(target, "5588ffdd")

        listed = run(BIN / "swmctl", "window", "list").splitlines()

        matches = [line for line in listed if line.split("\t", 1)[0] == target]

        if len(matches) != 2:

            raise Failure(f"duplicate title missing from toplevel list: {listed!r}")

        run(BIN / "swmctl", "window", "activate", identifier)

        wait_title(target)

        selected = run(BIN / "swmctl", "workspace", "get", 2, "selected")

        if selected != "true\n":

            raise Failure("identifier activation selected the duplicate window")

        terminate(duplicate)

        virtual_keyboard(2, 1)

        virtual_keyboard(2)

    test("window control", window_control)

    def layouts() -> None:

        three = start_client("three-layout", "ffcc8855")

        names = ("one", "two", "three-layout")

        normal = {name: geometry(name)[:2] for name in names}

        virtual_keyboard(33)

        eventually(

            "max-stack geometry",

            lambda: len({tuple(geometry(name)[:2]) for name in names}) == 1,

        )

        focused = current_title()

        others = tuple(name for name in names if name != focused)

        maximized = geometry(focused)[:2]

        held_pointer_click("max-stack-resize", 100, "right")

        eventually(

            "normal layout restored",

            lambda: all(geometry(name)[:2] != maximized for name in others),

        )

        eventually(

            "max-stack resize persisted",

            lambda: geometry(focused)[:2] != normal[focused],

        )

        for _ in names:

            if current_title() == focused:

                break

            virtual_keyboard(36)

        wait_title(focused)

        virtual_keyboard(20)

        fullscreen_target = focused

        virtual_keyboard(33, 1)

        eventually("fullscreen enable", lambda: geometry(fullscreen_target)[2] == 1)

        virtual_keyboard(33, 1)

        eventually("fullscreen disable", lambda: geometry(fullscreen_target)[2] == 0)

        terminate(three)

        eventually("third layout window removed", lambda: current_title() != "three-layout")

    test("layouts and fullscreen", layouts)

    def workspaces() -> None:

        switcher = start_client("workspace-switcher", "8855ffdd")

        virtual_keyboard(3, 1)

        protocol("pointer", 100)

        pointer_target = current_title()

        virtual_keyboard(36)

        active_target = current_title()

        if active_target == pointer_target:

            raise Failure("could not select the second workspace tile")

        virtual_keyboard(3)

        wait_title("workspace-switcher")

        for _ in range(2):

            virtual_keyboard(15)

            wait_title(active_target)

            virtual_keyboard(15)

            wait_title("workspace-switcher")

        virtual_keyboard(15)

        wait_title(active_target)

        terminate(switcher)

        protocol("pointer", 100)

        current = current_title()

        virtual_keyboard(20)

        virtual_keyboard(20)

        virtual_keyboard(3, 1)

        eventually("focus after moving client", lambda: current_title() not in {"", current})

        virtual_keyboard(3)

        wait_title(current)

        virtual_keyboard(2, 1)

        wait_title("")

        virtual_keyboard(2)

        wait_title(current)

        virtual_keyboard(103)

        wait_title("")

        virtual_keyboard(108)

        wait_title(current)

    test("workspaces", workspaces)

    def workspace_metadata() -> None:

        def field(name: str) -> str:

            return run(BIN / "swmctl", "workspace", "get", 3, name).removesuffix("\n")

        subscription_log = client_log("workspace-subscribe")

        subscriber = spawn(

            subscription_log,

            BIN / "swmctl",

            "workspace",

            "subscribe",

            "--format=waybar",

            env=os.environ.copy(),

        )

        eventually("initial workspace metadata", lambda: subscription_log.stat().st_size)

        run(BIN / "swmctl", "workspace", "set", 3, "title", "code")

        run(BIN / "swmctl", "workspace", "set", 3, "color", "#81a1c180")

        state = {

            "workspace": 3,

            "title": field("title"),

            "color": field("color"),

            "selected": field("selected") == "true",

        }

        if state != {

            "workspace": 3,

            "title": "code",

            "color": "#81a1c180",

            "selected": False,

        }:

            raise Failure(f"unexpected workspace metadata: {state!r}")

        listed = run(BIN / "swmctl", "workspace", "list").splitlines()

        if "3" not in listed:

            raise Failure(f"workspace missing from list: {listed!r}")

        run(BIN / "swmctl", "workspace", "set", 3, "color", "#00000000")

        if field("color"):

            raise Failure("zero RGBA color was not cleared")

        run(BIN / "swmctl", "workspace", "set", 3, "color", "#000000")

        if field("color") != "#000000":

            raise Failure("opaque black color was not retained")

        run(BIN / "swmctl", "workspace", "clear", 3, "color")

        if field("title") != "code" or field("color"):

            raise Failure("workspace color was not cleared")

        run(BIN / "swmctl", "workspace", "set", 3, "color", "#81a1c1")

        protocol("workspace", 3)

        eventually(

            "subscribed workspace metadata",

            lambda: "#81a1c1" in subscription_log.read_text()

            and "code" in subscription_log.read_text(),

        )

        run(BIN / "swmctl", "workspace", "clear", 3, "title")

        run(BIN / "swmctl", "workspace", "clear", 3, "color")

        if field("title") or field("color") or field("selected") != "true":

            raise Failure("workspace metadata was not cleared")

        protocol("workspace", 1)

        terminate(subscriber)

    test("workspace metadata", workspace_metadata)

    def layout_configuration() -> None:

        before_one = geometry("one")

        before_two = geometry("two")

        virtual_keyboard(36, 1)

        eventually(

            "client swap",

            lambda: geometry("one")[0] == before_two[0] and geometry("two")[0] == before_one[0],

        )

        for key in (28, 50, 19, 57):

            virtual_keyboard(key)

        eventually("master-top layout", lambda: geometry("one")[0] == geometry("two")[0])

        before = geometry("one")

        virtual_keyboard(35)

        eventually("master size change", lambda: geometry("one") != before)

    test("layout configuration", layout_configuration)

    def protocols() -> None:

        target = current_title()

        protocol("foreign", target, "fullscreen")

        eventually("foreign fullscreen", lambda: geometry(target)[2] == 1)

        protocol("foreign", target, "unfullscreen")

        eventually("foreign unfullscreen", lambda: geometry(target)[2] == 0)

        protocol("pointer", 100)

        protocol("pointer", 900)

        held_pointer_click("held-left", 100)

        held_pointer_click("held-right", 100, "right")

        for key, modifiers in ((51, 0), (52, 0), (51, 1), (52, 1), (57, 1), (43, 1)):

            virtual_keyboard(key, modifiers)

        protocol("output-power")

        protocol("output-management")

    test("control protocols", protocols)

    def isolation() -> None:

        protocol("workspace", 3)

        wait_title("")

        three = start_client("three", "ff8855aa")

        full = start_client("full", "ffaa5588", True)

        eventually("fullscreen state", lambda: geometry("full")[2] == 1)

        protocol("foreign", "full", "close")

        wait_title("three")

        protocol("workspace", 1)

        eventually("workspace protocol activation", lambda: current_title() in {"one", "two"})

        terminate(three)

        terminate(full)

    test("fullscreen isolation", isolation)

    def max_stack_floating() -> None:

        protocol("workspace", 3)

        wait_title("")

        first = start_client("max-float-one", "55aaffcc")

        second = start_client("max-float-two", "ffaa55cc")

        third = start_client("max-float-three", "aa55ffcc")

        names = ("max-float-one", "max-float-two", "max-float-three")

        virtual_keyboard(33)

        eventually(

            "max-stack geometry before floating toggle",

            lambda: len({tuple(geometry(name)[:2]) for name in names}) == 1,

        )

        virtual_keyboard(20)

        eventually(

            "normal layout restored by floating toggle",

            lambda: len({tuple(geometry(name)[:2]) for name in names}) > 1,

        )

        protocol("workspace", 1)

        eventually("workspace left after floating toggle", lambda: current_title() in {"one", "two"})

        protocol("workspace", 3)

        eventually("workspace returned after floating toggle", lambda: current_title() in names)

        time.sleep(0.1)

        if current_title() not in names:

            raise Failure("workspace did not remain selected after floating toggle")

        terminate(first)

        terminate(second)

        terminate(third)

        protocol("workspace", 1)

        eventually("workspace restored after floating toggle", lambda: current_title() in {"one", "two"})

    test("max-stack floating transition", max_stack_floating)

    def lifecycle() -> None:

        terminate(one)

        terminate(two)

        protocol("xdg-lifecycle", 2)

        protocol("transient", 2)

        protocol("layer", 2)

        protocol("session-lock", 2)

        protocol("text-input")

        protocol("x11")

    test("protocol lifecycle", lifecycle)

    terminate(one)

    terminate(two)

def integration_multioutput() -> None:

    """Exercise workspace movement and output removal."""

    client = start_client("multi", "ff778899")

    def multioutput() -> None:

        virtual_keyboard(105, 1)

        wait_title("")

        virtual_keyboard(106, 1)

        wait_title("multi")

        virtual_keyboard(105, 4)

        wait_title("")

        virtual_keyboard(105, 1)

        wait_title("multi")

        protocol("output-management", "disable-second")

        wait_title("multi")

    test("multiple outputs", multioutput)

    terminate(client)

def fixture_main(kind: str) -> None:

    """Run one fixture inside swm's process tree."""

    result = Path(os.environ["SWM_TEST_RESULT"])

    status = "ok"

    try:

        if kind == "session":

            integration_session()

        elif kind == "multioutput":

            integration_multioutput()

        else:

            raise Failure(f"unknown fixture: {kind}")

    except Exception as error:

        status = str(error)

    finally:

        cleanup_processes()

        result.write_text(status + "\n")

        os.kill(os.getppid(), signal.SIGTERM)

    if status != "ok":

        print(status, file=sys.stderr)

        raise SystemExit(1)

def run_fixture(kind: str, outputs: int) -> None:

    """Run one isolated headless compositor fixture."""

    runtime = BUILD / f"runtime-{kind}"

    state = BUILD / f"state-{kind}"

    fixture = BUILD / f"fixture-{kind}"

    logs = LOGS / kind

    for path in (runtime, state, fixture, logs):

        shutil.rmtree(path, ignore_errors=True)

        path.mkdir(parents=True, exist_ok=True)

    runtime.chmod(0o700)

    result = BUILD / f"{kind}.result"

    progress = BUILD / f"{kind}.progress"

    result.write_text("")

    progress.write_text("")

    env = os.environ.copy()

    env.update(

        {

            "XDG_RUNTIME_DIR": str(runtime),

            "XDG_STATE_HOME": str(state),

            "SWM_NO_AUTOSTART": "1",

            "WLR_BACKENDS": "headless",

            "WLR_RENDERER": "pixman",

            "WLR_LIBINPUT_NO_DEVICES": "1",

            "WLR_HEADLESS_OUTPUTS": str(outputs),

            "SWM_TEST_RESULT": str(result),

            "SWM_TEST_PROGRESS": str(progress),

            "SWM_TEST_DIR": str(fixture),

            "SWM_TEST_LOG_DIR": str(logs),

            "PYTHONPATH": str(BUILD / "python"),

            "LLVM_PROFILE_FILE": str(PROFILES / "integration-%m-%p.profraw"),

        }

    )

    log = LOGS / f"swm-{kind}.log"

    startup = f"{shlex.quote(sys.executable)} {shlex.quote(str(SCRIPT))} --fixture {kind}"

    command = ["timeout", "--foreground", "30", str(BIN / "swm"), "-s", startup]

    compositor = spawn(log, *command, env=env)

    progress_offset = 0

    while compositor.poll() is None:

        with progress.open() as stream:

            stream.seek(progress_offset)

            update = stream.read()

            progress_offset = stream.tell()

        if update:

            print(update, end="", flush=True)

        time.sleep(0.02)

    status = compositor.wait()

    CHILDREN.remove(compositor)

    with progress.open() as stream:

        stream.seek(progress_offset)

        update = stream.read()

    if update:

        print(update, end="", flush=True)

    outcome = result.read_text().strip()

    if status or outcome != "ok":

        raise Failure(

            f"fixture {kind} failed\nresult: {outcome!r}\nstatus: {status}\n"

            f"compositor log tail ({log}):\n{tail(log, 60)}"

        )

    contents = log.read_text(errors="replace")

    if re.search(r"pool exhausted|AddressSanitizer|runtime error:|protocol error", contents):

        raise Failure(f"fixture {kind} reported a runtime failure\n{tail(log, 60)}")

    for path in (runtime, state, fixture):

        shutil.rmtree(path)

def run_integration() -> None:

    """Run every integration fixture."""

    run_fixture("session", 1)

    run_fixture("multioutput", 2)

def coverage_report() -> None:

    """Merge profiles and report production coverage."""

    for executable in ("llvm-profdata", "llvm-cov"):

        if not shutil.which(executable):

            raise Failure(f"required coverage tool not found: {executable}")

    raw = glob.glob(str(PROFILES / "*.profraw"))

    if not raw:

        raise Failure("no LLVM coverage profiles were produced")

    profdata = BUILD / "coverage.profdata"

    run("llvm-profdata", "merge", "-sparse", *raw, "-o", profdata)

    report = run(

        "llvm-cov",

        "report",

        BIN / "swm",

        "-instr-profile",

        profdata,

        "-object",

        BIN / "unit",

        SOURCE / "swm.c",

        ROOT / "util.c",

    )

    files: list[str] = []

    total: list[str] | None = None

    for line in report.splitlines():

        fields = line.split()

        if len(fields) < 10:

            continue

        if fields[0] == "TOTAL":

            total = fields

        elif fields[0].endswith(".c"):

            files.append(f"{Path(fields[0]).name} {fields[9]}")

    if total is None:

        raise Failure("could not parse llvm-cov TOTAL line")

    percent = float(total[9].rstrip("%"))

    print(f"coverage: {', '.join(files)}, aggregate {percent:.2f}%")

def clean() -> None:

    """Remove all generated test artifacts."""

    shutil.rmtree(BUILD, ignore_errors=True)

    print(f"clean: removed {BUILD}")

def main() -> None:

    """Dispatch public runner modes and private fixture modes."""

    if len(sys.argv) == 3 and sys.argv[1] == "--fixture":