#!/usr/bin/env python3

"""Protocol clients used by the Python integration runner."""

from __future__ import annotations

import ctypes

import mmap

import os

import struct

import subprocess

import sys

import tempfile

import time

from pathlib import Path

from pywayland.client import Display

from protocols.ext_session_lock_v1 import (

    ExtSessionLockManagerV1,

)

from protocols.ext_workspace_v1 import ExtWorkspaceManagerV1

from protocols.idle_inhibit_unstable_v1 import ZwpIdleInhibitManagerV1

from protocols.input_method_unstable_v2 import ZwpInputMethodManagerV2

from protocols.keyboard_shortcuts_inhibit_unstable_v1 import (

    ZwpKeyboardShortcutsInhibitManagerV1,

)

from protocols.pointer_constraints_unstable_v1 import ZwpPointerConstraintsV1

from protocols.relative_pointer_unstable_v1 import ZwpRelativePointerManagerV1

from protocols.text_input_unstable_v3 import ZwpTextInputManagerV3

from protocols.virtual_keyboard_unstable_v1 import ZwpVirtualKeyboardManagerV1

from protocols.wayland import (

    WlCompositor,

    WlKeyboard,

    WlOutput,

    WlSeat,

    WlShm,

)

from protocols.wlr_foreign_toplevel_management_unstable_v1 import (

    ZwlrForeignToplevelManagerV1,

)

from protocols.wlr_layer_shell_unstable_v1 import (

    ZwlrLayerShellV1,

    ZwlrLayerSurfaceV1,

)

from protocols.wlr_output_power_management_unstable_v1 import (

    ZwlrOutputPowerManagerV1,

)

from protocols.wlr_virtual_pointer_unstable_v1 import ZwlrVirtualPointerManagerV1

from protocols.xdg_shell import XdgPositioner, XdgToplevel, XdgWmBase

class Failure(RuntimeError):

    """Report a protocol-client failure."""

class Connection:

    """Bind requested globals on one Wayland connection."""

    def __init__(self, *interfaces):

        self.display = Display()

        self.display.connect()

        self.registry = self.display.get_registry()

        self.interfaces = {interface.name: interface for interface in interfaces}

        self.globals: dict[str, object] = {}

        self._versions: dict[str, int] = {}

        self.registry.dispatcher["global"] = self._global

        self.display.roundtrip()

    def _global(self, registry, name: int, interface: str, version: int) -> None:

        wanted = self.interfaces.get(interface)

        if wanted is not None and interface not in self.globals:

            self._versions[interface] = version

            self.globals[interface] = registry.bind(name, wanted, min(version, wanted.version))

    def get(self, interface):

        """Return a required bound global."""

        try:

            return self.globals[interface.name]

        except KeyError as error:

            raise Failure(f"compositor does not advertise {interface.name}") from error

    def require_version(self, interface, version: int) -> None:

        """Require a minimum advertised version for one global."""

        advertised = self._versions.get(interface.name, 0)

        if advertised < version:

            raise Failure(

                f"compositor advertises {interface.name} version {advertised}, "

                f"but version {version} is required"

            )

    def roundtrips(self, count: int = 1) -> None:

        """Complete several request/event exchanges."""

        for _ in range(count):

            if self.display.roundtrip() < 0:

                raise Failure("Wayland roundtrip failed")

    def close(self) -> None:

        """Disconnect from the compositor."""

        self.display.disconnect()

def create_buffer(shm, width: int, height: int, color: int):

    """Create a solid-color shared-memory buffer."""

    size = width * height * 4

    with tempfile.TemporaryFile(dir=os.environ.get("SWM_TEST_DIR")) as stream:

        stream.truncate(size)

        pixels = mmap.mmap(stream.fileno(), size)

        pixels[:] = struct.pack("=I", color) * (width * height)

        pixels.flush()

        pool = shm.create_pool(stream.fileno(), size)

        buffer = pool.create_buffer(0, width, height, width * 4, WlShm.format.argb8888)

        pool.destroy()

    buffer.dispatcher["release"] = lambda proxy: proxy.destroy()

    return buffer

