Andrew Hamilton
fec8266154
Two end-to-end tests driving the focus-based wl_data_device path with wl-copy/wl-paste: nested->host (exercises bridge_host_set_selection and the focus-serial defer/retry) and host->nested (exercises bridge_dd_selection and the text prefetch mirror). Exact-byte matching with unique tokens and a poll deadline; skipped when wl-clipboard tools are absent.
348 lines
No EOL
12 KiB
Python
348 lines
No EOL
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Smoke tests for tabbed-wlroots.
|
|
|
|
The compositor is launched as a nested display (child of the host
|
|
Wayland session) and driven over the stdin event feed: raw keyboard
|
|
events are packed into binary records and dispatched into the *real*
|
|
host key handlers (host_kbd_key / host_kbd_modifiers), so the key
|
|
handler logic is under test without needing a focused window or wtype.
|
|
|
|
Each test launches its own instance (function-scoped fixture), so tests
|
|
are independent and can run in any order.
|
|
|
|
Requires: a live host Wayland session (WAYLAND_DISPLAY) and a prior
|
|
./build.sh. Missing prereqs skip the suite.
|
|
"""
|
|
|
|
import glob
|
|
import os
|
|
import re
|
|
import shutil
|
|
import struct
|
|
import subprocess
|
|
import time
|
|
|
|
import pytest
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
BIN = os.path.join(ROOT, "build", "tabbed-wlroots")
|
|
WL = os.path.join(ROOT, "build", "wltest")
|
|
LOG = "/tmp/tabtest.log"
|
|
RED = "#ff0000"
|
|
BLUE = "#0000ff"
|
|
|
|
# Raw keycodes used by the compositor's own key handler (see
|
|
# host_kbd_key in tabbed-wlroots.c). Names, not numbers, in tests.
|
|
F9 = 67
|
|
F10 = 68
|
|
F11 = 87
|
|
PAGE_UP = 104
|
|
PAGE_DOWN = 109
|
|
KEY_PRESSED = 1
|
|
KEY_RELEASED = 0
|
|
|
|
# Ctrl modifier bit, as read from host_mods_depressed (1u << 2).
|
|
CTRL = 1 << 2
|
|
|
|
CTL_CMD_KEY = 1
|
|
CTL_CMD_MODS = 2
|
|
|
|
ACTIVE_RE = re.compile(r"active app=(\d+)")
|
|
SOCKET_RE = re.compile(r"Wayland socket: (tabbed-wayland-\d+)")
|
|
|
|
|
|
class Session:
|
|
"""A running tabbed-wlroots instance plus helpers."""
|
|
|
|
def __init__(self, proc):
|
|
self.proc = proc
|
|
self._socket = None
|
|
|
|
# -- log watching ----------------------------------------------------
|
|
def read_log(self):
|
|
"""Whole current log contents (rescanned fresh each call)."""
|
|
with open(LOG, "r", errors="replace") as f:
|
|
return f.read()
|
|
|
|
def wait_for(self, pattern, timeout=15.0):
|
|
"""Wait until `pattern` matches somewhere in the whole log.
|
|
|
|
Re-scans from the start each poll: tabbed's log lines can arrive
|
|
out of order (e.g. `map app=1` before `map app=0`), so a line may
|
|
enter the file before we happen to be looking for it.
|
|
"""
|
|
rx = re.compile(pattern)
|
|
since = time.monotonic()
|
|
while time.monotonic() - since < timeout:
|
|
if rx.search(self.read_log()) is not None:
|
|
return
|
|
time.sleep(0.1)
|
|
tail = self.read_log()[-1000:]
|
|
pytest.fail(f"log never matched {rx.pattern};\n{tail}")
|
|
|
|
@property
|
|
def socket(self):
|
|
if self._socket is None:
|
|
self.wait_for(SOCKET_RE)
|
|
m = SOCKET_RE.search(self.read_log())
|
|
assert m, "no 'Wayland socket:' line in log"
|
|
self._socket = m.group(1)
|
|
return self._socket
|
|
|
|
# -- input -----------------------------------------------------------
|
|
def key(self, code, state):
|
|
"""Feed one key event into the compositor's real key handler."""
|
|
record = struct.pack("<BII", CTL_CMD_KEY, code, state)
|
|
self.proc.stdin.write(record)
|
|
self.proc.stdin.flush()
|
|
|
|
def mods(self, pattern):
|
|
"""Feed a keyboard-modifier set (e.g. CTRL, or 0 to clear)."""
|
|
record = struct.pack("<BIIII", CTL_CMD_MODS, pattern, 0, 0, 0)
|
|
self.proc.stdin.write(record)
|
|
self.proc.stdin.flush()
|
|
|
|
def tap(self, code):
|
|
"""Press then release one key."""
|
|
self.key(code, KEY_PRESSED)
|
|
self.key(code, KEY_RELEASED)
|
|
|
|
def chord(self, mods, *keys):
|
|
"""Hold `mods`, tap each key in turn, then release `mods`."""
|
|
self.mods(pattern=mods)
|
|
for code in keys:
|
|
self.tap(code)
|
|
self.mods(pattern=0)
|
|
|
|
def cycle(self, direction):
|
|
"""Swap to the next (direction>0) or previous tab via Ctrl+Page."""
|
|
page = PAGE_DOWN if direction > 0 else PAGE_UP
|
|
self.chord(CTRL, page)
|
|
|
|
def active(self):
|
|
"""Current active app index from the log, or None."""
|
|
m = ACTIVE_RE.findall(self.read_log())
|
|
return int(m[-1]) if m else None
|
|
|
|
def wait_active(self, want, timeout=3.0):
|
|
since = time.monotonic()
|
|
while time.monotonic() - since < timeout:
|
|
if self.active() == want:
|
|
return
|
|
time.sleep(0.1)
|
|
pytest.fail(f"never reached active app={want}")
|
|
|
|
def wait_active_change(self, away_from, timeout=3.0):
|
|
"""Wait for active to differ from `away_from`; return the new value.
|
|
|
|
Unlike wait_active(), this ignores historical log lines: it only
|
|
returns once a *new* active tab shows up, so a stale launch-time
|
|
line can't satisfy it before the key event is processed.
|
|
"""
|
|
since = time.monotonic()
|
|
while time.monotonic() - since < timeout:
|
|
cur = self.active()
|
|
if cur is not None and cur != away_from:
|
|
return cur
|
|
time.sleep(0.1)
|
|
pytest.fail(f"active app never changed away from {away_from}")
|
|
|
|
def wait_any_active(self, timeout=10.0):
|
|
"""Wait until the active tab has settled, return its index.
|
|
|
|
The two test clients map in nondeterministic order, so the first
|
|
`active app=` line can name the first-mapped tab before the other
|
|
maps. Only once *both* tabs have logged an active line is the
|
|
last one the settled active tab.
|
|
"""
|
|
rx = re.compile(r"active app=(\d+)")
|
|
since = time.monotonic()
|
|
while time.monotonic() - since < timeout:
|
|
seen = set(rx.findall(self.read_log()))
|
|
if {"0", "1"}.issubset(seen):
|
|
return self.active()
|
|
time.sleep(0.1)
|
|
pytest.fail("both tabs never logged an active line")
|
|
|
|
# -- frame dumps ------------------------------------------------------
|
|
def dump_host(self, timeout=5.0):
|
|
rx = re.compile(r"\[dump\] host back buffer -> (\S+\.ppm)")
|
|
before = len(rx.findall(self.read_log()))
|
|
self.tap(F10)
|
|
# Wait for a *new* "host back buffer -> ...ppm" line: it is logged
|
|
# only after the file is fully written and closed. The file
|
|
# itself appears at fopen() while still mid-write, so file
|
|
# appearance alone is not enough.
|
|
since = time.monotonic()
|
|
while time.monotonic() - since < timeout:
|
|
m = rx.findall(self.read_log())
|
|
if len(m) > before:
|
|
return m[-1]
|
|
time.sleep(0.1)
|
|
raise AssertionError("F10 produced no 'host back buffer ->' line")
|
|
|
|
def centre_of(self, path):
|
|
"""Centre pixel of an 8-bit P6 ppm as an (R, G, B) tuple."""
|
|
with open(path, "rb") as f:
|
|
data = f.read()
|
|
lines = data.split(b"\n", 3)
|
|
if len(lines) < 3:
|
|
raise AssertionError(f"bad PPM {path}")
|
|
w, h = map(int, lines[1].split())
|
|
off = (len(lines[0]) + 1 + len(lines[1]) + 1 + len(lines[2]) + 1)
|
|
cx, cy = w // 2, h // 2
|
|
i = off + 3 * (cy * w + cx)
|
|
return tuple(data[i:i + 3])
|
|
|
|
def close(self):
|
|
if self.proc.poll() is None:
|
|
self.proc.terminate()
|
|
try:
|
|
self.proc.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
self.proc.kill()
|
|
self.proc.wait()
|
|
for ppm in glob.glob("/tmp/host_*.ppm"):
|
|
os.unlink(ppm)
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def session():
|
|
if not os.environ.get("WAYLAND_DISPLAY"):
|
|
pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
|
|
for path in (BIN, WL):
|
|
if not os.access(path, os.X_OK):
|
|
pytest.skip(f"build first: {path}")
|
|
try:
|
|
os.unlink(LOG)
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
proc = subprocess.Popen(
|
|
["env", "TABBED_LOG=2", "stdbuf", "-oL",
|
|
BIN, f"{WL} '{RED}'", f"{WL} '{BLUE}'"],
|
|
stdin=subprocess.PIPE,
|
|
stdout=open(LOG, "ab"), stderr=subprocess.STDOUT,
|
|
)
|
|
s = Session(proc)
|
|
try:
|
|
yield s
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
# -- clipboard bridge (GNOME/Mutter focus-based wl_data_device backend) -----
|
|
|
|
def _run_display(display, args, **kw):
|
|
"""Run a command against a specific Wayland display, capturing output."""
|
|
env = dict(os.environ)
|
|
env["WAYLAND_DISPLAY"] = display
|
|
return subprocess.run(
|
|
args, env=env, capture_output=True, text=True,
|
|
stdin=subprocess.DEVNULL, timeout=10, **kw)
|
|
|
|
|
|
def _poll_paste(display, want, deadline=10.0):
|
|
"""Wait until `wl-paste` on `display` returns exactly `want`."""
|
|
r = None
|
|
since = time.monotonic()
|
|
while time.monotonic() - since < deadline:
|
|
r = _run_display(display, ["wl-paste", "-n"])
|
|
if r.stdout == want:
|
|
return
|
|
time.sleep(0.2)
|
|
raise AssertionError(
|
|
f"wl-paste on {display} never returned {want!r}; "
|
|
f"last={r.stdout!r} err={r.stderr!r}")
|
|
|
|
|
|
NEED_WL_CLIPBOARD = shutil.which("wl-copy") is None or shutil.which("wl-paste") is None
|
|
|
|
|
|
@pytest.mark.skipif(NEED_WL_CLIPBOARD, reason="wl-clipboard tools not installed")
|
|
def test_clipboard_nested_to_host(session):
|
|
token = f"tabbed-n2h-{os.getpid()}-{time.monotonic_ns()}"
|
|
copy = subprocess.Popen(
|
|
["wl-copy", "-f", token],
|
|
env={**os.environ, "WAYLAND_DISPLAY": session.socket},
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
try:
|
|
_poll_paste(os.environ.get("WAYLAND_DISPLAY", "wayland-0"), token)
|
|
finally:
|
|
copy.terminate()
|
|
copy.wait(timeout=5)
|
|
|
|
|
|
@pytest.mark.skipif(NEED_WL_CLIPBOARD, reason="wl-clipboard tools not installed")
|
|
def test_clipboard_host_to_nested(session):
|
|
token = f"tabbed-h2n-{os.getpid()}-{time.monotonic_ns()}"
|
|
copy = subprocess.Popen(
|
|
["wl-copy", "-f", token],
|
|
env=dict(os.environ),
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
try:
|
|
_poll_paste(session.socket, token)
|
|
finally:
|
|
copy.terminate()
|
|
copy.wait(timeout=5)
|
|
|
|
|
|
def test_tabs_map(session):
|
|
session.wait_for(r"\[diag\] map app=0")
|
|
session.wait_for(r"\[diag\] map app=1")
|
|
|
|
|
|
def test_cycle_tabs(session):
|
|
start = session.wait_any_active()
|
|
assert start in (0, 1)
|
|
session.cycle(+1)
|
|
other = session.wait_active_change(start)
|
|
assert other == 1 - start
|
|
session.cycle(-1)
|
|
assert session.wait_active_change(other) == start
|
|
|
|
|
|
def test_host_colour_tracks_tab(session):
|
|
# warm up the present chain, then find each tab's colour
|
|
session.tap(F10)
|
|
session.wait_for(r"\[dump\] host")
|
|
start = session.wait_any_active()
|
|
assert start in (0, 1)
|
|
|
|
host_ppm = session.dump_host()
|
|
c0 = session.centre_of(host_ppm)
|
|
|
|
session.cycle(+1)
|
|
other = session.wait_active_change(start)
|
|
assert other == 1 - start
|
|
host_ppm = session.dump_host()
|
|
c1 = session.centre_of(host_ppm)
|
|
|
|
assert c0 != c1, f"host colour did not change across tabs ({c0} vs {c1})"
|
|
|
|
session.cycle(-1)
|
|
assert session.wait_active_change(other) == start
|
|
host_ppm = session.dump_host()
|
|
cb = session.centre_of(host_ppm)
|
|
assert cb == c0, f"host colour did not return to start ({cb} vs {c0})"
|
|
|
|
|
|
def test_f11_toggles_fullscreen(session):
|
|
session.tap(F11)
|
|
session.wait_for("host fullscreen=1")
|
|
session.tap(F11)
|
|
session.wait_for("host fullscreen=0")
|
|
assert session.proc.poll() is None
|
|
|
|
|
|
def test_sigterm_removes_socket(session):
|
|
sock_path = os.path.join(
|
|
os.environ.get("XDG_RUNTIME_DIR", f"/run/user/{os.getuid()}"),
|
|
session.socket,
|
|
)
|
|
session.proc.terminate()
|
|
session.proc.wait(timeout=5)
|
|
assert session.proc.returncode == 0
|
|
assert not os.path.exists(sock_path), f"socket left behind: {sock_path}" |