a73x

test/native_wheel.py

Ref:   Size: 15.7 KiB   History

#!/usr/bin/env python3
"""Wheel input through real SDL/Wayland, history pixels and independent PTY bytes."""
import os
from pathlib import Path
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, one_divider
from native_selection import SelectionRig, cell_background
from native_tiling import eventually, require


class WheelRig(SelectionRig):
    def wheel_at(self, point, delta, flipped=False, native=True):
        if self.env['SDL_VIDEO_DRIVER'] == 'wayland' and native and not flipped and int(delta) == delta:
            # Initialize the established compositor input adapter without a click.
            self.send('mousemove:' + point)
            self.pointer.wheel(*map(float, point.split(',')), int(delta))
        else:
            self.send(f'wheel:{point},{delta}' + (',flipped' if flipped else ''))

    def wheel(self, pane, delta, col=4, row=3, **kwargs):
        state = self.state()
        self.wheel_at(self.cell_point(state, pane, col, row), delta, **kwargs)


def painted(rig, pane):
    return by_id(rig.state())[pane]['painted_text']


def history(rig, pane):
    rig.focus(pane)
    # Known numbered rows span many screens; full markers occur only in output.
    program = rig.root / f'history-{pane}.py'
    program.write_text('import sys\n'
                       'sys.stdout.write("\\033[?25l\\033[2J\\033[H")\n'
                       'for n in range(300): print(f"HISTORY-{n:04d} alpha café")\n'
                       'print("WHEEL-" + "LIVE", flush=True)\n')
    rig.shell("export PS1=''; python3 " + shlex.quote(str(program)))
    rig.wait_state(lambda s: 'WHEEL-LIVE' in by_id(s)[pane]['painted_text'])
    return painted(rig, pane)


def first_number(text):
    first = text.splitlines()[0]
    require(first.startswith('HISTORY-'), 'expected a numbered history row, got ' + repr(first))
    return int(first[8:12])