class Window:

    """Own one mapped XDG toplevel and report configure state."""

    def __init__(

        self,

        connection: Connection,

        title: str,

        color: int,

        parent=None,

        fullscreen: bool = False,

        report: Path | None = None,

        fixed_size: bool = False,

    ):

        self.connection = connection

        self.compositor = connection.get(WlCompositor)

        self.shm = connection.get(WlShm)

        self.wm_base = connection.get(XdgWmBase)

        self.title = title

        self.color = color

        self.width = 320

        self.height = 200

        self.fullscreen = False

        self.running = True

        self.report = report

        self.wm_base.dispatcher["ping"] = lambda proxy, serial: proxy.pong(serial)

        self.surface = self.compositor.create_surface()

        self.xdg_surface = self.wm_base.get_xdg_surface(self.surface)

        self.xdg_surface.dispatcher["configure"] = self._surface_configure

        self.toplevel = self.xdg_surface.get_toplevel()

        self.toplevel.dispatcher["configure"] = self._toplevel_configure

        self.toplevel.dispatcher["close"] = lambda proxy: setattr(self, "running", False)

        self.toplevel.set_title(title)

        self.toplevel.set_app_id("swm-test")

        if parent is not None:

            self.toplevel.set_parent(parent)

        if fullscreen:

            self.toplevel.set_fullscreen(None)

        if fixed_size:

            self.toplevel.set_min_size(self.width, self.height)

            self.toplevel.set_max_size(self.width, self.height)

        self.surface.commit()

    def _toplevel_configure(self, proxy, width: int, height: int, states) -> None:

        if width > 0:

            self.width = width

        if height > 0:

            self.height = height

        self.fullscreen = int(XdgToplevel.state.fullscreen) in list(states)

        self._write_report()

    def _surface_configure(self, proxy, serial: int) -> None:

        proxy.ack_configure(serial)

        self.draw()

    def _write_report(self) -> None:

        if self.report:

            self.report.write_text(f"{self.width} {self.height} {int(self.fullscreen)}\n")

    def draw(self) -> None:

        """Attach a fresh buffer and commit the surface."""

        buffer = create_buffer(self.shm, self.width, self.height, self.color)

        self.surface.attach(buffer, 0, 0)

        self.surface.damage_buffer(0, 0, self.width, self.height)

        self.surface.commit()

        self._write_report()

    def unmap(self) -> None:

        """Detach the surface buffer."""

        self.surface.attach(None, 0, 0)

        self.surface.commit()

    def destroy(self) -> None:

        """Destroy the complete toplevel role."""

        self.toplevel.destroy()

        self.xdg_surface.destroy()

        self.surface.destroy()

def xdg_client(arguments: list[str]) -> None:

    """Run one persistent XDG window."""

    color = int(arguments[0], 16) if arguments else 0xFF336699

    title = arguments[1] if len(arguments) > 1 else "swm-test"

    report = Path(os.environ["SWM_CLIENT_REPORT"]) if os.environ.get("SWM_CLIENT_REPORT") else None

    connection = Connection(WlCompositor, WlShm, XdgWmBase)

    window = Window(

        connection,

        title,

        color,

        fullscreen="fullscreen" in arguments[2:],

        report=report,

    )

    while window.running:

        connection.display.dispatch(block=True)

    window.destroy()

    connection.close()

