test/wayland_pointer.py

Ref: main worktree-web-client-spec   Size: 2.6 KiB   History

"""Drive the retained virtual-pointer helper on an isolated Sway output.

MUXG_TEST_POINTER names the helper binary (line protocol: move x y w h,
button 0/1, wheel NOTCHES; each command returns ok). A real input serial is required for
Wayland clipboard ownership; SDL-injected events cannot establish it.
"""
import json
import select
import subprocess


class Pointer:
    def __init__(self, binary, env, pid):
        self.env, self.pid = env, pid
        self.proc = subprocess.Popen([binary], env=env, stdin=subprocess.PIPE,
                                     stdout=subprocess.PIPE, text=True)

    def command(self, line):
        self.proc.stdin.write(line + '\n')
        self.proc.stdin.flush()
        if not select.select([self.proc.stdout], [], [], 3)[0]:
            raise RuntimeError('virtual pointer command timed out')
        if self.proc.stdout.readline().strip() != 'ok':
            raise RuntimeError('virtual pointer command failed')

    def query(self, kind):
        return json.loads(subprocess.check_output(['swaymsg', '-r', '-t', kind],
                                                 env=self.env, timeout=3))

    def move(self, x, y):
        def find(node):
            if node.get('pid') == self.pid:
                return node
            for child in node.get('nodes', []) + node.get('floating_nodes', []):
                found = find(child)
                if found:
                    return found
        node = find(self.query('get_tree'))
        if node is None:
            raise RuntimeError('owned native window is not mapped')
        outputs = self.query('get_outputs')
        if len(outputs) != 1 or not outputs[0]['name'].startswith('HEADLESS-'):
            raise RuntimeError('pointer fixture requires one isolated headless output')
        rect, screen = node['rect'], outputs[0]['rect']
        self.command(f"move {int(rect['x'] + x - screen['x'])} "
                     f"{int(rect['y'] + y - screen['y'])} {screen['width']} {screen['height']}")

    def event(self, kind, x, y):
        self.move(x, y)
        if kind in ('mousedown', 'click'):
            self.command('button 1')
        if kind in ('mouseup', 'click'):
            self.command('button 0')

    def wheel(self, x, y, notches):
        self.move(x, y)
        self.command(f'wheel {notches}')

    def close(self):
        self.proc.stdin.close()
        try:
            if self.proc.wait(timeout=3):
                raise RuntimeError('virtual pointer failed on shutdown')
        except subprocess.TimeoutExpired:
            self.proc.kill()
            self.proc.wait(timeout=3)
            raise
        finally:
            self.proc.stdout.close()