def shell_history(rig, panes):
    live = {pane: history(rig, pane) for pane in panes}
    target, other, focused = panes[1], panes[0], panes[2]
    rig.focus(focused)
    baseline = first_number(live[target])
    rig.wheel(target, 1)
    state = rig.wait_state(lambda s: first_number(by_id(s)[target]['painted_text']) == baseline - 3)
    require(state['focus'] == focused, 'wheel moved keyboard focus')
    for pane in (other, focused):
        require(by_id(state)[pane]['painted_text'] == live[pane], 'wheel changed another pane')
    # Inspect completed framebuffer pixels as well as the passive text snapshot.
    before_pixels = rig.last_pixels()[2]
    rig.wheel(target, 1)
    rig.wait_state(lambda s: first_number(by_id(s)[target]['painted_text']) == baseline - 6)
    eventually(lambda: rig.last_pixels()[2] != before_pixels, 'history text did not change framebuffer')
    rig.ok('off-origin unfocused pane scrolls three rows per notch without changing neighbours or focus')

    state = rig.state()
    background = cell_background(rig, state, other, 1, 0)
    selected = by_id(state)[other]['painted_text'].splitlines()[0][:12]
    rig.select(other, (0, 0), (11, 0))
    rig.copied(selected)
    eventually(lambda: cell_background(rig, state, other, 1, 0) != background,
               'selection did not paint before scrolling a neighbour')
    rig.wheel(target, 1)
    rig.wait_state(lambda s: first_number(by_id(s)[target]['painted_text']) == baseline - 9)
    require(cell_background(rig, state, other, 1, 0) != background,
            'scrolling another pane cleared the selection')
    rig.unchanged(selected)
    rig.ok('scrolling a neighbouring pane preserves selected text and its highlight')

    rig.wheel(target, -1000)
    rig.wait_state(lambda s: by_id(s)[target]['painted_text'] == live[target])
    rig.wheel(target, .5, native=False)
    rig.wheel(other, .5, native=False)
    require(painted(rig, target) == live[target] and painted(rig, other) == live[other],
            'fractional notches leaked between panes')
    rig.wheel(target, .5, native=False)
    rig.wait_state(lambda s: first_number(by_id(s)[target]['painted_text']) == baseline - 3)
    require(painted(rig, other) == live[other], 'target consumed another pane remainder')
    rig.wheel(other, -.5, native=False)  # cancel its remainder
    rig.wheel(target, 1, flipped=True)
    rig.wait_state(lambda s: by_id(s)[target]['painted_text'] == live[target])
    rig.ok('fractional wheel events accumulate per pane and flipped direction returns to live')

    rig.wheel(target, 1000)
    oldest = rig.wait_state(lambda s: 'HISTORY-0000' in by_id(s)[target]['painted_text'])
    oldest_text = by_id(oldest)[target]['painted_text']
    rig.wheel(target, 1000)
    require(painted(rig, target) == oldest_text, 'wheel moved beyond oldest history')
    rig.wheel(target, -1000)
    rig.wait_state(lambda s: by_id(s)[target]['painted_text'] == live[target])
    rig.wheel(target, 4)
    state = rig.wait_state(lambda s: first_number(by_id(s)[target]['painted_text']) == baseline - 12)
    first = by_id(state)[target]['painted_text'].splitlines()[0]
    background = cell_background(rig, state, target, 1, 0)
    rig.select(target, (0, 0), (11, 0))
    rig.copied(first[:12])
    eventually(lambda: cell_background(rig, state, target, 1, 0) != background,
               'history selection did not paint')
    rig.wheel(target, 1)
    rig.wait_state(lambda s: first_number(by_id(s)[target]['painted_text']) == baseline - 15)
    eventually(lambda: cell_background(rig, state, target, 1, 3) != background and
               cell_background(rig, state, target, 1, 0) == background,
               'highlight did not follow its text three rows down')
    rig.wheel(target, 1000)
    rig.wait_state(lambda s: 'HISTORY-0000' in by_id(s)[target]['painted_text'])
    eventually(lambda: all(cell_background(rig, state, target, 1, row) == background
                          for row in range(by_id(state)[target]['rows'])),
               'offscreen selection left a highlight in the viewport')
    rig.wheel(target, -1000)
    rig.wait_state(lambda s: by_id(s)[target]['painted_text'] == live[target])
    rig.wheel(target, 4)
    rig.wait_state(lambda s: first_number(by_id(s)[target]['painted_text']) == baseline - 12)
    eventually(lambda: cell_background(rig, state, target, 1, 0) != background,
               'highlight did not return with selected history text')
    rig.key('copy')
    rig.unchanged(first[:12])
    rig.ok('released history highlight follows text, survives offscreen/live round trips and remains copyable')

    # Keep the original anchor while a held drag moves through history. Release
    # at its new screen row to copy the same source row, then repeat with an
    # explicit motion to extend the selection across the newly visible rows.
    held = by_id(rig.state())[target]['painted_text'].splitlines()[1]
    rig.select(target, (0, 1), (11, 1), release=False)
    rig.wheel(target, 1, col=11, row=1)
    rig.wait_state(lambda s: first_number(by_id(s)[target]['painted_text']) == baseline - 15)
    rig.send('mouseup:' + rig.cell_point(state, target, 11, 4))
    rig.copied(held[:12])
    rig.select(target, (0, 0), (11, 0), release=False)
    rig.wheel(target, 1, col=11, row=0)
    moved = rig.wait_state(lambda s: first_number(by_id(s)[target]['painted_text']) == baseline - 18)
    lines = by_id(moved)[target]['painted_text'].splitlines()
    rig.send('mousemove:' + rig.cell_point(moved, target, 11, 4))
    rig.key('copy')
    rig.copied(lines[3] + '\n' + lines[4][:12])
    rig.send('mouseup:' + rig.cell_point(moved, target, 11, 4))
    rig.copied(lines[3] + '\n' + lines[4][:12])
    rig.ok('held drag retains its text anchor and extends using the scrolled pane coordinates')

    rig.wheel(target, -1000)
    rig.wait_state(lambda s: by_id(s)[target]['painted_text'] == live[target])
    rig.select(target, (0, 0), (11, 0))
    rig.copied(live[target].splitlines()[0][:12])
    rig.wheel(target, 1)
    rig.wait_state(lambda s: first_number(by_id(s)[target]['painted_text']) == baseline - 3)
    eventually(lambda: cell_background(rig, state, target, 1, 3) != background,
               'live selection disappeared on entering history')
    rig.wheel(target, -1)
    rig.wait_state(lambda s: by_id(s)[target]['painted_text'] == live[target])
    eventually(lambda: cell_background(rig, state, target, 1, 0) != background,
               'live selection disappeared on returning from history')
    rig.shell("printf 'RETURN-%s\\n' LIVE")
    rig.wait_state(lambda s: 'RETURN-LIVE' in by_id(s)[target]['painted_text'])
    eventually(lambda: cell_background(rig, state, target, 1, 0) == background,
               'input failed to clear the selection')
    rig.ok('live selection survives history navigation; typing returns live and clears it')