def xdg_resize(arguments: list[str]) -> None:

    """Commit a client-selected size after an XDG window maps.

    The window maps with fixed size limits so the compositor cannot choose its

    initial size, then drops those limits and commits each requested size on

    its own once the runner triggers it.

    """

    if len(arguments) != 4:

        raise Failure("xdg-resize requires WIDTH HEIGHT WIDTH HEIGHT")

    trigger = Path(os.environ["SWM_RESIZE_TRIGGER"])

    committed = Path(os.environ["SWM_RESIZE_COMMITTED"])

    connection = Connection(WlCompositor, WlShm, XdgWmBase)

    window = Window(

        connection,

        "client-resize",

        0xFF336699,

        fixed_size=True,

        report=Path(os.environ["SWM_CLIENT_REPORT"]),

    )

    connection.roundtrips(3)

    # The window mapped at a size the compositor could not override. Unlock the

    # size limits so later commits choose their own size the way an application

    # that resizes itself does.

    window.toplevel.set_min_size(0, 0)

    window.toplevel.set_max_size(0, 0)

    window.surface.commit()

    connection.roundtrips(2)

    for index in range(2):

        deadline = time.monotonic() + 5

        while not trigger.exists() or trigger.read_text().strip() != str(index):

            if time.monotonic() >= deadline:

                raise Failure(f"timed out waiting for resize trigger {index}")

            time.sleep(0.01)

        window.width = int(arguments[index * 2])

        window.height = int(arguments[index * 2 + 1])

        window.xdg_surface.set_window_geometry(0, 0, window.width, window.height)

        window.draw()

        connection.display.flush()

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

    while window.running:

        connection.display.dispatch(block=True)

    window.destroy()

    connection.close()

def xdg_lifecycle(arguments: list[str]) -> None:

    """Repeatedly map and unmap a focused XDG toplevel."""

    count = int(arguments[0]) if arguments else 8

    connection = Connection(WlCompositor, WlShm, XdgWmBase)

    fallback = Window(connection, "swm-xdg-fallback", 0xFF446622)

    window = Window(connection, "swm-xdg-lifecycle", 0xFF224466)

    connection.roundtrips(3)

    for index in range(count):

        window.unmap()

        connection.roundtrips(2)

        window.color += 0x00050311

        window.surface.commit()

        connection.roundtrips(2)

    window.destroy()

    fallback.destroy()

    connection.close()

def transient(arguments: list[str]) -> None:

    """Exercise popup, parented, and fullscreen XDG surfaces."""

    count = int(arguments[0]) if arguments else 4

    connection = Connection(WlCompositor, WlShm, XdgWmBase)

    parent = Window(connection, "swm-child-parent", 0xFF335577)

    fullscreen = Window(connection, "swm-fullscreen", 0xFF224466, fullscreen=True)

    connection.roundtrips(4)

    if not fullscreen.fullscreen:

        raise Failure("fullscreen window was not configured fullscreen")

    positioner = connection.get(XdgWmBase).create_positioner()

    positioner.set_size(120, 60)

    positioner.set_anchor_rect(30, 20, 10, 10)

    positioner.set_anchor(XdgPositioner.anchor.bottom_right)

    positioner.set_gravity(XdgPositioner.gravity.bottom_right)

    positioner.set_constraint_adjustment(

        XdgPositioner.constraint_adjustment.slide_x

        | XdgPositioner.constraint_adjustment.slide_y

    )

    popup_surface = connection.get(WlCompositor).create_surface()

    popup_xdg = connection.get(XdgWmBase).get_xdg_surface(popup_surface)

    popup_configured = []

    def configure(proxy, serial) -> None:

        proxy.ack_configure(serial)

        buffer = create_buffer(connection.get(WlShm), 120, 60, 0xFF557733)

        popup_surface.attach(buffer, 0, 0)

        popup_surface.damage_buffer(0, 0, 120, 60)

        popup_surface.commit()

        popup_configured.append(True)

    popup_xdg.dispatcher["configure"] = configure

    popup = popup_xdg.get_popup(parent.xdg_surface, positioner)

    popup.dispatcher["configure"] = lambda proxy, x, y, width, height: None

    popup.dispatcher["popup_done"] = lambda proxy: None

    popup.dispatcher["repositioned"] = lambda proxy, token: None

    positioner.destroy()

    popup_surface.commit()

    connection.roundtrips(4)

    if not popup_configured:

        raise Failure("transient popup was not configured")

    popup.destroy()

    popup_xdg.destroy()

    popup_surface.destroy()

    connection.roundtrips()

    for index in range(count):

        child = Window(

            connection,

            f"swm-child-{index}",

            0xFF773355 + index * 0x0003070D,

            parent=parent.toplevel,

        )

        connection.roundtrips(3)

        if not fullscreen.fullscreen:

            raise Failure("transient child stole fullscreen")

        child.destroy()

        connection.roundtrips()

    fullscreen.destroy()

    parent.destroy()

    connection.close()

