test/native_mouse.py
Ref: Size: 9.2 KiB History
#!/usr/bin/env python3
"""Application mouse reports and OSC 52 through native SDL input and real PTYs.
The FIFO's generic `mouse:KIND,X,Y,BUTTON,MODS` form is deliberately used
here: it reaches the same SDL dispatch path as a user event, while the
fixture can pin all button and modifier combinations without a compositor.
"""
import os
import shlex
import subprocess
import sys
import time
sys.dont_write_bytecode = True
from native_lifecycle import start_workspace
from native_resize import by_id
from native_selection import SelectionRig
from native_tiling import eventually, require
SHIFT = 1
class MouseRig(SelectionRig):
def mouse(self, kind, point, button=0, mods=0):
"""Inject a normalized SDL mouse event through the generic hook."""
x, y = point.split(',')
self.send(f'mouse:{kind},{x},{y},{button},{mods}')
def mouse_cell(self, state, pane, col, row, kind, button=0, mods=0):
self.mouse(kind, self.cell_point(state, pane, col, row), button, mods)
def desktop(self):
return self.clipboard()
def primary(self):
value = self.artifact('primary', '.txt').read_text()
if self.env['SDL_VIDEO_DRIVER'] == 'wayland':
external = subprocess.run(['wl-paste', '--primary', '--no-newline'], env=self.env,
capture_output=True, timeout=3)
if external.returncode or external.stdout.decode() != value:
return None
return value
def wait_desktop(self, text):
eventually(lambda: self.desktop() == text, 'desktop clipboard did not become ' + repr(text))
def wait_primary(self, text):
eventually(lambda: self.primary() == text, 'primary selection did not become ' + repr(text))
def reader_program(rig, tag):
"""A raw foreground terminal app with independent input and output oracles."""
source = rig.root / ('mouse-reader-' + tag + '.py')
received = rig.root / ('mouse-received-' + tag + '.bin')
control = rig.root / ('mouse-control-' + tag)
source.write_text(
'import os, select, sys, termios, tty, time\n'
'from pathlib import Path\n'
f'received=Path({str(received)!r}); control=Path({str(control)!r})\n'
'fd=sys.stdin.fileno(); old=termios.tcgetattr(fd); tty.setraw(fd)\n'
'try:\n'
' sys.stdout.write("\\033[?25l\\033[2J\\033[HAPP-MOUSE-READY\\033[2;1Halpha café omega\\033[?1002h\\033[?1006h"); sys.stdout.flush()\n'
' seen=0\n'
' while True:\n'
' if control.exists():\n'
' commands=control.read_text()[seen:]; seen += len(commands)\n'
' for command in commands.splitlines():\n'
' if command == "off":\n'
' sys.stdout.write("\\033[?1002l\\033[4;1HMODE-OFF"); sys.stdout.flush(); continue\n'
' if command == "on":\n'
' sys.stdout.write("\\033[?1002h\\033[4;1HMODE-ON "); sys.stdout.flush(); continue\n'
' target={"c":"c","p":"p","s":"s","bad":"c","q":"q"}.get(command)\n'
' if target:\n'
' value={"c":"DESKTOP-C","p":"PRIMARY-P","s":"PRIMARY-S","bad":"\\x00","q":"IGNORED-Q"}[command]\n'
' import base64\n'
' sys.stdout.write("\\033]52;"+target+";"+base64.b64encode(value.encode()).decode()+"\\a"); sys.stdout.flush()\n'
' ready, _, _ = select.select([fd], [], [], .02)\n'
' if ready:\n'
' data=os.read(fd, 4096)\n'
' if not data: break\n'
' with received.open("ab") as out: out.write(data)\n'
'finally:\n'
' termios.tcsetattr(fd, termios.TCSADRAIN, old)\n')
return source, received, control
def start_reader(rig, pane, tag):
source, received, control = reader_program(rig, tag)
rig.focus(pane) # A real Wayland pointer serial precedes any OSC 52 write.
rig.shell('python3 ' + shlex.quote(str(source)))
rig.wait_state(lambda s: 'APP-MOUSE-READY' in by_id(s)[pane]['painted_text'])
return received, control
def bytes_after(path, offset, expected, message):
eventually(lambda: path.exists() and path.stat().st_size >= offset + len(expected), message)
actual = path.read_bytes()[offset:]
require(actual == expected, f'{message}: got {actual!r}, expected {expected!r}')
def application_drag(rig, pane, other):
received, control = start_reader(rig, pane, 'direct')
state = rig.state()
# 1002 + 1006: left press, held motion, then SGR release. The cells
# are one-based on the wire and relative to the selected source pane.
rig.mouse_cell(state, pane, 0, 1, 'down')
rig.mouse_cell(state, pane, 4, 1, 'move')
rig.mouse_cell(state, pane, 4, 1, 'up')
expected = b'\033[<0;1;2M\033[<32;5;2M\033[<0;5;2m'
bytes_after(received, 0, expected, 'application press/drag/release did not reach its PTY')
# Middle/right belong to the application too. Adding Shift after
# press changes report modifiers without turning this into local copy.
for button in (1, 2):
offset = received.stat().st_size
rig.mouse_cell(state, pane, 0, 1, 'down', button=button)
rig.mouse_cell(state, pane, 4, 1, 'move', button=button, mods=SHIFT)
rig.mouse_cell(state, pane, 4, 1, 'up', button=button, mods=SHIFT)
expected = (f'\033[<{button};1;2M\033[<{button + 36};5;2M'
f'\033[<{button + 4};5;2m').encode()
bytes_after(received, offset, expected, 'application modified button drag did not reach its PTY')
# A terminal mode change cancels a held app gesture but sends one
# release in the original SGR format, so the application cannot keep
# its button state stuck. Subsequent pointer events stay local until
# the application enables reporting again.
offset = received.stat().st_size
rig.mouse_cell(state, pane, 0, 1, 'down')
bytes_after(received, offset, b'\033[<0;1;2M', 'application press before mode change missing')
with control.open('a') as out:
out.write('off\n')
rig.wait_state(lambda s: 'MODE-OFF' in by_id(s)[pane]['painted_text'])
eventually(lambda: received.read_bytes()[offset:] == b'\033[<0;1;2M\033[<0;1;2m',
'mode change did not release the held application button')
rig.mouse_cell(state, pane, 4, 1, 'up')
time.sleep(.08)
require(received.read_bytes()[offset:] == b'\033[<0;1;2M\033[<0;1;2m',
'stale release after mode cancellation reached the application')
with control.open('a') as out:
out.write('on\n')
rig.wait_state(lambda s: 'MODE-ON' in by_id(s)[pane]['painted_text'])
# Shift latches a local selection even while the application has mouse
# reporting. It must not append any input to the terminal application.
offset = received.stat().st_size
rig.mouse_cell(state, pane, 0, 1, 'down', mods=SHIFT)
rig.mouse_cell(state, pane, 4, 1, 'move')
rig.mouse_cell(state, pane, 4, 1, 'up')
rig.wait_desktop('alpha')
time.sleep(.12)
require(received.stat().st_size == offset, 'Shift local drag leaked application mouse bytes')
# Capture stays with the original pane. A move over another tile must
# report the source pane's clamped edge, never a neighbour coordinate.
state = rig.state()
source = by_id(state)[pane]
other_content = by_id(state)[other]['content']
endpoint = rig.cell_point(state, other, 4, 1)
end_x = other_content['x'] + 4.5 * state['cell_w']
end_y = other_content['y'] + 1.5 * state['cell_h']
offset = received.stat().st_size
rig.mouse_cell(state, pane, 0, 1, 'down')
rig.mouse('move', endpoint)
rig.mouse('up', endpoint)
edge_x = max(0, min(source['cols'] - 1, int((end_x - source['content']['x']) / state['cell_w']))) + 1
edge_y = max(0, min(source['rows'] - 1, int((end_y - source['content']['y']) / state['cell_h']))) + 1
expected = (f'\033[<0;1;2M\033[<32;{edge_x};{edge_y}M\033[<0;{edge_x};{edge_y}m').encode()
bytes_after(received, offset, expected, 'cross-pane capture changed application source or missed release')
# OSC 52 C owns desktop clipboard; P and S update primary only. An
# unsafe decoded NUL and an unsupported target leave prior ownership.
for command, primary in (('c', None), ('p', 'PRIMARY-P'), ('s', 'PRIMARY-S')):
with control.open('a') as out:
out.write(command + '\n')
if command == 'c':
rig.wait_desktop('DESKTOP-C')
else:
rig.wait_primary(primary)
require(rig.desktop() == 'DESKTOP-C', f'OSC 52 {command} overwrote desktop clipboard')
with control.open('a') as out:
out.write('bad\nq\n')
time.sleep(.15)
require(rig.desktop() == 'DESKTOP-C' and rig.primary() == 'PRIMARY-S',
'invalid or unsupported OSC 52 write changed clipboard ownership')
rig.ok('application mouse bytes, Shift local drag, source capture, and OSC 52 targets')
def main():
require(len(sys.argv) == 3, 'usage: native_mouse.py MUX MUXG')
rig = MouseRig(*sys.argv[1:])
try:
refs = start_workspace(rig)
panes = list(refs)
application_drag(rig, panes[1], panes[2])
rig.kernel_sizes()
rig.assert_cli_untouched()
rig.quit()
print('Native mouse acceptance passed', flush=True)
except BaseException:
rig.failure_artifacts()
raise
finally:
rig.close()
if __name__ == '__main__':
main()