def raw_reader(rig, pane):
    """A real foreground PTY application changes modes and records all received bytes."""
    rig.focus(pane)
    control, received, stopped = (rig.root / f'{name}-{pane}' for name in ('mode', 'received', 'stop'))
    program = rig.root / f'raw-reader-{pane}.py'
    program.write_text(
        'import os, select, termios, tty\nfrom pathlib import Path\n'
        f'control=Path({str(control)!r}); received=Path({str(received)!r}); stop=Path({str(stopped)!r})\n'
        'old=termios.tcgetattr(0); tty.setraw(0); previous=None\n'
        'reset="\\033[?9l\\033[?1000l\\033[?1002l\\033[?1003l\\033[?1005l\\033[?1006l\\033[?1015l\\033[?1016l\\033[?1l\\033[?1049l"\n'
        'try:\n'
        ' with received.open("wb", buffering=0) as output:\n'
        '  while not stop.exists():\n'
        '   current=control.read_text() if control.exists() else "ready|"\n'
        '   if current != previous:\n'
        '    tag, modes=current.split("|", 1)\n'
        '    os.write(1, (reset+modes+"\\033[2J\\033[HMODE-"+tag+"\\r\\n").encode()); previous=current\n'
        '   if select.select([0], [], [], .02)[0]: output.write(os.read(0, 4096))\n'
        'finally:\n'
        ' os.write(1, reset.encode()); termios.tcsetattr(0, termios.TCSANOW, old)\n')
    rig.shell('python3 ' + shlex.quote(str(program)))
    rig.wait_state(lambda s: by_id(s)[pane]['painted_text'].startswith('MODE-ready'))
    return control, received, stopped