def keymap_text() -> tuple[bytes, dict[str, int]]:

    """Build a default XKB keymap and return modifier indices."""

    xkb = ctypes.CDLL("libxkbcommon.so.0")

    xkb.xkb_context_new.argtypes = [ctypes.c_int]

    xkb.xkb_context_new.restype = ctypes.c_void_p

    xkb.xkb_keymap_new_from_names.argtypes = [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_int]

    xkb.xkb_keymap_new_from_names.restype = ctypes.c_void_p

    xkb.xkb_keymap_get_as_string.argtypes = [ctypes.c_void_p, ctypes.c_int]

    xkb.xkb_keymap_get_as_string.restype = ctypes.c_void_p

    xkb.xkb_keymap_mod_get_index.argtypes = [ctypes.c_void_p, ctypes.c_char_p]

    xkb.xkb_keymap_mod_get_index.restype = ctypes.c_uint

    xkb.xkb_keymap_unref.argtypes = [ctypes.c_void_p]

    xkb.xkb_context_unref.argtypes = [ctypes.c_void_p]

    context = xkb.xkb_context_new(0)

    keymap = xkb.xkb_keymap_new_from_names(context, None, 0)

    pointer = xkb.xkb_keymap_get_as_string(keymap, 1)

    text = ctypes.string_at(pointer)

    ctypes.CDLL(None).free(ctypes.c_void_p(pointer))

    indices = {

        "shift": xkb.xkb_keymap_mod_get_index(keymap, b"Shift"),

        "ctrl": xkb.xkb_keymap_mod_get_index(keymap, b"Control"),

        "alt": xkb.xkb_keymap_mod_get_index(keymap, b"Mod1"),

        "logo": xkb.xkb_keymap_mod_get_index(keymap, b"Mod4"),

    }

    xkb.xkb_keymap_unref(keymap)

    xkb.xkb_context_unref(context)

    return text, indices

def keyboard(arguments: list[str]) -> None:

    """Send a key and modifiers through virtual-keyboard-v1."""

    if len(arguments) != 2:

        raise Failure("keyboard requires KEYCODE|none MODIFIERS")

    key = None if arguments[0] == "none" else int(arguments[0], 0)

    requested = int(arguments[1], 0)

    connection = Connection(WlSeat, ZwpVirtualKeyboardManagerV1)

    seat = connection.get(WlSeat)

    manager = connection.get(ZwpVirtualKeyboardManagerV1)

    virtual = manager.create_virtual_keyboard(seat)

    text, indices = keymap_text()

    with tempfile.TemporaryFile(dir=os.environ.get("SWM_TEST_DIR")) as stream:

        stream.write(text + b"\0")

        stream.flush()

        virtual.keymap(WlKeyboard.keymap_format.xkb_v1, stream.fileno(), len(text) + 1)

    modifiers = 0

    modifier_keys = (

        (1, "shift", 42),

        (4, "ctrl", 29),

        (8, "alt", 56),

        (64, "logo", 125),

    )

    for flag, name, code in modifier_keys:

        if requested & flag:

            modifiers |= 1 << indices[name]

            virtual.key(0, code, WlKeyboard.key_state.pressed)

    virtual.modifiers(modifiers, 0, 0, 0)

    if key is not None:

        virtual.key(1, key, WlKeyboard.key_state.pressed)

        virtual.key(2, key, WlKeyboard.key_state.released)

    connection.roundtrips()

    ready = os.environ.get("SWM_KEYBOARD_READY")

    if ready:

        Path(ready).write_text("ready\n")

    release = os.environ.get("SWM_KEYBOARD_RELEASE")

    if release:

        deadline = time.monotonic() + 2

        while not Path(release).exists():

            if time.monotonic() >= deadline:

                raise Failure("timed out waiting to release virtual keyboard")

            time.sleep(0.002)

    for flag, _, code in modifier_keys:

        if requested & flag:

            virtual.key(3, code, WlKeyboard.key_state.released)

    virtual.modifiers(0, 0, 0, 0)

    virtual.destroy()

    connection.roundtrips()

    connection.close()

