test/native_selection.py
Ref: Size: 25.0 KiB History
#!/usr/bin/env python3
"""Real drag events, daemon extraction, framebuffer and desktop clipboard oracles."""
import os
import shlex
import signal
import subprocess
import sys
import time
sys.dont_write_bytecode = True
from native_lifecycle import LifecycleRig, start_workspace
from native_resize import by_id
from native_theme import pixel
from native_tiling import eventually, require
class SelectionRig(LifecycleRig):
pointer = None
def send(self, *lines):
if self.env['SDL_VIDEO_DRIVER'] != 'wayland':
return super().send(*lines)
from wayland_pointer import Pointer
for line in lines:
kind, _, coords = line.partition(':')
if kind in ('mousedown', 'mousemove', 'mouseup', 'click'):
if self.pointer is None:
binary = os.environ.get('MUXG_TEST_POINTER')
require(binary, 'Wayland clipboard acceptance requires MUXG_TEST_POINTER')
self.pointer = Pointer(binary, self.env, self.gui.pid)
subprocess.run(['swaymsg', f'[pid={self.gui.pid}] focus'], env=self.env,
capture_output=True, check=True, timeout=3)
self.pointer.move(*map(float, coords.split(',')))
self.state() # Deliver initial window focus before testing a pane click.
self.pointer.event(kind, *map(float, coords.split(',')))
else:
super().send(line)
def close(self, success=None):
try:
if self.pointer is not None:
self.pointer.close()
self.pointer = None
finally:
super().close(success=success)
def clipboard(self):
value = self.artifact('clipboard', '.txt').read_text()
if self.env['SDL_VIDEO_DRIVER'] == 'wayland':
# A separate client asks the compositor, not the application hook.
external = subprocess.run(['wl-paste', '--no-newline'], env=self.env,
capture_output=True, timeout=3)
self.clipboard_observed = {'sdl': value, 'desktop': external.stdout.decode(),
'error': external.stderr.decode()}
# SDL publishes asynchronously to the compositor. Poll for both
# clients to agree instead of treating offer propagation as failure.
if external.returncode or external.stdout.decode() != value:
return None
return value
def set_clipboard(self, text):
self.serial += 1
source = self.root / (str(self.serial) + '-clipboard-source.txt')
source.write_text(text)
self.send('clipboard-set:' + str(source))
# The state artifact is ordered after the setter in the same FIFO.
self.state()
eventually(lambda: self.clipboard() == text,
'SDL clipboard setter did not publish ' + repr(text[:80]))
def cell_point(self, state, pane_id, col, row):
content = by_id(state)[pane_id]['content']
return self.point(state, content['x'] + (col + .5) * state['cell_w'],
content['y'] + (row + .5) * state['cell_h'])
def select(self, pane_id, start, finish, release=True):
state = self.state()
a, b = (self.cell_point(state, pane_id, *cell) for cell in (start, finish))
self.send('mousedown:' + a, 'mousemove:' + b)
if release:
self.send('mouseup:' + b)
return state
def copied(self, text):
try:
eventually(lambda: self.clipboard() == text, 'clipboard did not become ' + repr(text))
except AssertionError as error:
raise AssertionError(f'{error}; observed {getattr(self, "clipboard_observed", None)}') from error
def unchanged(self, text):
# Observe over several event-loop turns so a queued reply has a chance
# to arrive; a single immediate equality would miss a late overwrite.
until = time.monotonic() + .35
while time.monotonic() < until:
require(self.clipboard() == text, 'cancelled selection changed clipboard')
time.sleep(.025)
def specimen(rig, pane_id, tag):
rig.focus(pane_id)
pane = by_id(rig.state())[pane_id]
require(pane['cols'] >= 22 and pane['rows'] >= 9, 'selection specimen needs 22x9 cells')
wrap = 'W' * pane['cols'] + 'RAP'
marker = 'READY-' + tag + '-' + str(rig.serial)
output = ('\033[0m\033[2J\033[H' + tag + '\033[2;1Halpha café 界 omega'
'\033[3;1Hhard one\r\nhard two '
'\033[5;1H' + wrap + '\033[8;1H' + marker + '\033[9;1H')
escaped = output.replace('\033', '\\033').replace('\r', '\\r').replace('\n', '\\n')
rig.shell("export PS1=''; printf '%b' " + shlex.quote(escaped))
rig.wait_state(lambda s: marker in by_id(s)[pane_id]['painted_text'])
return wrap
def cell_background(rig, state, pane_id, col, row):
rect = by_id(state)[pane_id]['content']
# Cell corner avoids glyph ink and samples actual completed painted state.
return pixel(rig.last_pixels(), rect['x'] + col * state['cell_w'] + 1,
rect['y'] + row * state['cell_h'] + 1)
def arm_output(rig, pane_id, label, row=7):
"""Shell-owned output released by a file, with no input during the drag."""
trigger = rig.root / label
rig.focus(pane_id)
rig.shell('(while ! test -e ' + shlex.quote(str(trigger)) +
"; do sleep .02; done; printf '\\033[" + str(row) + ";1H" + label + "') &")
return trigger
def copy_shortcut(rig, pane):
"""Observe copy and SIGINT independently in a real foreground PTY process."""
rig.focus(pane)
interrupted, stop = (rig.root / name for name in ('copy-sigint', 'copy-stop'))
program = rig.root / 'copy-foreground.py'
program.write_text(
'import signal, time\nfrom pathlib import Path\n'
f'interrupted = Path({str(interrupted)!r})\nstop = Path({str(stop)!r})\n'
'def on_interrupt(signum, frame):\n'
' with interrupted.open("a") as out: out.write("INT\\n")\n'
'signal.signal(signal.SIGINT, on_interrupt)\n'
'print("\\033[0m\\033[2J\\033[HCOPY-READY\\033[2;1Halpha café omega\\033[4;1H", end="", flush=True)\n'
'while not stop.exists(): time.sleep(.02)\n')
rig.shell('python3 ' + shlex.quote(str(program)))
state = rig.wait_state(lambda s: by_id(s)[pane]['painted_text'].startswith('COPY-READY'))
background = cell_background(rig, state, pane, 1, 1)
try:
rig.select(pane, (6, 1), (9, 1))
rig.copied('café')
# Copy before release proves the shortcut invoked the copy path itself.
rig.select(pane, (0, 1), (4, 1), release=False)
eventually(lambda: cell_background(rig, state, pane, 1, 1) != background,
'held selection was not painted before copy shortcut')
rig.key('copy')
rig.copied('alpha')
eventually(lambda: cell_background(rig, state, pane, 1, 1) != background,
'copy shortcut cleared held selection')
rig.send('mouseup:' + rig.cell_point(state, pane, 4, 1))
rig.key('copy')
rig.unchanged('alpha')
require(not interrupted.exists(), 'copy shortcut sent SIGINT to foreground process')
eventually(lambda: cell_background(rig, state, pane, 1, 1) != background,
'copy shortcut cleared released selection')
rig.focus(pane) # Header press clears the range.
eventually(lambda: cell_background(rig, state, pane, 1, 1) == background,
'header press did not clear selection before copy shortcut')
rig.key('copy')
rig.unchanged('alpha')
require(not interrupted.exists(), 'copy without selection sent SIGINT')
rig.key('interrupt')
eventually(lambda: interrupted.exists() and interrupted.read_text() == 'INT\n',
'ordinary Ctrl+C did not interrupt the foreground PTY process')
rig.unchanged('alpha')
rig.ok('Ctrl+Shift+C copies and retains selection without SIGINT; plain Ctrl+C still interrupts')
finally:
stop.touch()
specimen(rig, pane, 'PANE-' + str(pane))
def start_paste_reader(rig, pane):
"""Run a raw foreground application which records exact pasted bytes."""
source = rig.root / 'paste-foreground.py'
received = rig.root / 'paste-received.bin'
control = rig.root / 'paste-control'
source.write_text(
'import os, select, sys, termios, tty\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[?2004l\\033[2J\\033[HPASTE-READY"); sys.stdout.flush()\n'
' seen=0; running=True\n'
' while running:\n'
' if control.exists():\n'
' commands=control.read_text()[seen:]; seen += len(commands)\n'
' for command in commands.splitlines():\n'
' if command == "on":\n'
' sys.stdout.write("\\033[?2004h\\033[2;1HPASTE-MODE-ON "); sys.stdout.flush()\n'
' elif command == "off":\n'
' sys.stdout.write("\\033[?2004l\\033[2;1HPASTE-MODE-OFF"); sys.stdout.flush()\n'
' elif command == "stop": running=False\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'
' print("\\033[?2004l\\r\\nPASTE-DONE", flush=True)\n')
rig.focus(pane)
rig.shell('python3 ' + shlex.quote(str(source)))
rig.wait_state(lambda state: 'PASTE-READY' in by_id(state)[pane]['painted_text'])
return received, control
def append_control(path, command):
with path.open('a') as out:
out.write(command + '\n')
def pasted_bytes(received, offset, expected, message):
eventually(lambda: received.exists() and received.stat().st_size >= offset + len(expected),
message)
actual = received.read_bytes()[offset:]
require(actual == expected, f'{message}: got {actual!r}, expected {expected!r}')
def paste_shortcut(rig, pane):
received, control = start_paste_reader(rig, pane)
try:
raw = 'raw first line\nsecond café 界'
rig.set_clipboard(raw)
rig.key('paste')
pasted_bytes(received, 0, raw.encode(), 'plain clipboard bytes did not reach the focused PTY')
append_control(control, 'on')
rig.wait_state(lambda state: 'PASTE-MODE-ON' in by_id(state)[pane]['painted_text'])
bracketed = 'bracketed\ntext café 界'
rig.set_clipboard(bracketed)
offset = received.stat().st_size
rig.key('paste')
pasted_bytes(received, offset, b'\033[200~' + bracketed.encode() + b'\033[201~',
'bracketed clipboard bytes were not wrapped exactly once')
large = ('0123456789abcdef' * 2300) + '\nlarge café 界'
require(len(large.encode()) > 32 * 1024, 'large paste fixture did not cross a pump chunk')
rig.set_clipboard(large)
offset = received.stat().st_size
rig.key('paste')
pasted_bytes(received, offset, b'\033[200~' + large.encode() + b'\033[201~',
'chunked bracketed paste lost, duplicated or rewrapped bytes')
rig.ok('native paste sends raw, Unicode, multiline and one chunked bracket envelope')
finally:
append_control(control, 'stop')
rig.wait_state(lambda state: 'PASTE-DONE' in by_id(state)[pane]['painted_text'])
specimen(rig, pane, 'PANE-' + str(pane))
def scale_selection(rig, pane):
output = os.environ.get('MUXG_TEST_SCALE_OUTPUT')
if not output:
return
require(rig.pointer is not None, 'scale acceptance requires real Wayland input')
original = next(o['scale'] for o in rig.pointer.query('get_outputs') if o['name'] == output)
def scale(value):
response = subprocess.run(['swaymsg', '-r', 'output', output, 'scale', str(value)],
env=rig.env, capture_output=True, text=True, check=True, timeout=3)
require('"success": true' in response.stdout, 'compositor refused fixture scale')
return rig.wait_state(lambda s: abs(s['width'] / s['logical_width'] - value) < .01)
try:
for value in (1, 1.5, 2):
specimen(rig, pane, 'SCALE-' + str(value))
rig.select(pane, (0, 1), (4, 1))
rig.copied('alpha')
rig.select(pane, (6, 1), (9, 1), release=False)
state = scale(value)
rig.send('mouseup:' + rig.cell_point(state, pane, 9, 1))
rig.unchanged('alpha')
specimen(rig, pane, 'SCALED-' + str(value))
rig.select(pane, (6, 1), (9, 1))
rig.copied('café')
rig.kernel_sizes()
finally:
scale(original)
rig.ok('real held drags cancel at 100/150/200% transitions; new drags copy and PTYs agree')
def active_output(rig, pane):
"""A real PTY keeps repainting while selection and clipboard are observed."""
rig.focus(pane)
stop = rig.root / 'counter-stop'
program = rig.root / 'selection-counter.py'
program.write_text(
'import os, time\nfrom pathlib import Path\n'
f'stop=Path({str(stop)!r})\n'
'os.write(1,b"\\033[?25l\\033[2J\\033[HCOUNTER-READY")\n'
'n=0\n'
'while not stop.exists():\n'
' os.write(1,("\\033[2;1Halpha café omega\\033[2;21HCOUNT-%08d" % n).encode())\n'
' n+=1; time.sleep(.025)\n'
'print("\\r\\nCOUNTER-DONE",flush=True)\n')
tmux_socket = rig.root / 'counter-tmux.sock'
nested = os.environ.get('MUXG_TEST_TMUX') == '1'
command = 'python3 ' + shlex.quote(str(program))
if nested:
command = ('tmux -S ' + shlex.quote(str(tmux_socket)) +
' -f /dev/null new-session -s counter ' + shlex.quote(command))
rig.shell(command)
def count(state):
text = by_id(state)[pane]['painted_text']
return int(text.split('COUNT-', 1)[1][:8]) if 'COUNT-' in text else -1
rig.wait_state(lambda state: count(state) >= 0)
wayland = rig.env['SDL_VIDEO_DRIVER'] == 'wayland'
windowed = rig.state()
original_size = (windowed['logical_width'], windowed['logical_height'])
try:
for fullscreen in ((False, True) if wayland else (False,)):
if wayland:
subprocess.run(['swaymsg', f'[pid={rig.gui.pid}] fullscreen ' +
('enable' if fullscreen else 'disable')],
env=rig.env, capture_output=True, check=True, timeout=3)
rig.wait_state(lambda state: ((state['logical_width'], state['logical_height']) !=
original_size) == fullscreen)
# Let resize output settle while proving the application still runs.
initial = count(rig.state())
state = rig.wait_state(lambda state: count(state) >= initial + 8)
background = cell_background(rig, state, pane, 1, 1)
rig.select(pane, (0, 1), (4, 1), release=False)
initial = count(rig.state())
rig.wait_state(lambda state: count(state) >= initial + 12)
eventually(lambda: cell_background(rig, state, pane, 1, 1) != background,
'active counter cleared the held selection')
rig.key('copy')
rig.copied('alpha')
# A second copy during the same held gesture must use the newly
# extended range. Release then registers its final, shorter range.
rig.send('mousemove:' + rig.cell_point(state, pane, 6, 1))
# Wayland motion and the FIFO keyboard hook arrive independently.
# Observe the extended highlight before asking the other channel to copy.
eventually(lambda: cell_background(rig, rig.state(), pane, 6, 1) != background,
'held drag did not visibly extend before copying')
rig.key('copy')
rig.copied('alpha c')
rig.send('mouseup:' + rig.cell_point(state, pane, 4, 1))
rig.copied('alpha')
initial = count(rig.state())
rig.wait_state(lambda state: count(state) >= initial + 12)
eventually(lambda: cell_background(rig, state, pane, 1, 1) != background,
'active counter cleared the released highlight')
rig.key('copy')
rig.copied('alpha')
rig.ok('live PTY counter keeps advancing during held/released selection and copy' +
(' through tmux' if nested else '') +
(' in windowed and fullscreen Wayland' if wayland else ''))
finally:
stop.touch()
if nested:
subprocess.run(['tmux', '-S', str(tmux_socket), 'kill-server'],
env=rig.env, capture_output=True, timeout=3)
if wayland:
subprocess.run(['swaymsg', f'[pid={rig.gui.pid}] fullscreen disable'],
env=rig.env, capture_output=True, timeout=3)
if wayland:
rig.wait_state(lambda state: (state['logical_width'], state['logical_height']) == original_size)
if nested:
rig.wait_state(lambda state: count(state) < 0)
else:
rig.wait_state(lambda state: 'COUNTER-DONE' in by_id(state)[pane]['painted_text'])
def selection_basics(rig, refs):
"""Nondestructive clipboard and pointer checks on the shared workspace."""
panes = list(refs)
target, neighbour = panes[1], panes[2]
copy_shortcut(rig, target)
paste_shortcut(rig, target)
wraps = {pane: specimen(rig, pane, 'PANE-' + str(pane)) for pane in panes}
rig.focus(panes[0])
state = rig.state()
before = {p: cell_background(rig, state, p, 1, 1) for p in panes}
rig.select(target, (0, 1), (4, 1), release=False)
eventually(lambda: cell_background(rig, state, target, 1, 1) != before[target],
'drag did not visibly highlight the off-origin pane')
for p in (panes[0], neighbour):
require(cell_background(rig, state, p, 1, 1) == before[p], 'highlight leaked into neighbour')
require(rig.state()['focus'] == target, 'press did not focus selected pane')
rig.send('mouseup:' + rig.cell_point(state, target, 4, 1))
rig.copied('alpha')
rig.ok('off-origin held drag paints only its pane and release copies daemon text')
for pane in panes:
rig.select(pane, (4, 1), (0, 1))
rig.copied('alpha')
rig.select(pane, (6, 1), (10, 1))
rig.copied('café')
# Endpoint on the wide glyph continuation must include the whole glyph.
rig.select(pane, (10, 1), (12, 1))
rig.copied(' 界')
rig.select(pane, (0, 2), (10, 3))
rig.copied('hard one\nhard two')
rig.select(pane, (0, 4), (2, 5))
rig.copied(wraps[pane])
rig.ok('both directions, Unicode, wide continuation, trimming, hard newline and soft wrap')
baseline = wraps[panes[-1]]
state = rig.state()
same = rig.cell_point(state, target, 0, 1)
rig.send('mousedown:' + same, 'mousemove:' + same, 'mouseup:' + same)
rig.unchanged(baseline)
rig.focus(neighbour) # Header press is deliberately not terminal text.
rig.unchanged(baseline)
rig.chord('enter')
rig.picker('hosts')
rig.select(target, (0, 1), (4, 1))
dismissed = rig.wait_state(lambda s: not s.get('picker'))
require(dismissed['focus'] == neighbour,
'outside picker gesture focused the covered terminal pane')
rig.unchanged(baseline)
rig.key('escape')
rig.wait_state(lambda s: s['pending'] is None)
rig.ok('same-cell tremor, header press and modal pointer input preserve clipboard')
# Start in one pane, enter its neighbour, then release back over the origin.
# The neighbour must not supply either the endpoint or copied text.
state = rig.select(target, (0, 1), (4, 1), release=False)
rig.send('mousemove:' + rig.cell_point(state, neighbour, 16, 1),
'mouseup:' + rig.cell_point(state, target, 4, 1))
rig.copied('alpha')
rig.ok('cross-pane motion keeps the original session selection')
trigger = arm_output(rig, neighbour, 'NEIGHBOUR-OUTPUT')
specimen(rig, neighbour, 'PANE-' + str(neighbour))
background = cell_background(rig, rig.state(), target, 7, 1)
state = rig.select(target, (6, 1), (9, 1), release=False)
eventually(lambda: cell_background(rig, state, target, 7, 1) != background,
'selection was not painted before neighbouring output')
trigger.touch()
rig.wait_state(lambda s: 'NEIGHBOUR-OUTPUT' in by_id(s)[neighbour]['painted_text'])
rig.send('mouseup:' + rig.cell_point(state, target, 9, 1))
rig.copied('café')
trigger = arm_output(rig, target, 'SELECTED-OUTPUT', row=2)
specimen(rig, target, 'PANE-' + str(target))
background = cell_background(rig, rig.state(), target, 1, 1)
state = rig.select(target, (0, 1), (4, 1), release=False)
eventually(lambda: cell_background(rig, state, target, 1, 1) != background,
'selection was not painted before selected-pane output')
trigger.touch()
rig.wait_state(lambda s: 'SELECTED-OUTPUT' in by_id(s)[target]['painted_text'])
rig.send('mouseup:' + rig.cell_point(state, target, 4, 1))
rig.copied('SELEC')
rig.ok('redraw preserves selection; overwriting selected text copies the current range')
rig.shell('wait') # Reap the fixture writer before painting the next specimen.
specimen(rig, target, 'PANE-' + str(target))
rig.select(target, (0, 1), (4, 1))
rig.copied('alpha')
def selection_races(rig, refs):
"""Real paused-daemon replies, scale cancellation, and destructive detach."""
panes = list(refs)
target, neighbour = panes[1], panes[2]
active_output(rig, target)
specimen(rig, target, 'RACES')
rig.select(target, (0, 1), (4, 1))
rig.copied('alpha')
# Pause only this fixture's real daemon. The queued selection is still
# decoded/extracted by that daemon when resumed, not by a protocol mock.
sock = refs[target][0]
daemon = next(proc for path, proc in rig.daemons if path == sock)
daemon.send_signal(signal.SIGSTOP)
try:
rig.select(target, (6, 1), (9, 1))
rig.state() # Event queue barrier, no repaint or data repair.
rig.focus(neighbour)
finally:
daemon.send_signal(signal.SIGCONT)
rig.unchanged('alpha')
rig.ok('a delayed real-daemon reply cannot copy after a new press clears selection')
# Resize while a reply is waiting; then validate the real PTY sizes.
daemon.send_signal(signal.SIGSTOP)
try:
rig.select(target, (6, 1), (9, 1))
rig.state()
rig.drag('beside', dx=rig.state()['cell_w'] * -2)
rig.state() # Process the resize events before allowing the reply.
finally:
daemon.send_signal(signal.SIGCONT)
rig.unchanged('alpha')
rig.kernel_sizes()
rig.ok('geometry change cancels pending copy and all PTYs match the resized panes')
specimen(rig, target, 'PANE-' + str(target))
state = rig.state()
blank = cell_background(rig, state, target, 16, 8)
rig.select(target, (15, 8), (18, 8), release=False)
eventually(lambda: cell_background(rig, state, target, 16, 8) != blank,
'empty-copy fixture did not select its in-bounds blank cells')
rig.send('mouseup:' + rig.cell_point(state, target, 18, 8))
rig.state()
rig.unchanged('alpha')
daemon.send_signal(signal.SIGSTOP)
try:
rig.select(target, (6, 1), (9, 1))
rig.wait_state(lambda s: 'Selection unavailable' in s['notice'])
finally:
daemon.send_signal(signal.SIGCONT)
rig.unchanged('alpha')
rig.ok('empty copy and a real unanswered request preserve clipboard; timeout is visible')
scale_selection(rig, target)
specimen(rig, target, 'DETACH')
rig.select(target, (0, 1), (4, 1))
rig.copied('alpha')
daemon.send_signal(signal.SIGSTOP)
try:
rig.select(target, (6, 1), (9, 1))
rig.state()
rig.chord('d')
rig.wait_state(lambda s: target not in by_id(s))
finally:
daemon.send_signal(signal.SIGCONT)
rig.unchanged('alpha')
require(rig.status(*refs[target])['cols'] > 0, 'detach ended the selected shell')
rig.ok('detach cancels copying with a stalled daemon and leaves its shell alive')
rig.assert_cli_untouched()
return refs
def main():
require(len(sys.argv) == 3, 'usage: native_selection.py MUX MUXG')
rig = SelectionRig(*sys.argv[1:])
try:
refs = start_workspace(rig)
# Offscreen basics belong to the journey; Wayland also checks the
# compositor clipboard through an independent wl-paste client.
if rig.env['SDL_VIDEO_DRIVER'] == 'wayland':
selection_basics(rig, refs)
selection_races(rig, refs)
rig.quit()
print('Native selection acceptance passed', flush=True)
except BaseException:
rig.failure_artifacts()
raise
finally:
for _, proc in rig.daemons:
if proc.poll() is None:
proc.send_signal(signal.SIGCONT)
rig.close()
if __name__ == '__main__':
main()