def application_wheel(rig, panes):
    pane, focus = panes[1], panes[2]
    control, received, stopped = raw_reader(rig, pane)
    def set_mode(value):
        pending = control.with_suffix('.next')
        pending.write_text(value)
        pending.replace(control)
    rig.focus(focus)
    offset = 0
    try:
        cases = [('arrows', '\033[?1049h', b'\033[A' * 3, b'\033[B' * 3),
                 ('app-arrows', '\033[?1049h\033[?1h', b'\033OA' * 3, b'\033OB' * 3),
                 ('sgr', '\033[?1000h\033[?1006h', b'\033[<64;5;4M', b'\033[<65;5;4M'),
                 ('legacy', '\033[?1000h', b'\033[M`%$', b'\033[Ma%$'),
                 ('utf8', '\033[?1000h\033[?1005h', b'\033[M`%$', b'\033[Ma%$'),
                 ('urxvt', '\033[?1000h\033[?1015h', b'\033[96;5;4M', b'\033[97;5;4M')]
        for label, modes, up, down in cases:
            set_mode(label + '|' + modes)
            rig.wait_state(lambda s: by_id(s)[pane]['painted_text'].startswith('MODE-' + label))
            rig.wheel(pane, 1)
            rig.wheel(pane, -1)
            expected = up + down
            eventually(lambda: received.stat().st_size >= offset + len(expected), label + ' PTY bytes missing')
            time.sleep(.08)
            actual = received.read_bytes()[offset:]
            require(actual == expected, f'{label} PTY received {actual!r}, expected {expected!r}')
            offset += len(expected)
            require(rig.state()['focus'] == focus, 'application wheel changed keyboard focus')
        rig.ok('independent PTY bytes prove alternate arrows, cursor-key mode and negotiated cell mouse formats')

        set_mode('pixels|\033[?1000h\033[?1016h')
        rig.wait_state(lambda s: by_id(s)[pane]['painted_text'].startswith('MODE-pixels'))
        scale_output = os.environ.get('MUXG_TEST_SCALE_OUTPUT')
        original = None
        scales = (None,)
        if scale_output:
            original = next(o['scale'] for o in rig.pointer.query('get_outputs') if o['name'] == scale_output)
            scales = (2, 1, 1.5, 2)
        try:
            for scale in scales:
                if scale is not None:
                    subprocess.run(['swaymsg', 'output', scale_output, 'scale', str(scale)],
                                   env=rig.env, capture_output=True, check=True, timeout=3)
                    rig.wait_state(lambda s: abs(s['width'] / s['logical_width'] - scale) < .01)
                    rig.kernel_sizes()
                state = rig.state()
                point = rig.cell_point(state, pane, 4, 3)
                x, y = map(float, point.split(','))
                if rig.env['SDL_VIDEO_DRIVER'] == 'wayland':
                    x, y = int(x), int(y)  # virtual pointer uses logical integer coordinates
                content = by_id(state)[pane]['content']
                px = int(x * state['width'] / state['logical_width']) - content['x'] + 1
                py = int(y * state['height'] / state['logical_height']) - content['y'] + 1
                rig.wheel_at(point, 1)
                expected = f'\033[<64;{px};{py}M'.encode()
                eventually(lambda: received.stat().st_size >= offset + len(expected), 'pixel report missing')
                require(received.read_bytes()[offset:] == expected, 'pixel report not relative to pane at current DPI')
                offset += len(expected)
        finally:
            if original is not None:
                subprocess.run(['swaymsg', 'output', scale_output, 'scale', str(original)],
                               env=rig.env, capture_output=True, check=True, timeout=3)
        rig.ok('SGR pixel coordinates are pane-relative, including configured Wayland scale transitions')

        rig.key('prefix')
        rig.wheel(pane, 1)
        time.sleep(.1)
        require(received.stat().st_size == offset, 'command prefix leaked wheel input')
        rig.key('escape')
        rig.chord('enter')
        rig.picker()
        rig.wheel(pane, 1)
        time.sleep(.1)
        require(received.stat().st_size == offset, 'picker leaked wheel input')
        rig.key('escape')
        rig.wait_state(lambda s: not s.get('picker'))
        rig.chord('p')
        rig.wait_state(lambda s: s.get('recovery'))
        rig.wheel(pane, 1)
        time.sleep(.1)
        require(received.stat().st_size == offset, 'recovery menu leaked wheel input')
        rig.key('escape')
        state = rig.wait_state(lambda s: not s.get('recovery'))
        header = by_id(state)[pane]['header']
        rig.wheel_at(rig.point(state, header['x'] + header['w']/2, header['y'] + header['h']/2), 1)
        rect = one_divider(state, 'beside')['rect']
        point = rig.point(state, rect['x'] + rect['w']/2, rect['y'] + rect['h']/4)
        rig.wheel_at(point, 1)
        rig.send('mousedown:' + point)
        rig.wheel(pane, 1)
        rig.send('mouseup:' + point)
        time.sleep(.1)
        require(received.stat().st_size == offset, 'header/divider/resize leaked wheel input')
        rig.ok('picker, recovery, headers, dividers and held resize intercept wheel events')
    finally:
        stopped.touch()


def main():
    rig = WheelRig(*sys.argv[1:3])
    try:
        refs = start_workspace(rig)
        panes = list(refs)
        if rig.env['SDL_VIDEO_DRIVER'] == 'wayland':
            shell_history(rig, panes)
        application_wheel(rig, panes)
        rig.kernel_sizes()
        rig.assert_cli_untouched()
        rig.quit()
        rig.ok('three real PTYs retain geometry, persistent identities and terminal layout state')
        print('Native application wheel acceptance passed', flush=True)
    except Exception:
        rig.failure_artifacts()
        raise
    finally:
        rig.close()


if __name__ == '__main__':
    main()