def pointer(arguments: list[str]) -> None:

    """Move and click through virtual-pointer-v1."""

    x = int(arguments[0])

    button = 0x111 if len(arguments) > 1 and arguments[1] == "right" else 0x110

    connection = Connection(WlSeat, ZwlrVirtualPointerManagerV1)

    manager = connection.get(ZwlrVirtualPointerManagerV1)

    virtual = manager.create_virtual_pointer(connection.get(WlSeat))

    virtual.motion_absolute(0, x, 500, 1000, 1000)

    virtual.frame()

    virtual.button(1, button, 1)

    virtual.motion(2, 25.0, 20.0)

    virtual.frame()

    virtual.button(3, button, 0)

    virtual.axis_source(0)

    virtual.axis_discrete(3, 0, 1.0, 1)

    virtual.axis_stop(4, 0)

    virtual.frame()

    connection.roundtrips()

    virtual.destroy()

    connection.close()

def foreign(arguments: list[str]) -> None:

    """Apply one foreign-toplevel action to a named window."""

    title, action = arguments

    connection = Connection(WlSeat, ZwlrForeignToplevelManagerV1)

    seat = connection.get(WlSeat)

    manager = connection.get(ZwlrForeignToplevelManagerV1)

    acted = False

    observed = False

    def toplevel(proxy, handle):

        nonlocal acted, observed

        state = {"title": "", "fullscreen": False}

        fullscreen_state = int(handle.interface.state.fullscreen)

        handle.dispatcher["title"] = lambda p, value: state.update(title=value)

        def states(p, values):

            nonlocal acted, observed

            state["fullscreen"] = fullscreen_state in list(values)

            if state["title"] != title:

                return

            if not acted:

                if action == "activate":

                    handle.activate(seat)

                elif action == "fullscreen":

                    handle.set_fullscreen(None)

                elif action == "unfullscreen":

                    handle.unset_fullscreen()

                elif action == "close":

                    handle.close()

                else:

                    raise Failure(f"unknown foreign action: {action}")

                acted = True

            observed = (

                action == "close"

                or action == "fullscreen" and state["fullscreen"]

                or action in {"activate", "unfullscreen"} and not state["fullscreen"]

            )

        handle.dispatcher["state"] = states

        handle.dispatcher["done"] = lambda p: states(

            p, [fullscreen_state] if state["fullscreen"] else []

        )

        handle.dispatcher["closed"] = lambda p: None

    manager.dispatcher["toplevel"] = toplevel

    for _ in range(20):

        connection.roundtrips()

        if acted and observed:

            break

    if not acted or not observed:

        raise Failure(f"foreign-toplevel action {action} was not observed")

    if action == "close":

        connection.roundtrips(2)

    connection.close()

def workspace(arguments: list[str]) -> None:

    """Activate a named workspace through ext-workspace-v1."""

    if len(arguments) != 1:

        raise Failure("workspace requires a workspace name")

    target = arguments[0]

    connection = Connection(ExtWorkspaceManagerV1)

    manager = connection.get(ExtWorkspaceManagerV1)

    workspaces: dict[str, object] = {}

    def announced(proxy, handle):

        handle.dispatcher["name"] = lambda p, name: workspaces.update({name: handle})

        handle.dispatcher["removed"] = lambda p: None

    manager.dispatcher["workspace_group"] = lambda proxy, group: None

    manager.dispatcher["workspace"] = announced

    manager.dispatcher["done"] = lambda proxy: None

    manager.dispatcher["finished"] = lambda proxy: None

    connection.roundtrips(2)

    if target not in workspaces:

        raise Failure(f"workspace {target!r} was not advertised")

    workspaces[target].activate()

    manager.commit()

    connection.roundtrips(2)

    connection.close()

def output_power(arguments: list[str]) -> None:

    """Turn one output off and back on through output-power-v1."""

    connection = Connection(WlOutput, ZwlrOutputPowerManagerV1)

    power = connection.get(ZwlrOutputPowerManagerV1).get_output_power(connection.get(WlOutput))

    modes: list[int] = []

    failed: list[bool] = []

    power.dispatcher["mode"] = lambda proxy, mode: modes.append(mode)

    power.dispatcher["failed"] = lambda proxy: failed.append(True)

    connection.roundtrips()

    if not modes or modes[-1] != 1:

        raise Failure("output did not begin powered on")

    power.set_mode(0)

    connection.roundtrips(2)

    if modes[-1] != 0:

        raise Failure("output did not power off")

    power.set_mode(1)

    connection.roundtrips(2)

    if modes[-1] != 1 or failed:

        raise Failure("output did not power back on")

    power.destroy()

    connection.close()

def output_management(arguments: list[str]) -> None:

    """Exercise output-management through the installed reference client."""

    command = ["wlr-randr"]

    if arguments and arguments[0] in {"disable-first", "disable-second"}:

        output = "HEADLESS-1" if arguments[0] == "disable-first" else "HEADLESS-2"

        command += ["--output", output, "--off"]

    else:

        command += ["--output", "HEADLESS-1", "--on"]

        subprocess.run([*command, "--dryrun"], check=True)

    subprocess.run(command, check=True)

def layer(arguments: list[str]) -> None:

    """Create, configure, map, and destroy layer-shell surfaces."""

    count = int(arguments[0]) if arguments else 2

    connection = Connection(

        WlCompositor,

        WlSeat,

        WlShm,

        ZwpVirtualKeyboardManagerV1,

        ZwlrLayerShellV1,

        ZwlrVirtualPointerManagerV1,

    )

    compositor = connection.get(WlCompositor)

    seat = connection.get(WlSeat)

    shm = connection.get(WlShm)

    shell = connection.get(ZwlrLayerShellV1)

    connection.require_version(ZwlrLayerShellV1, 4)

    virtual_keyboard = connection.get(ZwpVirtualKeyboardManagerV1).create_virtual_keyboard(

        seat

    )

    keymap, _ = keymap_text()

    keymap_stream = tempfile.TemporaryFile(dir=os.environ.get("SWM_TEST_DIR"))

    keymap_stream.write(keymap + b"\0")

    keymap_stream.flush()

    virtual_keyboard.keymap(

        WlKeyboard.keymap_format.xkb_v1,

        keymap_stream.fileno(),

        len(keymap) + 1,

    )

    connection.roundtrips()

    keyboard = seat.get_keyboard()

    keyboard_enters = []

    keyboard.dispatcher["enter"] = lambda proxy, serial, surface, keys: (

        keyboard_enters.append(surface)

    )

    keyboard.dispatcher["leave"] = lambda proxy, serial, surface: None

    layers = []

    for index in range(count):

        surface = compositor.create_surface()

        role = shell.get_layer_surface(

            surface,

            None,

            ZwlrLayerShellV1.layer.top,

            f"swm-layer-{index}",

        )

        role.set_keyboard_interactivity(

            ZwlrLayerSurfaceV1.keyboard_interactivity.on_demand

        )

        role.set_anchor(

            ZwlrLayerSurfaceV1.anchor.top

            | (

                ZwlrLayerSurfaceV1.anchor.left

                if index == 0

                else ZwlrLayerSurfaceV1.anchor.right

            )

        )

        configured = []

        def configure(

            proxy,

            serial,

            width,

            height,

            surface=surface,

            configured=configured,

            index=index,

        ):

            proxy.ack_configure(serial)

            width = width or 80

            height = height or 40

            surface.attach(create_buffer(shm, width, height, 0xCC112233 + index), 0, 0)

            surface.damage_buffer(0, 0, width, height)

            surface.commit()

            configured.append(True)

        role.dispatcher["configure"] = configure

        role.dispatcher["closed"] = lambda proxy: None

        role.set_size(80 + index, 40 + index)

        surface.commit()

        layers.append((surface, role, configured))

    connection.roundtrips(3)

    if not all(configured for _, _, configured in layers):

        raise Failure("layer surface was not configured")

    if keyboard_enters:

        raise Failure("on-demand layer surface received keyboard focus when mapped")

    pointer = connection.get(ZwlrVirtualPointerManagerV1).create_virtual_pointer(seat)

    pointer.motion_absolute(0, 20, 20, 1000, 1000)

    pointer.frame()

    pointer.button(1, 0x110, 1)

    pointer.frame()

    pointer.button(2, 0x110, 0)

    pointer.frame()

    connection.roundtrips(2)

    if not keyboard_enters:

        raise Failure("on-demand layer surface did not receive keyboard focus when pressed")

    pointer.destroy()

    keyboard.release()

    virtual_keyboard.destroy()

    keymap_stream.close()

    for surface, role, _ in reversed(layers):

        role.destroy()

        surface.destroy()

    connection.roundtrips()

    connection.close()

def session_lock(arguments: list[str]) -> None:

    """Create and release session locks."""

    count = int(arguments[0]) if arguments else 2

    connection = Connection(WlCompositor, WlShm, WlOutput, ExtSessionLockManagerV1)

    compositor = connection.get(WlCompositor)

    shm = connection.get(WlShm)

    output = connection.get(WlOutput)

    manager = connection.get(ExtSessionLockManagerV1)

    for index in range(count):

        lock = manager.lock()

        locked: list[bool] = []

        finished: list[bool] = []

        lock.dispatcher["locked"] = lambda proxy: locked.append(True)

        lock.dispatcher["finished"] = lambda proxy: finished.append(True)

        surface = compositor.create_surface()

        role = lock.get_lock_surface(surface, output)

        configured: list[bool] = []

        def configure(proxy, serial, width, height):

            proxy.ack_configure(serial)

            surface.attach(create_buffer(shm, width, height, 0xFF101010 + index), 0, 0)

            surface.damage_buffer(0, 0, width, height)

            surface.commit()

            configured.append(True)

        role.dispatcher["configure"] = configure

        connection.roundtrips(3)

        if not finished and (not locked or not configured):

            raise Failure("session lock was not configured")

        role.destroy()

        surface.destroy()

        if locked:

            lock.unlock_and_destroy()

        else:

            lock.destroy()

        connection.roundtrips()

    connection.close()

def text_input(arguments: list[str]) -> None:

    """Exercise text input, input methods, and input inhibitors together."""

    connection = Connection(

        WlCompositor,

        WlShm,

        WlSeat,

        XdgWmBase,

        ZwpTextInputManagerV3,

        ZwpInputMethodManagerV2,

        ZwpKeyboardShortcutsInhibitManagerV1,

        ZwpPointerConstraintsV1,

        ZwpRelativePointerManagerV1,

        ZwpIdleInhibitManagerV1,

        ZwlrVirtualPointerManagerV1,

    )

    seat = connection.get(WlSeat)

    pointer = seat.get_pointer()

    for event in ("enter", "leave", "motion", "button", "axis"):

        pointer.dispatcher[event] = lambda proxy, *values: None

    text = connection.get(ZwpTextInputManagerV3).get_text_input(seat)

    method = connection.get(ZwpInputMethodManagerV2).get_input_method(seat)

    state = {

        "activated": False,

        "surrounding": False,

        "cause": False,

        "content": False,

        "preedit": False,

        "commit": False,

        "delete": False,

        "done": False,

        "rectangle": False,

        "shortcuts": False,

        "locked": False,

        "relative": False,

        "serial": 0,

    }

    def text_enter(proxy, surface) -> None:

        if surface != window.surface:

            return

        text.enable()

        text.set_surrounding_text("before after", 6, 6)

        text.set_text_change_cause(1)

        text.set_content_type(1, 0)

        text.set_cursor_rectangle(7, 11, 3, 15)

        text.commit()

    def method_done(proxy) -> None:

        state["serial"] += 1

        if not all(state[name] for name in ("activated", "surrounding", "cause", "content")):

            return

        method.set_preedit_string("preedit", 2, 4)

        method.commit_string("committed")

        method.delete_surrounding_text(2, 3)

        method.commit(state["serial"])

    text.dispatcher["enter"] = text_enter

    text.dispatcher["leave"] = lambda proxy, surface: None

    text.dispatcher["preedit_string"] = lambda proxy, value, begin, end: state.update(

        preedit=value == "preedit" and begin == 2 and end == 4

    )

    text.dispatcher["commit_string"] = lambda proxy, value: state.update(

        commit=value == "committed"

    )

    text.dispatcher["delete_surrounding_text"] = lambda proxy, before, after: state.update(

        delete=before == 2 and after == 3

    )

    text.dispatcher["done"] = lambda proxy, serial: state.update(done=True)

    popup_surface = connection.get(WlCompositor).create_surface()

    popup = method.get_input_popup_surface(popup_surface)

    popup.dispatcher["text_input_rectangle"] = lambda proxy, x, y, width, height: state.update(

        rectangle=(x, y, width, height) == (0, -15, 3, 15)

    )

    def activate(proxy) -> None:

        state["activated"] = True

        buffer = create_buffer(connection.get(WlShm), 32, 16, 0xFF8855AA)

        popup_surface.attach(buffer, 0, 0)

        popup_surface.damage_buffer(0, 0, 32, 16)

        popup_surface.commit()

    method.dispatcher["activate"] = activate

    method.dispatcher["deactivate"] = lambda proxy: None

    method.dispatcher["surrounding_text"] = lambda proxy, value, cursor, anchor: state.update(

        surrounding=value == "before after" and cursor == 6 and anchor == 6

    )

    method.dispatcher["text_change_cause"] = lambda proxy, cause: state.update(cause=cause == 1)

    method.dispatcher["content_type"] = lambda proxy, hint, purpose: state.update(

        content=hint == 1 and purpose == 0

    )

    method.dispatcher["done"] = method_done

    method.dispatcher["unavailable"] = lambda proxy: (_ for _ in ()).throw(

        Failure("input method is unavailable")

    )

    window = Window(connection, "text-input", 0xFF556677)

    idle = connection.get(ZwpIdleInhibitManagerV1).create_inhibitor(window.surface)

    shortcuts = connection.get(ZwpKeyboardShortcutsInhibitManagerV1).inhibit_shortcuts(

        window.surface, seat

    )

    shortcuts.dispatcher["active"] = lambda proxy: state.update(shortcuts=True)

    shortcuts.dispatcher["inactive"] = lambda proxy: state.update(shortcuts=False)

    locked = connection.get(ZwpPointerConstraintsV1).lock_pointer(

        window.surface, pointer, None, 2

    )

    locked.dispatcher["locked"] = lambda proxy: state.update(locked=True)

    locked.dispatcher["unlocked"] = lambda proxy: state.update(locked=False)

    relative = connection.get(ZwpRelativePointerManagerV1).get_relative_pointer(pointer)

    relative.dispatcher["relative_motion"] = lambda proxy, hi, lo, dx, dy, ux, uy: state.update(

        relative=dx != 0

    )

    virtual = connection.get(ZwlrVirtualPointerManagerV1).create_virtual_pointer(seat)

    for index in range(100):

        virtual.motion_absolute(0, index % 10 * 100 + 50, index // 10 * 100 + 50, 1000, 1000)

        virtual.frame()

        connection.roundtrips()

        if state["locked"]:

            break

    virtual.motion(10, 5.0, 0.0)

    virtual.frame()

    connection.roundtrips(3)

    required = (

        "preedit",

        "commit",

        "delete",

        "done",