Andrew Hamilton
13154c78b5
Commands fork one at a time, each gated on the previous one's client connecting, so tab order always mirrors the command-line order (no pid matching, no /proc ancestry). The window stays black until the last command's tab maps, which is then crowned active. A new tab never steals focus on its own: only Ctrl+Shift+T terminals, sub-windows of the tab currently shown, and the startup crown activate. Sub-windows (a second toplevel from an already-connected client) get their own slot glued right after their parent tab. A command that never connects stalls the chain and the crown, leaving the window black (accepted). Tests: wait_any_active waits for the single deterministic crown; an externally-spawned client is now born hidden (cycle to it). Add wltest --extra (same-client second toplevel) plus tests for sub-window placement/focus and Ctrl+Shift+T focus-on-map. Update README.
1739 lines
67 KiB
Python
1739 lines
67 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 ctypes
|
|
import glob
|
|
import os
|
|
import re
|
|
import shutil
|
|
import signal
|
|
import struct
|
|
import subprocess
|
|
import tempfile
|
|
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"
|
|
GREEN = "#00ff00"
|
|
|
|
# Raw keycodes used by the compositor's own key handler (see
|
|
# host_kbd_key in tabbed-wlroots.c). Names, not numbers, in tests.
|
|
F = 33
|
|
F11 = 87
|
|
UP = 103
|
|
DOWN = 108
|
|
DELETE = 111
|
|
KEY_PRESSED = 1
|
|
KEY_RELEASED = 0
|
|
|
|
# Ctrl modifier bit, as read from host_mods_depressed (1u << 2).
|
|
CTRL = 1 << 2
|
|
# Alt (Mod1) modifier bit (1u << 3).
|
|
ALT = 1 << 3
|
|
# Shift modifier bit (1u << 0).
|
|
SHIFT = 1 << 0
|
|
# Super/Mod4 modifier bit (1u << 6).
|
|
SUPER = 1 << 6
|
|
|
|
CTL_CMD_KEY = 1
|
|
CTL_CMD_MODS = 2
|
|
CTL_CMD_WARP = 3
|
|
CTL_CMD_PTR_MOTION = 4
|
|
|
|
ACTIVE_RE = re.compile(r"active app=(\d+)")
|
|
CLAIM_RE = re.compile(r"\[claim\] app=(\d+) pid=(\d+)")
|
|
SOCKET_RE = re.compile(r"Wayland socket: (tabbed-wayland-\d+)")
|
|
|
|
|
|
# Font paths the compositor searches (overlay_pick_font in tabbed-wlroots.c);
|
|
# used to make the standalone headless session build a real glyph atlas.
|
|
FONT_PATHS = [
|
|
"/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
|
|
"/usr/share/fonts/dejavu/DejaVuSans.ttf",
|
|
"/usr/share/fonts/TTF/DejaVuSans.ttf",
|
|
"/usr/share/fonts/truetype/liberation/LiberationSans-Regular.ttf",
|
|
"/usr/share/fonts/truetype/liberation2/LiberationSans-Regular.ttf",
|
|
"/usr/share/fonts/liberation/LiberationSans-Regular.ttf",
|
|
"/usr/share/fonts/noto/NotoSans-Regular.ttf",
|
|
]
|
|
|
|
|
|
def _pick_font():
|
|
"""First existing font path, from $TABBED_FONT or the default list."""
|
|
env_font = os.environ.get("TABBED_FONT")
|
|
if env_font and os.path.isfile(env_font):
|
|
return env_font
|
|
for path in FONT_PATHS:
|
|
if os.path.isfile(path):
|
|
return path
|
|
return None
|
|
|
|
|
|
_libc = ctypes.CDLL(None)
|
|
_PR_SET_PDEATHSIG = 1
|
|
|
|
|
|
def _preexec_death(expected_ppid):
|
|
"""preexec_fn: have the kernel SIGTERM this child if the test process
|
|
(its parent) ever dies, SIGKILL included. The compositor's SIGTERM handler
|
|
tears the whole session down (clients, socket), and the short-lived
|
|
wl-copy/wl-paste helpers just die -- so an aborted pytest run can never
|
|
leave strays behind. Check getppid() after setting it: if the parent died
|
|
between fork() and prctl() the signal is already pending, but our
|
|
expectations (and the child) are meaningless anyway, so bail out."""
|
|
def _fn():
|
|
_libc.prctl(_PR_SET_PDEATHSIG, signal.SIGTERM)
|
|
if os.getppid() != expected_ppid:
|
|
os._exit(1)
|
|
return _fn
|
|
|
|
|
|
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, timeout=60.0)
|
|
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 warp(self, x, y):
|
|
"""Move the standalone cursor to an output pixel (x, y)."""
|
|
record = struct.pack("<BII", CTL_CMD_WARP, x, y)
|
|
self.proc.stdin.write(record)
|
|
self.proc.stdin.flush()
|
|
|
|
def ptr_motion(self, dx, dy):
|
|
"""Feed a relative pointer-motion delta into the same path libinput
|
|
pointer devices take (wlr_cursor_move + seat forward)."""
|
|
record = struct.pack("<BII", CTL_CMD_PTR_MOTION, dx, dy)
|
|
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 Alt+arrow."""
|
|
key = DOWN if direction > 0 else UP
|
|
self.chord(ALT, key)
|
|
|
|
def cycle_ack(self, direction, timeout=5.0):
|
|
"""Tap Alt+arrow and block until the compositor logs a *new*
|
|
active-tab line reflecting the cycle, then return the fresh value.
|
|
|
|
cycle() alone races: `active()` reads the log file, which lags the
|
|
compositor's async key processing, so a tight `while active() != N`
|
|
loop can queue extra cycles past the target before its reads catch
|
|
up. Waiting for the log to grow guarantees the tap took effect
|
|
before the caller makes its next decision.
|
|
|
|
The count is only trustworthy once the fixture tabs have stopped
|
|
emitting their own startup "active app=" lines (both clients map
|
|
asynchronously, toggling active as they land); otherwise one of
|
|
those can satisfy the "grew" check and hand back a stale value.
|
|
So quiesce the log before tapping.
|
|
"""
|
|
key = DOWN if direction > 0 else UP
|
|
since = time.monotonic()
|
|
before = None
|
|
while time.monotonic() - since < timeout:
|
|
before = len(ACTIVE_RE.findall(self.read_log()))
|
|
time.sleep(0.2)
|
|
if len(ACTIVE_RE.findall(self.read_log())) == before:
|
|
break
|
|
self.chord(ALT, key)
|
|
since = time.monotonic()
|
|
while time.monotonic() - since < timeout:
|
|
if len(ACTIVE_RE.findall(self.read_log())) > before:
|
|
return self.active()
|
|
time.sleep(0.05)
|
|
pytest.fail("cycle tap never reflected in the log")
|
|
|
|
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 startup crown has activated a tab, return its index.
|
|
|
|
Only the last command-line command is ever auto-activated on boot
|
|
(the "crown"); every earlier tab maps hidden with no `active app=`
|
|
line, so the first active line is the settled tab -- there is no
|
|
churn to wait out. For the fixtures' N-tab sessions the crown is
|
|
always index N-1.
|
|
"""
|
|
rx = re.compile(r"active app=(\d+)")
|
|
since = time.monotonic()
|
|
while time.monotonic() - since < timeout:
|
|
m = rx.findall(self.read_log())
|
|
if m:
|
|
return int(m[-1])
|
|
time.sleep(0.1)
|
|
pytest.fail("no tab was ever activated (startup crown never fired)")
|
|
|
|
def wait_app_active(self, index, timeout=10.0):
|
|
"""Wait until the single app `index` reports itself active (for
|
|
sessions launched with fewer than two tabs)."""
|
|
rx = re.compile(r"active app=%d\b" % index)
|
|
since = time.monotonic()
|
|
while time.monotonic() - since < timeout:
|
|
if rx.search(self.read_log()):
|
|
return True
|
|
time.sleep(0.1)
|
|
pytest.fail(f"app {index} never logged an active line")
|
|
|
|
# -- frame dumps ------------------------------------------------------
|
|
def dump_host(self, timeout=5.0):
|
|
rx = re.compile(r"\[dump\] host back buffer -> (\S+\.ppm)")
|
|
# The SIGUSR1 handler is only installed once the compositor is well
|
|
# into startup; signaling too early kills it (default action), so
|
|
# wait until the socket line exists first.
|
|
self.socket
|
|
before = len(rx.findall(self.read_log()))
|
|
os.kill(self.proc.pid, signal.SIGUSR1)
|
|
# 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("SIGUSR1 produced no 'host back buffer ->' line")
|
|
|
|
def dump_scene(self, timeout=5.0):
|
|
"""SIGUSR1-request a scene dump; wait until the dump path has run.
|
|
|
|
The scene buffer is a dmabuf and never CPU-readable, so no ppm is
|
|
written; the `[dump] scene` line (either "not CPU-readable" or the
|
|
stats line) proves the request reached the dump path.
|
|
"""
|
|
rx = re.compile(r"\[dump\] scene")
|
|
self.socket
|
|
before = len(rx.findall(self.read_log()))
|
|
os.kill(self.proc.pid, signal.SIGUSR1)
|
|
since = time.monotonic()
|
|
while time.monotonic() - since < timeout:
|
|
m = rx.findall(self.read_log())
|
|
if len(m) > before:
|
|
return
|
|
time.sleep(0.1)
|
|
raise AssertionError("SIGUSR1 produced no '[dump] scene' line")
|
|
|
|
def active_count(self):
|
|
"""How many `active app=` transitions the log has seen so far."""
|
|
return len(ACTIVE_RE.findall(self.read_log()))
|
|
|
|
def launch_pids(self, timeout=10.0):
|
|
"""Map app index -> connected client PID from [claim] lines."""
|
|
since = time.monotonic()
|
|
while time.monotonic() - since < timeout:
|
|
d = dict((int(a), int(p)) for a, p in CLAIM_RE.findall(self.read_log()))
|
|
if {0, 1}.issubset(d):
|
|
return d
|
|
time.sleep(0.1)
|
|
pytest.fail("both apps never logged a [claim] line")
|
|
|
|
def pixel_at(self, path, x, y):
|
|
"""Pixel (x, y) of an 8-bit P6 ppm as an (R, G, B) tuple.
|
|
|
|
The dump writes the back buffer top-down (window coords), so row 0
|
|
of the file is the top of the window and column 0 is the left edge.
|
|
"""
|
|
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)
|
|
i = off + 3 * (y * w + x)
|
|
return tuple(data[i:i + 3])
|
|
|
|
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())
|
|
return self.pixel_at(path, w // 2, h // 2)
|
|
|
|
def observed_host_colour(self, want, timeout=2.0):
|
|
"""Poll SIGUSR1 dumps until the host centre pixel equals `want`.
|
|
|
|
Returns True if the colour was observed, False on timeout. Never
|
|
fails: the host readback races the EGL present/swap path, and when
|
|
the host compositor stops pacing frames the back buffer can sit
|
|
frozen on an old tab indefinitely, so the caller decides whether an
|
|
unobservable pixel is a real failure.
|
|
"""
|
|
since = time.monotonic()
|
|
while time.monotonic() - since < timeout:
|
|
path = self.dump_host()
|
|
if path and self.centre_of(path) == want:
|
|
return True
|
|
time.sleep(0.1)
|
|
return False
|
|
|
|
def ppm_rgb(self, path):
|
|
"""(width, height, raw RGB bytes) of an 8-bit P6 ppm.
|
|
|
|
Reading once lets a test scan many pixels without reopening the
|
|
file for every sample.
|
|
"""
|
|
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)
|
|
return w, h, data[off:]
|
|
|
|
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") + glob.glob("/tmp/scene_*.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", "TABBED_SYNTH_INPUT=1", "stdbuf", "-oL",
|
|
BIN, f"{WL} '{RED}'", f"{WL} '{BLUE}'"],
|
|
stdin=subprocess.PIPE,
|
|
stdout=open(LOG, "ab"), stderr=subprocess.STDOUT,
|
|
preexec_fn=_preexec_death(os.getpid()),
|
|
)
|
|
s = Session(proc)
|
|
try:
|
|
yield s
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def standalone_session():
|
|
"""Standalone mode on a headless backend, exercised over the stdin feed.
|
|
|
|
The native backend autocreate is bypassed with WLR_BACKENDS=headless (real
|
|
DRM is unavailable in the test environment), so this checks the standalone
|
|
render/input/present paths without a host window. Requires a GPU renderer
|
|
(PIXMAN/GBlob every time CPU backends), which the build always links.
|
|
"""
|
|
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
|
|
|
|
# Clear any leftover XDG/compositor interaction: standalone must not
|
|
# connect to the host compositor (it would pick the wayland backend and
|
|
# open a second nested window). WLR_BACKENDS=headless forces the
|
|
# headless backend and the standalone output path.
|
|
env = dict(os.environ)
|
|
env["TABBED_LOG"] = "2"
|
|
env["TABBED_MODE"] = "standalone"
|
|
env["WLR_BACKENDS"] = "headless"
|
|
env.pop("WAYLAND_DISPLAY", None)
|
|
proc = subprocess.Popen(
|
|
["env", "-i",
|
|
"PATH=" + env["PATH"],
|
|
"HOME=" + env.get("HOME", "/"),
|
|
"XDG_RUNTIME_DIR=" + env.get("XDG_RUNTIME_DIR", "/tmp"),
|
|
"XDG_DATA_HOME=" + env.get("XDG_DATA_HOME", ""),
|
|
"XDG_DATA_DIRS=" + env.get("XDG_DATA_DIRS", ""),
|
|
"TABBED_LOG=2",
|
|
"TABBED_MODE=standalone",
|
|
"WLR_BACKENDS=headless",
|
|
"TABBED_FONT=" + (env.get("TABBED_FONT") or _pick_font() or ""),
|
|
"XCURSOR_THEME=Adwaita",
|
|
"stdbuf", "-oL",
|
|
BIN, f"{WL} '{RED}'", f"{WL} '{BLUE}'"],
|
|
stdin=subprocess.PIPE,
|
|
stdout=open(LOG, "ab"), stderr=subprocess.STDOUT,
|
|
preexec_fn=_preexec_death(os.getpid()),
|
|
)
|
|
s = Session(proc)
|
|
try:
|
|
yield s
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def test_standalone_headless_boot_and_cycle(standalone_session):
|
|
"""Standalone on the headless backend boots, maps both tabs, and cycles
|
|
them with the same stdin-driven Alt+arrow handling as nested mode."""
|
|
s = standalone_session
|
|
assert s.proc.poll() is None, f"standalone exited early: {s.read_log()[-500:]}"
|
|
start = s.wait_any_active()
|
|
assert start in (0, 1)
|
|
s.cycle(+1)
|
|
assert s.wait_active_change(start) == 1 - start
|
|
assert s.proc.poll() is None
|
|
|
|
|
|
def _has_dark(w, h, px):
|
|
"""True when the central region looks like the dark overlay box."""
|
|
for y in range(h // 4, 3 * h // 4):
|
|
row = y * w * 3
|
|
for x in range(w // 4, 3 * w // 4):
|
|
i = row + x * 3
|
|
if px[i] < 40 and px[i + 1] < 40 and px[i + 2] < 40:
|
|
return True
|
|
return False
|
|
|
|
|
|
def test_standalone_headless_alt_overlay_scene(standalone_session):
|
|
"""Holding Alt in standalone mode routes through the shared hotkey path
|
|
and draws the overlay (which in standalone enters the second wlr render
|
|
pass before the frame is committed). SIGUSR1 dumps the composited frame,
|
|
whose centre darkens while Alt is held."""
|
|
s = standalone_session
|
|
s.wait_any_active()
|
|
|
|
# Before Alt: the frame should be the client content (no dark overlay).
|
|
w, h, px = s.ppm_rgb(s.dump_host())
|
|
assert not _has_dark(w, h, px), "content already dark before Alt"
|
|
|
|
s.mods(ALT)
|
|
deadline = time.monotonic() + 15.0
|
|
while time.monotonic() < deadline:
|
|
w, h, px = s.ppm_rgb(s.dump_host())
|
|
if _has_dark(w, h, px):
|
|
break
|
|
assert _has_dark(w, h, px), "standalone overlay box not visible while Alt"
|
|
|
|
s.mods(0)
|
|
assert s.proc.poll() is None
|
|
|
|
|
|
def test_standalone_headless_overlay_clears_on_alt_release(standalone_session):
|
|
"""Releasing Alt in standalone fully clears the overlay. The overlay is
|
|
drawn into the scene output buffer outside the scene graph, so without
|
|
forcing whole-output damage on the transition a reused swapchain buffer
|
|
keeps the stale overlay baked in and it reappears on alternating frames
|
|
(until a client repaint covers it)."""
|
|
s = standalone_session
|
|
s.wait_any_active()
|
|
|
|
s.mods(ALT)
|
|
deadline = time.monotonic() + 15.0
|
|
seen = False
|
|
while time.monotonic() < deadline:
|
|
w, h, px = s.ppm_rgb(s.dump_host())
|
|
if _has_dark(w, h, px):
|
|
seen = True
|
|
break
|
|
assert seen, "overlay did not appear while Alt"
|
|
|
|
s.mods(0)
|
|
deadline = time.monotonic() + 15.0
|
|
while time.monotonic() < deadline:
|
|
w, h, px = s.ppm_rgb(s.dump_host())
|
|
if not _has_dark(w, h, px):
|
|
break # a freshly presented, overlay-free frame
|
|
time.sleep(0.08)
|
|
else:
|
|
raise AssertionError("overlay remnant never cleared after Alt release")
|
|
|
|
# Once clear, keep sampling: any reappearing dark frame is a stale buffer
|
|
# slot echoing the baked-in overlay, i.e. the bug returning.
|
|
deadline = time.monotonic() + 2.0
|
|
while time.monotonic() < deadline:
|
|
w, h, px = s.ppm_rgb(s.dump_host())
|
|
assert not _has_dark(w, h, px), "overlay remnant reappeared after Alt release"
|
|
time.sleep(0.08)
|
|
assert s.proc.poll() is None
|
|
|
|
|
|
def test_standalone_headless_cursor_visible(standalone_session):
|
|
"""The standalone output shows a real cursor. The headless backend has no
|
|
cursor plane, so wlroots composites the software cursor into the frame
|
|
buffer (the scene render pass picks it up). Warping the pointer must
|
|
change the composited pixels at the target -- the cursor visibly moves
|
|
there -- while moving the cursor away again repaints the spot back to pure
|
|
client colour without smearing.
|
|
|
|
The GL dump is vertically flipped (bottom-up readback), so a cursor at
|
|
compositor pixel (x, y) appears in the PPM at (x, H-1-y); the patch probes
|
|
through the same flip."""
|
|
s = standalone_session
|
|
s.wait_any_active()
|
|
assert s.proc.poll() is None
|
|
|
|
def patch_at(path, cx, cy, r=22):
|
|
"""Sample the PPM around compositor pixel (cx, cy), mapping compositor
|
|
row y to file row H-1-y for the bottom-up readback."""
|
|
w, h, px = s.ppm_rgb(path)
|
|
py = h - 1 - cy
|
|
hits = []
|
|
for y in range(max(0, py - r), min(h, py + r + 1)):
|
|
row = y * w * 3
|
|
for x in range(max(0, cx - r), min(w, cx + r + 1)):
|
|
i = row + x * 3
|
|
hits.append((px[i], px[i + 1], px[i + 2]))
|
|
return hits
|
|
|
|
def distinct(a, b):
|
|
return any(ca != cb for ca, cb in zip(a, b))
|
|
|
|
target = (320, 320)
|
|
|
|
# Get the pointer clear of the probe area first, then snapshot: the
|
|
# target patch must be pure client colour before we warp into it.
|
|
s.warp(60, 60)
|
|
time.sleep(0.3)
|
|
before = patch_at(s.dump_host(), *target)
|
|
assert len(set(before)) == 1, \
|
|
"probe area should be pure client colour before the warp"
|
|
|
|
s.warp(*target)
|
|
deadline = time.monotonic() + 5.0
|
|
while time.monotonic() < deadline:
|
|
if distinct(patch_at(s.dump_host(), *target), before):
|
|
break
|
|
time.sleep(0.1)
|
|
else:
|
|
pytest.fail("cursor never became visible at the warp target")
|
|
|
|
# Move away again: the target must end up back on pure client colour
|
|
# (no cursor smeared into the composited buffer).
|
|
s.warp(640, 400)
|
|
deadline = time.monotonic() + 5.0
|
|
while time.monotonic() < deadline:
|
|
if not distinct(patch_at(s.dump_host(), *target), before):
|
|
break
|
|
time.sleep(0.1)
|
|
else:
|
|
pytest.fail("cursor left a remnant at the warp target")
|
|
|
|
assert s.proc.poll() is None
|
|
|
|
|
|
def test_standalone_headless_cursor_relative_motion(standalone_session):
|
|
"""Relative pointer motion must move the visible cursor, exactly as libinput
|
|
pointer devices do on a real host (native_ptr_motion -> wlr_cursor_move). A
|
|
regressed compositor leaves the pointer pinned at the warp target while the
|
|
seat still moves, so this fails on code that forwards motion but never moves
|
|
the cursor."""
|
|
s = standalone_session
|
|
s.wait_any_active()
|
|
assert s.proc.poll() is None
|
|
|
|
def patch_at(path, cx, cy, r=22):
|
|
w, h, px = s.ppm_rgb(path)
|
|
py = h - 1 - cy
|
|
hits = []
|
|
for y in range(max(0, py - r), min(h, py + r + 1)):
|
|
row = y * w * 3
|
|
for x in range(max(0, cx - r), min(w, cx + r + 1)):
|
|
i = row + x * 3
|
|
hits.append((px[i], px[i + 1], px[i + 2]))
|
|
return hits
|
|
|
|
def distinct(a, b):
|
|
return any(ca != cb for ca, cb in zip(a, b))
|
|
|
|
start = (320, 320)
|
|
delta = (200, 120)
|
|
moved = (start[0] + delta[0], start[1] + delta[1])
|
|
|
|
# Clear the pointer, snapshot both probe areas as pure client colour.
|
|
s.warp(40, 40)
|
|
time.sleep(0.3)
|
|
before_start = patch_at(s.dump_host(), *start)
|
|
before_moved = patch_at(s.dump_host(), *moved)
|
|
assert len(set(before_start)) == 1, \
|
|
"start probe area should be pure client colour"
|
|
assert len(set(before_moved)) == 1, \
|
|
"moved probe area should be pure client colour"
|
|
|
|
# Place the cursor at `start` and confirm it is painted there.
|
|
s.warp(*start)
|
|
deadline = time.monotonic() + 5.0
|
|
while time.monotonic() < deadline:
|
|
if distinct(patch_at(s.dump_host(), *start), before_start):
|
|
break
|
|
time.sleep(0.1)
|
|
else:
|
|
pytest.fail("cursor never became visible at the start target")
|
|
|
|
# Relative motion (+dx, +dy) must carry the pointer to `moved`: the pixels
|
|
# there change, and `start` returns to pure client colour.
|
|
s.ptr_motion(*delta)
|
|
deadline = time.monotonic() + 5.0
|
|
while time.monotonic() < deadline:
|
|
path = s.dump_host()
|
|
if distinct(patch_at(path, *moved), before_moved) and \
|
|
not distinct(patch_at(path, *start), before_start):
|
|
break
|
|
time.sleep(0.1)
|
|
else:
|
|
pytest.fail("relative motion did not move the cursor %+d,%+d" % delta)
|
|
|
|
assert s.proc.poll() is None
|
|
|
|
|
|
def test_standalone_headless_screen_mode_cursor_stays_aligned():
|
|
"""--screen-mode + pointer motion must keep the visible cursor exactly on
|
|
the warp target plus the relative delta, on a non-default output size. The
|
|
seat forward and the visible arrow must both derive from the same
|
|
wlr_cursor position (single source of truth), not a parallel hand-clamped
|
|
copy -- on a DRM monitor a requested mode that differs from the geometry
|
|
the layout derives from the committed state would otherwise let the two
|
|
split diagonally."""
|
|
s = _spawn_standalone({}, "--screen-mode", "480x320",
|
|
"build/wltest '#ff0000'")
|
|
try:
|
|
s.wait_app_active(0)
|
|
assert s.proc.poll() is None, f"log tail:\n{s.read_log()[-400:]}"
|
|
|
|
def patch_at(path, cx, cy, r=22):
|
|
w, h, px = s.ppm_rgb(path)
|
|
py = h - 1 - cy
|
|
hits = []
|
|
for y in range(max(0, py - r), min(h, py + r + 1)):
|
|
row = y * w * 3
|
|
for x in range(max(0, cx - r), min(w, cx + r + 1)):
|
|
i = row + x * 3
|
|
hits.append((px[i], px[i + 1], px[i + 2]))
|
|
return hits
|
|
|
|
def distinct(a, b):
|
|
return any(ca != cb for ca, cb in zip(a, b))
|
|
|
|
start = (120, 120)
|
|
delta = (100, 60)
|
|
moved = (start[0] + delta[0], start[1] + delta[1])
|
|
|
|
s.warp(30, 30)
|
|
time.sleep(0.3)
|
|
before_start = patch_at(s.dump_host(), *start)
|
|
before_moved = patch_at(s.dump_host(), *moved)
|
|
assert len(set(before_start)) == 1, \
|
|
"start probe area should be pure client colour"
|
|
assert len(set(before_moved)) == 1, \
|
|
"moved probe area should be pure client colour"
|
|
|
|
s.warp(*start)
|
|
deadline = time.monotonic() + 5.0
|
|
while time.monotonic() < deadline:
|
|
if distinct(patch_at(s.dump_host(), *start), before_start):
|
|
break
|
|
time.sleep(0.1)
|
|
else:
|
|
pytest.fail("cursor never became visible at the start target")
|
|
|
|
s.ptr_motion(*delta)
|
|
deadline = time.monotonic() + 5.0
|
|
while time.monotonic() < deadline:
|
|
path = s.dump_host()
|
|
if distinct(patch_at(path, *moved), before_moved) and \
|
|
not distinct(patch_at(path, *start), before_start):
|
|
break
|
|
time.sleep(0.1)
|
|
else:
|
|
pytest.fail("relative motion did not move the cursor %+d,%+d" % delta)
|
|
|
|
assert s.proc.poll() is None
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def _spawn_standalone(extra_env, *apps):
|
|
"""Standalone compositor on the headless backend with a custom env for
|
|
cursor-theme tests; XCURSOR_THEME/XCURSOR_PATH come only from extra_env."""
|
|
try:
|
|
os.unlink(LOG)
|
|
except FileNotFoundError:
|
|
pass
|
|
env = dict(os.environ)
|
|
env["TABBED_LOG"] = "2"
|
|
env["TABBED_MODE"] = "standalone"
|
|
env["WLR_BACKENDS"] = "headless"
|
|
env.pop("WAYLAND_DISPLAY", None)
|
|
env.pop("XCURSOR_THEME", None)
|
|
env.pop("XCURSOR_PATH", None)
|
|
env.update(extra_env or {})
|
|
cmd = ["env", "-i",
|
|
"PATH=" + env["PATH"],
|
|
"HOME=" + env.get("HOME", "/"),
|
|
"XDG_RUNTIME_DIR=" + env.get("XDG_RUNTIME_DIR", "/tmp"),
|
|
"XDG_DATA_HOME=" + env.get("XDG_DATA_HOME", ""),
|
|
"XDG_DATA_DIRS=" + env.get("XDG_DATA_DIRS", ""),
|
|
"TABBED_LOG=2",
|
|
"TABBED_MODE=standalone",
|
|
"WLR_BACKENDS=headless",
|
|
"TABBED_FONT=" + (env.get("TABBED_FONT") or _pick_font() or "")]
|
|
for key in ("XCURSOR_PATH", "XCURSOR_THEME"):
|
|
if env.get(key):
|
|
cmd.append("%s=%s" % (key, env[key]))
|
|
cmd += ["stdbuf", "-oL", BIN]
|
|
cmd += list(apps)
|
|
proc = subprocess.Popen(
|
|
cmd, stdin=subprocess.PIPE,
|
|
stdout=open(LOG, "ab"), stderr=subprocess.STDOUT,
|
|
preexec_fn=_preexec_death(os.getpid()))
|
|
return Session(proc)
|
|
|
|
|
|
def test_standalone_headless_cursor_theme_missing_uses_builtin(tmp_path):
|
|
"""A theme that does not exist anywhere must not stop the compositor:
|
|
wlroots substitutes a builtin arrow, so the pointer stays visible. We
|
|
merely log the substitution, naming the requested theme."""
|
|
empty = tmp_path / "empty-cursors"
|
|
empty.mkdir()
|
|
s = _spawn_standalone({"XCURSOR_PATH": str(empty),
|
|
"XCURSOR_THEME": "DefinitelyMissingTheme"},
|
|
"build/wltest '#ff0000'")
|
|
try:
|
|
s.wait_app_active(0)
|
|
assert s.proc.poll() is None, f"log tail:\n{s.read_log()[-400:]}"
|
|
log = s.read_log()
|
|
assert "theme 'DefinitelyMissingTheme' not found; wlroots builtin " \
|
|
"cursor in use" in log, \
|
|
f"substitution should be logged naming the theme:\n{log[-400:]}"
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def test_standalone_headless_cursor_theme_none_boots(tmp_path):
|
|
"""No theme requested and no cursor theme installed: the builtin arrow is
|
|
used silently (a request named literally 'default' is indistinguishable
|
|
from the builtin), and there must be no warning and no failure."""
|
|
empty = tmp_path / "empty-cursors"
|
|
empty.mkdir()
|
|
s = _spawn_standalone({"XCURSOR_PATH": str(empty)},
|
|
"build/wltest '#ff0000'")
|
|
try:
|
|
s.wait_app_active(0)
|
|
assert s.proc.poll() is None, f"log tail:\n{s.read_log()[-400:]}"
|
|
log = s.read_log()
|
|
assert "builtin cursor in use" not in log, \
|
|
f"no warning expected for a 'default' request:\n{log[-400:]}"
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def test_standalone_headless_cursor_theme_installed_no_warning():
|
|
"""An installed theme must be used with no builtin-substitution warning.
|
|
Uses the host's Adwaita theme as a real installed example."""
|
|
if not os.path.exists("/usr/share/icons/Adwaita/cursors"):
|
|
pytest.skip("host has no Adwaita cursor theme")
|
|
s = _spawn_standalone({"XCURSOR_THEME": "Adwaita"},
|
|
"build/wltest '#ff0000'")
|
|
try:
|
|
s.wait_app_active(0)
|
|
assert s.proc.poll() is None, f"log tail:\n{s.read_log()[-400:]}"
|
|
log = s.read_log()
|
|
assert "builtin cursor in use" not in log, \
|
|
f"installed theme must not be substituted:\n{log[-400:]}"
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def test_standalone_headless_last_tab_exits_cleanly(standalone_session):
|
|
"""Closing the final tab exits standalone with rc 0 instead of crashing
|
|
on the NULL host-window proxies (which only exist in nested mode)."""
|
|
s = standalone_session
|
|
assert s.proc.poll() is None, f"standalone exited early: {s.read_log()[-500:]}"
|
|
s.wait_any_active()
|
|
for pid in s.launch_pids().values():
|
|
os.kill(pid, signal.SIGTERM)
|
|
rc = s.proc.wait(timeout=10)
|
|
assert rc == 0, f"rc={rc} (139 = SIGSEGV); log tail:\n{s.read_log()[-800:]}"
|
|
|
|
|
|
def test_standalone_headless_screen_mode_sets_output_size():
|
|
"""--screen-mode WxH sizes the output; on the headless backend it becomes
|
|
the add_output() size, observable through the dumped frame dimensions."""
|
|
s = _spawn_standalone({}, "--screen-mode", "480x320", "build/wltest '#ff0000'")
|
|
try:
|
|
s.wait_app_active(0)
|
|
assert s.proc.poll() is None, f"log tail:\n{s.read_log()[-400:]}"
|
|
w, h, _ = s.ppm_rgb(s.dump_host())
|
|
assert (w, h) == (480, 320), \
|
|
f"output should honour --screen-mode 480x320, got {w}x{h}"
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def test_standalone_headless_screen_mode_not_consumed_as_app():
|
|
"""--screen-mode is stripped from argv, so it must not become a tab
|
|
launcher (no 'starts with -' warning) while both real apps still boot."""
|
|
s = _spawn_standalone({}, "--screen-mode", "640x480",
|
|
"build/wltest '#ff0000'", "build/wltest '#0000ff'")
|
|
try:
|
|
s.wait_any_active()
|
|
assert s.proc.poll() is None, f"log tail:\n{s.read_log()[-400:]}"
|
|
assert "starts with '-'" not in s.read_log(), \
|
|
f"--screen-mode leaked into the app list:\n{s.read_log()[-400:]}"
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def test_standalone_headless_screen_mode_bad_value_exits(tmp_path):
|
|
"""A malformed --screen-mode value is a startup error (rc 2) naming the
|
|
problem, before any window or backend work."""
|
|
env = dict(os.environ)
|
|
for case in (["--screen-mode"], ["--screen-mode=abc"], ["--screen-mode", "1280"]):
|
|
proc = subprocess.Popen(
|
|
["env", "-i",
|
|
"PATH=" + env["PATH"],
|
|
"HOME=" + env.get("HOME", "/"),
|
|
"XDG_RUNTIME_DIR=" + env.get("XDG_RUNTIME_DIR", "/tmp"),
|
|
"TABBED_MODE=standalone",
|
|
"WLR_BACKENDS=headless",
|
|
"stdbuf", "-oL",
|
|
BIN] + case,
|
|
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
|
preexec_fn=_preexec_death(os.getpid()),
|
|
)
|
|
out, _ = proc.communicate(timeout=15)
|
|
assert proc.returncode == 2, \
|
|
f"--screen-mode {case} should exit rc 2; rc={proc.returncode}\n{out}"
|
|
assert b"--screen-mode" in out, \
|
|
f"error should mention --screen-mode:\n{out}"
|
|
|
|
|
|
def test_standalone_headless_overlay_font_initialized(standalone_session):
|
|
"""Standalone builds the overlay glyph atlas with the wlr renderer, so tab
|
|
titles are drawn next to their icons. Regressed when overlay_font_init()
|
|
only ever ran via the nested init_egl() path."""
|
|
if _pick_font() is None:
|
|
pytest.skip("no usable font on this machine")
|
|
s = standalone_session
|
|
assert s.proc.poll() is None, f"standalone exited early: {s.read_log()[-500:]}"
|
|
s.wait_any_active()
|
|
assert "[overlay] font " in s.read_log(), (
|
|
"no standalone font atlas; overlay shows icons without text:\n"
|
|
f"{s.read_log()[-800:]}")
|
|
|
|
|
|
def test_standalone_headless_overlay_no_needless_ellipsis(standalone_session):
|
|
"""Titles that fit their overlay row whole must not be shrunk to "..." —
|
|
a row's text budget is its title's natural width, so forcing "..." space
|
|
chopped the last characters (e.g. "foot" -> "fo...")."""
|
|
if _pick_font() is None:
|
|
pytest.skip("no usable font on this machine")
|
|
s = standalone_session
|
|
s.wait_any_active()
|
|
|
|
s.mods(ALT)
|
|
deadline = time.monotonic() + 15.0
|
|
while time.monotonic() < deadline:
|
|
w, h, px = s.ppm_rgb(s.dump_host())
|
|
if _has_dark(w, h, px):
|
|
break
|
|
assert _has_dark(w, h, px), "standalone overlay box not visible while Alt"
|
|
|
|
log = s.read_log()
|
|
assert "[overlay] ellipsis" not in log, (
|
|
"overlay ellipsized a title that fit its row:\n"
|
|
f"{log[-800:]}")
|
|
|
|
s.mods(0)
|
|
|
|
|
|
# -- clipboard bridge (GNOME/Mutter focus-based wl_data_device backend) -----
|
|
|
|
def _run_display(display, args, timeout=6, **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=timeout,
|
|
preexec_fn=_preexec_death(os.getpid()), **kw)
|
|
|
|
|
|
def _poll_paste(display, want, deadline=45.0):
|
|
"""Wait until `wl-paste` on `display` returns exactly `want`."""
|
|
r = None
|
|
since = time.monotonic()
|
|
while time.monotonic() - since < deadline:
|
|
try:
|
|
r = _run_display(display, ["wl-paste", "-n"])
|
|
except subprocess.TimeoutExpired:
|
|
# A single wl-paste can stall while the host clipboard daemon is
|
|
# sluggish; that's "not there yet", not a failure.
|
|
continue
|
|
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,
|
|
preexec_fn=_preexec_death(os.getpid()))
|
|
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,
|
|
preexec_fn=_preexec_death(os.getpid()))
|
|
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):
|
|
# wltest renders each tab as a solid colour; the fixture's tabs are
|
|
# tab 0 = RED, tab 1 = BLUE.
|
|
def tab_colour(index):
|
|
return (0, 0, 255) if index == 1 else (255, 0, 0)
|
|
|
|
session.dump_host() # warm up the present chain
|
|
start = session.wait_any_active()
|
|
|
|
# The host pixel readback only works while the host compositor keeps
|
|
# presenting the nested window; when it stalls, the back buffer freezes
|
|
# on whatever tab presented first. Track whether the pixel is
|
|
# observable at all, and skip (after asserting the compositor-owned
|
|
# switching, held by the wait_active_change asserts) rather than flake.
|
|
saw = session.observed_host_colour(tab_colour(start))
|
|
|
|
session.cycle(+1)
|
|
other = session.wait_active_change(start)
|
|
assert other == 1 - start
|
|
saw |= session.observed_host_colour(tab_colour(other))
|
|
|
|
session.cycle(-1)
|
|
assert session.wait_active_change(other) == start
|
|
saw |= session.observed_host_colour(tab_colour(start))
|
|
|
|
if not saw:
|
|
pytest.skip("host presents not advancing in this session; "
|
|
"tab switching verified from the compositor log")
|
|
|
|
|
|
def test_function_keys_pass_through(session):
|
|
"""F9/F10/F11 are not grabbed by the compositor: they reach the focused
|
|
client, so tapping them must not toggle fullscreen or trigger dumps."""
|
|
start = session.wait_any_active()
|
|
base_active = session.active_count()
|
|
for code in (F11, 67, 68): # F11, F9, F10
|
|
session.tap(code)
|
|
time.sleep(0.3)
|
|
log = session.read_log()
|
|
assert "host fullscreen=" not in log, "function key toggled fullscreen"
|
|
assert "[dump] scene" not in log, "function key triggered a scene dump"
|
|
assert "[dump] host" not in log, "function key triggered a host dump"
|
|
assert session.active_count() == base_active, "function key cycled tabs"
|
|
assert session.proc.poll() is None
|
|
|
|
|
|
def test_client_fullscreen_round_trip(session):
|
|
"""A client's fullscreen request is honored and mirrored onto the host
|
|
window; the host configure then propagates fullscreen to every tab.
|
|
wltest toggles xdg_toplevel fullscreen on the 'F' key, so tap it and
|
|
watch both directions of the mirror."""
|
|
session.wait_any_active()
|
|
session.tap(F)
|
|
session.wait_for("host fullscreen=1")
|
|
session.wait_for("clients fullscreen=1")
|
|
session.tap(F)
|
|
session.wait_for("host fullscreen=0")
|
|
session.wait_for("clients 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}"
|
|
|
|
|
|
def test_cycle_wraps(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
|
|
session.cycle(-1)
|
|
assert session.wait_active_change(start) == other
|
|
session.cycle(-1)
|
|
assert session.wait_active_change(other) == start
|
|
|
|
|
|
def test_hold_alt_down_repeats(session):
|
|
start = session.wait_any_active()
|
|
assert start in (0, 1)
|
|
# Press and hold Alt+Down and wait for the compositor's own repeat
|
|
# timer (600ms delay, then 25Hz) to cycle several tabs. base is counted
|
|
# BEFORE the press so the press's own immediate cycle always counts toward
|
|
# the >=3 transitions: the count can't race with the press being processed.
|
|
# We still poll for the transitions rather than sleeping a fixed 1.3s so
|
|
# any scheduling slack never pushes the repeats past the window.
|
|
session.mods(ALT)
|
|
base = session.active_count()
|
|
session.key(DOWN, KEY_PRESSED)
|
|
deadline = time.monotonic() + 5.0
|
|
while time.monotonic() < deadline and session.active_count() - base < 3:
|
|
time.sleep(0.05)
|
|
transitions = session.active_count() - base
|
|
# Clean release: let go of the key and the modifier.
|
|
session.key(DOWN, KEY_RELEASED)
|
|
session.mods(0)
|
|
assert transitions >= 3, (
|
|
f"held Alt+Down produced only {transitions} active transitions "
|
|
"(expected repeats to cycle several tabs)"
|
|
)
|
|
settled = session.active_count()
|
|
time.sleep(0.3)
|
|
assert session.active_count() == settled, "repeats continued after release"
|
|
|
|
|
|
def test_alt_holds_tab_overlay(session):
|
|
"""Holding Alt shows the tab-title overlay (a centered box darkening the
|
|
content); releasing Alt hides it again. The overlay is drawn into the
|
|
host frame in the same blit that produces the SIGUSR1 dump, so a dark
|
|
box pixel appears in the centered window while Alt is held and reverts
|
|
after release."""
|
|
session.wait_any_active()
|
|
|
|
def has_dark(w, h, px):
|
|
for y in range(h // 4, 3 * h // 4):
|
|
row = y * w * 3
|
|
for x in range(w // 4, 3 * w // 4):
|
|
i = row + x * 3
|
|
if px[i] < 40 and px[i + 1] < 40 and px[i + 2] < 40:
|
|
return True
|
|
return False
|
|
|
|
w, h, px = session.ppm_rgb(session.dump_host())
|
|
assert not has_dark(w, h, px), "content already dark before Alt"
|
|
|
|
session.mods(ALT)
|
|
deadline = time.monotonic() + 15.0
|
|
while time.monotonic() < deadline:
|
|
w, h, px = session.ppm_rgb(session.dump_host())
|
|
if has_dark(w, h, px):
|
|
break
|
|
assert has_dark(w, h, px), "overlay box not visible while Alt is held"
|
|
|
|
session.mods(0)
|
|
deadline = time.monotonic() + 15.0
|
|
while time.monotonic() < deadline:
|
|
w, h, px = session.ppm_rgb(session.dump_host())
|
|
if not has_dark(w, h, px):
|
|
break
|
|
assert not has_dark(w, h, px), "overlay did not hide on Alt release"
|
|
assert session.proc.poll() is None
|
|
|
|
|
|
def test_non_ctrl_key_ignored(session):
|
|
start = session.wait_any_active()
|
|
assert start in (0, 1)
|
|
base_active = session.active_count()
|
|
session.key(30, KEY_PRESSED) # 'A'
|
|
session.key(30, KEY_RELEASED)
|
|
time.sleep(0.3)
|
|
assert session.active_count() == base_active, "'A' without ctrl cycled tabs"
|
|
assert "host fullscreen=" not in session.read_log(), "'A' toggled fullscreen"
|
|
assert session.proc.poll() is None
|
|
session.tap(F11)
|
|
time.sleep(0.3)
|
|
assert "host fullscreen=" not in session.read_log(), "F11 toggled fullscreen"
|
|
assert session.proc.poll() is None
|
|
|
|
|
|
def test_active_switch_on_client_exit(session):
|
|
start = session.wait_any_active()
|
|
assert start in (0, 1)
|
|
pids = session.launch_pids()
|
|
other = 1 - start
|
|
|
|
# Kill the active tab: the compositor must fall through to the other.
|
|
# [claim] pid is the connected client (wltest); SIGTERM closes its
|
|
# Wayland socket, which is what makes the compositor destroy the tab.
|
|
os.kill(pids[start], signal.SIGTERM)
|
|
assert session.wait_active_change(start) == other, \
|
|
"active did not fall through to the other tab after its tab died"
|
|
assert session.proc.poll() is None, "compositor exited after one tab died"
|
|
|
|
# Kill the last tab: the compositor should exit cleanly on its own.
|
|
os.kill(pids[other], signal.SIGTERM)
|
|
session.proc.wait(timeout=5)
|
|
assert session.proc.returncode == 0, \
|
|
f"compositor exited with {session.proc.returncode} after last tab closed"
|
|
|
|
|
|
def test_tab_death_activates_left_neighbor(session):
|
|
"""With three tabs [0,1,2], killing index 2 must activate its left
|
|
neighbor (index 1), not the leftmost tab; killing the first tab (index 0)
|
|
must fall through to index 1 rather than wrapping around to the right."""
|
|
start = session.wait_any_active()
|
|
assert start in (0, 1)
|
|
# Ensure the tab at index 1 is active BEFORE the third client connects,
|
|
# so the anchored new tab lands at index 2 (not stealing slot 1).
|
|
if start == 0:
|
|
session.cycle(+1)
|
|
session.wait_active(1)
|
|
# Both fixture tabs must have claimed slots 0 and 1 already, otherwise a
|
|
# slot steal reorders things and the "index 2" assertion below lies.
|
|
pids = session.launch_pids()
|
|
|
|
third = subprocess.Popen(
|
|
[WL, GREEN],
|
|
env=dict(os.environ, WAYLAND_DISPLAY=session.socket),
|
|
stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
start_new_session=True,
|
|
preexec_fn=_preexec_death(os.getpid()))
|
|
try:
|
|
# The new tab is anchored after the active tab (index 1): claim app=2.
|
|
session.wait_for(r"\[claim\] app=2 pid=(\d+)")
|
|
pid2 = int(re.search(r"\[claim\] app=2 pid=(\d+)",
|
|
session.read_log()).group(1))
|
|
pid0 = pids[0]
|
|
|
|
# The externally-spawned client is born hidden (no map-autofocus
|
|
# anymore), so cycle to it: active is 1 (the crown), +1 reaches 2.
|
|
session.cycle(+1)
|
|
session.wait_active(2)
|
|
|
|
# Rightmost tab dies: fall to its left neighbor (index 1), not 0.
|
|
os.kill(pid2, signal.SIGTERM)
|
|
assert session.wait_active_change(2) == 1, \
|
|
"active did not fall to index 1 (left neighbor) after index 2 died"
|
|
assert session.proc.poll() is None, \
|
|
"compositor exited after middle tab died"
|
|
|
|
# First tab dies: no left neighbor, fall right to index 1 (no wrap).
|
|
session.cycle(-1)
|
|
session.wait_active(0)
|
|
os.kill(pid0, signal.SIGTERM)
|
|
assert session.wait_active_change(0) == 1, \
|
|
"active did not fall to index 1 after first tab died"
|
|
assert session.proc.poll() is None, \
|
|
"compositor exited after first tab died"
|
|
finally:
|
|
if third.poll() is None:
|
|
third.terminate()
|
|
third.wait(timeout=5)
|
|
|
|
|
|
def _nested_session(*tabs, extra_args=()):
|
|
"""Launch a nested compositor with an arbitrary tab list (each tab is a
|
|
shell command string, as on the CLI). extra_args are passed on the CLI
|
|
before the tabs (e.g. --modifier super). The `session` fixture exists for
|
|
the common two-tab case; tests needing a specific tab lineup use this."""
|
|
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", "TABBED_SYNTH_INPUT=1", "stdbuf", "-oL",
|
|
BIN, *extra_args] + list(tabs),
|
|
stdin=subprocess.PIPE,
|
|
stdout=open(LOG, "ab"), stderr=subprocess.STDOUT,
|
|
preexec_fn=_preexec_death(os.getpid()),
|
|
)
|
|
return Session(proc)
|
|
|
|
|
|
def test_ctrl_shift_t_becomes_active(session):
|
|
"""Ctrl+Shift+T's fresh terminal still becomes the active tab on map
|
|
(the add_tab exception to the no-steal rule), while other new tabs never
|
|
steal focus."""
|
|
old_term = os.environ.get("TERMINAL")
|
|
os.environ["TERMINAL"] = f"{WL} '#ffff00'"
|
|
try:
|
|
session.wait_any_active()
|
|
before = session.active_count()
|
|
session.chord(CTRL | SHIFT, 20)
|
|
deadline = time.monotonic() + 15.0
|
|
while session.active_count() <= before:
|
|
assert session.proc.poll() is None, \
|
|
"compositor exited after Ctrl+Shift+T"
|
|
assert time.monotonic() < deadline, \
|
|
"Ctrl+Shift+T's terminal never became active"
|
|
time.sleep(0.1)
|
|
assert session.active_count() == before + 1, \
|
|
"Ctrl+Shift+T should produce exactly one focus transition"
|
|
assert session.proc.poll() is None
|
|
finally:
|
|
if old_term is None:
|
|
os.environ.pop("TERMINAL", None)
|
|
else:
|
|
os.environ["TERMINAL"] = old_term
|
|
|
|
|
|
def test_sub_window_hidden_when_parent_not_active():
|
|
"""A second toplevel from an already-running client is a sub-window of
|
|
that tab: it gets its own slot glued right after the parent tab, and
|
|
because the parent is not the tab being shown it must NOT steal focus
|
|
(no churn -- the crowned tab stays active)."""
|
|
if not os.environ.get("WAYLAND_DISPLAY"):
|
|
pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
|
|
# Tab 0 (RED) opens a green extra toplevel from the same client.
|
|
s = _nested_session(f"{WL} --extra '{GREEN}' '{RED}'", f"{WL} '{BLUE}'")
|
|
try:
|
|
s.wait_any_active()
|
|
p0 = s.launch_pids()[0]
|
|
# The extra claims the slot right after tab 0 (app=1) with the same
|
|
# client pid as RED. BLUE can also claim app=1 (before the extra
|
|
# inserts and shifts it), so match app=1 by pid on the extra.
|
|
deadline = time.monotonic() + 15.0
|
|
extra_claimed = False
|
|
while time.monotonic() < deadline:
|
|
for a, p in re.findall(r"\[claim\] app=(\d+) pid=(\d+)",
|
|
s.read_log()):
|
|
if a == "1" and int(p) == p0:
|
|
extra_claimed = True
|
|
break
|
|
if extra_claimed:
|
|
break
|
|
assert s.proc.poll() is None, "compositor exited while extra mapped"
|
|
time.sleep(0.1)
|
|
assert extra_claimed, \
|
|
"sub-window never claimed a slot right of its parent"
|
|
# The extra must have appeared hidden: only the crown's active line
|
|
# may exist.
|
|
time.sleep(0.5)
|
|
assert len(ACTIVE_RE.findall(s.read_log())) == 1, \
|
|
"sub-window stole focus from the crowned tab"
|
|
assert s.proc.poll() is None
|
|
finally:
|
|
s.close()
|
|
if s.proc.poll() is None:
|
|
s.proc.terminate()
|
|
|
|
|
|
def test_sub_window_activates_when_parent_active():
|
|
"""A sub-window whose parent tab is the one being shown becomes the
|
|
active tab on map (the second exception to the no-steal rule)."""
|
|
if not os.environ.get("WAYLAND_DISPLAY"):
|
|
pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
|
|
s = _nested_session(f"{WL} --extra '{GREEN}' '{RED}'")
|
|
try:
|
|
s.wait_any_active() # the single tab is crowned, then ...
|
|
s.wait_for(r"active app=1") # ... its sub-window activates on map.
|
|
assert s.proc.poll() is None
|
|
finally:
|
|
s.close()
|
|
if s.proc.poll() is None:
|
|
s.proc.terminate()
|
|
|
|
|
|
def test_alt_delete_activates_right_neighbor():
|
|
"""Alt+Delete closes the active tab, making the tab to its right active
|
|
(SIGTERM, escalated to SIGKILL only if the client lingers). With three
|
|
tabs [0,1,2] and 1 active, closing 1 must activate 2, not 0."""
|
|
if not os.environ.get("WAYLAND_DISPLAY"):
|
|
pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
|
|
s = _nested_session(f"{WL} '{RED}'", f"{WL} '{GREEN}'", f"{WL} '{BLUE}'")
|
|
try:
|
|
start = s.wait_any_active()
|
|
assert start in (0, 1, 2)
|
|
# Ensure the third tab has claimed a slot so index 2 really exists.
|
|
s.wait_for(r"\[claim\] app=2 pid=(\d+)")
|
|
# Make tab 1 active so a right neighbor exists.
|
|
for _ in range(3):
|
|
if s.active() == 1:
|
|
break
|
|
if s.cycle_ack(+1) == 1:
|
|
break
|
|
if s.proc.poll() is not None:
|
|
pytest.fail("compositor exited while cycling to tab 1")
|
|
s.wait_active(1)
|
|
|
|
s.chord(ALT, DELETE)
|
|
assert s.wait_active_change(1) == 2, \
|
|
"Alt+Delete did not activate the right neighbor"
|
|
assert s.proc.poll() is None, "compositor exited after Alt+Delete"
|
|
s.wait_for(r"tabbed-wlroots: closing tab 1")
|
|
finally:
|
|
s.close()
|
|
if s.proc.poll() is None:
|
|
s.proc.terminate()
|
|
|
|
|
|
def test_alt_delete_rightmost_falls_left():
|
|
"""Closing the rightmost tab has no right neighbor to hand active off to,
|
|
so it falls back to the left neighbor (index 1 of [0,1,2])."""
|
|
if not os.environ.get("WAYLAND_DISPLAY"):
|
|
pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
|
|
s = _nested_session(f"{WL} '{RED}'", f"{WL} '{GREEN}'", f"{WL} '{BLUE}'")
|
|
try:
|
|
s.wait_any_active()
|
|
# Ensure the third tab has claimed a slot so index 2 really exists.
|
|
s.wait_for(r"\[claim\] app=2 pid=(\d+)")
|
|
while s.active() != 2:
|
|
if s.cycle_ack(+1) == 2:
|
|
break
|
|
if s.proc.poll() is not None:
|
|
pytest.fail("compositor exited while cycling to tab 2")
|
|
s.wait_active(2)
|
|
|
|
s.chord(ALT, DELETE)
|
|
assert s.wait_active_change(2) == 1, \
|
|
"rightmost Alt+Delete did not fall back to the left neighbor"
|
|
assert s.proc.poll() is None, \
|
|
"compositor exited after closing rightmost tab"
|
|
finally:
|
|
s.close()
|
|
if s.proc.poll() is None:
|
|
s.proc.terminate()
|
|
|
|
|
|
def test_alt_delete_second_tab_right_neighbor_two_tabs():
|
|
"""Two-tab session: closing tab 0 makes its right neighbor (1) active."""
|
|
if not os.environ.get("WAYLAND_DISPLAY"):
|
|
pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
|
|
s = _nested_session(f"{WL} '{RED}'", f"{WL} '{BLUE}'")
|
|
try:
|
|
s.wait_any_active()
|
|
if s.active() == 1:
|
|
if s.cycle_ack(-1) != 0:
|
|
s.wait_active(0)
|
|
s.chord(ALT, DELETE)
|
|
assert s.wait_active_change(0) == 1, \
|
|
"closing tab 0 did not activate tab 1"
|
|
assert s.proc.poll() is None, "compositor exited after closing tab 0"
|
|
finally:
|
|
s.close()
|
|
if s.proc.poll() is None:
|
|
s.proc.terminate()
|
|
|
|
|
|
def test_alt_shift_arrows_reorder_and_wrap():
|
|
"""Alt+Shift+Up/Down move the active tab within the overlay list without
|
|
switching the active surface; moving off an end wraps to the other end and
|
|
the whole list shifts (a rightmost tab moved off the end lands first with
|
|
every other tab pushed one place, not swapped into place). The resulting
|
|
order is provable behaviorally: a Ctrl+Shift+T terminal claims the slot
|
|
right after the active tab's new position."""
|
|
if not os.environ.get("WAYLAND_DISPLAY"):
|
|
pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
|
|
old_term = os.environ.get("TERMINAL")
|
|
os.environ["TERMINAL"] = f"{WL} '{RED}'"
|
|
try:
|
|
s = _nested_session(f"{WL} '{RED}'", f"{WL} '{GREEN}'",
|
|
f"{WL} '{BLUE}'")
|
|
try:
|
|
s.wait_any_active()
|
|
# Make the third tab (index 2) active so there is a middle and an
|
|
# end to move through.
|
|
for _ in range(3):
|
|
if s.active() == 2:
|
|
break
|
|
if s.cycle_ack(+1) == 2:
|
|
break
|
|
if s.proc.poll() is not None:
|
|
pytest.fail("compositor exited while cycling to tab 2")
|
|
s.wait_active(2)
|
|
before = s.active_count()
|
|
|
|
# Adjacent move UP: [R,G,B] -> [R,B,G].
|
|
s.chord(ALT | SHIFT, UP)
|
|
s.wait_for(r"\[reorder\] app=1 dir=-1")
|
|
# Adjacent move DOWN: -> [R,G,B].
|
|
s.chord(ALT | SHIFT, DOWN)
|
|
s.wait_for(r"\[reorder\] app=2 dir=1")
|
|
# Move DOWN off the end: wraps to the front and rotates the rest,
|
|
# [R,G,B] -> [B,R,G] (B first, R and G shifted, not swapped).
|
|
s.chord(ALT | SHIFT, DOWN)
|
|
s.wait_for(r"\[reorder\] app=0 dir=1")
|
|
time.sleep(0.3)
|
|
assert s.active() == 2, "reorder must not switch the active tab"
|
|
assert s.active_count() == before, \
|
|
"reorder must not emit an active-tab transition"
|
|
|
|
# With the active tab wrapped to slot 0, a new Ctrl+Shift+T tab
|
|
# is anchored right after it: the newest claim must be app=1.
|
|
claims = re.findall(r"\[claim\] app=(\d+) pid=(\d+)", s.read_log())
|
|
s.chord(CTRL | SHIFT, 20)
|
|
deadline = time.monotonic() + 15.0
|
|
while len(re.findall(r"\[claim\] app=(\d+) pid=(\d+)",
|
|
s.read_log())) <= len(claims):
|
|
assert s.proc.poll() is None, \
|
|
"compositor exited after reorder + new tab"
|
|
assert time.monotonic() < deadline, \
|
|
"Ctrl+Shift+T never claimed a slot"
|
|
time.sleep(0.1)
|
|
newest = re.findall(r"\[claim\] app=(\d+) pid=(\d+)",
|
|
s.read_log())[-1]
|
|
assert newest[0] == "1", \
|
|
f"new tab claimed slot {newest[0]}, expected 1 (after B->front)"
|
|
finally:
|
|
s.close()
|
|
if s.proc.poll() is None:
|
|
s.proc.terminate()
|
|
finally:
|
|
if old_term is None:
|
|
os.environ.pop("TERMINAL", None)
|
|
else:
|
|
os.environ["TERMINAL"] = old_term
|
|
|
|
|
|
def test_modifier_super_flag():
|
|
"""--modifier super moves the tab-management chords to Super/Mod4: cycle,
|
|
reorder and close all work, while the Alt chords stop doing anything."""
|
|
if not os.environ.get("WAYLAND_DISPLAY"):
|
|
pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
|
|
s = _nested_session(f"{WL} '{RED}'", f"{WL} '{BLUE}'",
|
|
extra_args=["--modifier", "super"])
|
|
try:
|
|
start = s.wait_any_active()
|
|
assert s.proc.poll() is None, "compositor exited with --modifier super"
|
|
|
|
# Super+Down cycles; a plain Alt+Down must not.
|
|
s.chord(SUPER, DOWN)
|
|
other = s.wait_active_change(start, timeout=5.0)
|
|
assert other == 1 - start, "Super+Down did not cycle the other tab"
|
|
s.chord(ALT, DOWN)
|
|
time.sleep(0.3)
|
|
assert s.active() == other, \
|
|
"Alt+Down still cycled in --modifier super mode"
|
|
|
|
# Super+Shift+Up reorders the active tab.
|
|
s.chord(SUPER | SHIFT, UP)
|
|
s.wait_for(r"\[reorder\] app=\d+ dir=-1")
|
|
|
|
# Super+Delete closes the active tab: the compositor logs the close
|
|
# and hands active off (a NEW active-tab line appears). Can't use
|
|
# wait_active_change here: reorder changed the active tab's slot
|
|
# without logging one, so the post-close index can match the
|
|
# pre-close value even though the tab underneath changed.
|
|
count = s.active_count()
|
|
s.chord(SUPER, DELETE)
|
|
s.wait_for(r"tabbed-wlroots: closing tab")
|
|
deadline = time.monotonic() + 5.0
|
|
while s.active_count() <= count:
|
|
assert s.proc.poll() is None, \
|
|
"compositor exited after Super+Delete"
|
|
assert time.monotonic() < deadline, \
|
|
"active never handed off after Super+Delete"
|
|
time.sleep(0.1)
|
|
finally:
|
|
s.close()
|
|
if s.proc.poll() is None:
|
|
s.proc.terminate()
|
|
|
|
|
|
def test_alt_delete_escalates_sigkill(tmp_path):
|
|
"""A client that traps SIGTERM must still be reaped by the SIGKILL
|
|
escalation, so Alt+Delete never strands a lingering tab."""
|
|
if not os.environ.get("WAYLAND_DISPLAY"):
|
|
pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
|
|
# A wrapper that ignores SIGTERM, then execs the test client so the
|
|
# tab's process group carries SIG_IGN for TERM across the exec.
|
|
wrapper = tmp_path / "trap-term.sh"
|
|
wrapper.write_text(
|
|
"trap '' TERM\nexec '{}' '{}'\n".format(WL, GREEN))
|
|
wrapper.chmod(0o755)
|
|
s = _nested_session(f"{WL} '{RED}'", str(wrapper))
|
|
try:
|
|
pids = s.launch_pids()
|
|
assert 1 in pids, "second tab never claimed a slot"
|
|
trap_pid = pids[1]
|
|
# Wait for the startup active-tab lines to settle (both clients map
|
|
# asynchronously and toggle `active app=` as they land), otherwise
|
|
# s.active() can return a transient value and the Alt+Delete below
|
|
# closes the wrong tab.
|
|
s.wait_any_active()
|
|
# The trap tab is the second; bring it active and close it.
|
|
while s.active() != 1:
|
|
if s.cycle_ack(+1) == 1:
|
|
break
|
|
if s.proc.poll() is not None:
|
|
pytest.fail("compositor exited while cycling to the trap tab")
|
|
s.wait_active(1)
|
|
|
|
s.chord(ALT, DELETE)
|
|
deadline = time.monotonic() + 5.0
|
|
alive = True
|
|
while time.monotonic() < deadline:
|
|
try:
|
|
os.kill(trap_pid, 0)
|
|
time.sleep(0.1)
|
|
except ProcessLookupError:
|
|
alive = False
|
|
break
|
|
assert not alive, "SIGTERM-trapping tab survived Alt+Delete (no SIGKILL escalation)"
|
|
# It must have moved active away from the closing tab, and stayed up.
|
|
assert s.active() != 1, "active still on the closing trap tab"
|
|
assert s.proc.poll() is None, "compositor exited after closing trap tab"
|
|
finally:
|
|
s.close()
|
|
if s.proc.poll() is None:
|
|
s.proc.terminate()
|
|
|
|
|
|
def test_plain_delete_not_grabbed(session):
|
|
"""Plain Delete (no Alt) must not close the active tab."""
|
|
session.wait_any_active()
|
|
base_active = session.active()
|
|
session.tap(DELETE)
|
|
time.sleep(0.3)
|
|
assert session.active() == base_active, \
|
|
"Delete without Alt closed the active tab"
|
|
assert session.proc.poll() is None
|
|
|
|
|
|
def test_standalone_headless_alt_delete_closes_tab(standalone_session):
|
|
"""Alt+Delete routes through the shared hotkey path in standalone mode:
|
|
the active tab closes and the neighbor becomes active, then closing the
|
|
last tab exits cleanly."""
|
|
s = standalone_session
|
|
start = s.wait_any_active()
|
|
other = 1 - start
|
|
s.chord(ALT, DELETE)
|
|
assert s.wait_active_change(start) == other, \
|
|
"standalone Alt+Delete did not activate the other tab"
|
|
assert s.proc.poll() is None, "standalone compositor exited mid-test"
|
|
|
|
# Close the final tab: the compositor should exit cleanly on its own.
|
|
s.chord(ALT, DELETE)
|
|
s.proc.wait(timeout=5)
|
|
assert s.proc.returncode == 0, \
|
|
f"standalone compositor exited with {s.proc.returncode} on last tab close"
|
|
|
|
|
|
def test_sigusr1_dumps(session):
|
|
scene_rx = re.compile(r"\[dump\] scene")
|
|
host_rx = re.compile(r"\[dump\] host back buffer -> (\S+\.ppm)")
|
|
# Warm the dump machinery so both baselines exist, then assert SIGUSR1
|
|
# produces a *new* scene dump attempt and a *new* host ppm.
|
|
session.dump_host()
|
|
session.wait_for(r"\[dump\] scene")
|
|
scene_before = len(scene_rx.findall(session.read_log()))
|
|
host_before = len(host_rx.findall(session.read_log()))
|
|
|
|
os.kill(session.proc.pid, signal.SIGUSR1)
|
|
since = time.monotonic()
|
|
while time.monotonic() - since < 5.0:
|
|
log = session.read_log()
|
|
if len(scene_rx.findall(log)) > scene_before and \
|
|
len(host_rx.findall(log)) > host_before:
|
|
return
|
|
time.sleep(0.1)
|
|
pytest.fail("SIGUSR1 produced no new scene+host dump lines")
|
|
|
|
|
|
def test_dash_h_prints_usage():
|
|
""" `-h` is the way to see usage: it prints it and exits 0 without
|
|
starting a session. """
|
|
r = subprocess.run([BIN, "-h"], stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
preexec_fn=_preexec_death(os.getpid()))
|
|
assert r.returncode == 0
|
|
assert b"usage:" in r.stdout
|
|
|
|
|
|
def test_no_args_opens_terminal():
|
|
"""With no command-line tabs, a single terminal tab opens and maps."""
|
|
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
|
|
|
|
# The one tab is the terminal named by TERMINAL (a wltest, so no real
|
|
# terminal emulator has to be installed on the test host).
|
|
proc = subprocess.Popen(
|
|
["env", "TABBED_LOG=2", "TABBED_SYNTH_INPUT=1",
|
|
f"TERMINAL={WL} '{RED}'", "stdbuf", "-oL", BIN],
|
|
stdin=subprocess.PIPE,
|
|
stdout=open(LOG, "ab"), stderr=subprocess.STDOUT,
|
|
preexec_fn=_preexec_death(os.getpid()),
|
|
)
|
|
s = Session(proc)
|
|
try:
|
|
s.wait_for(r"\[diag\] map app=0")
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def test_no_args_missing_terminal_fails():
|
|
"""A default terminal that is not on PATH aborts before any window."""
|
|
r = subprocess.run(
|
|
["env", "TERMINAL=/nonexistent/tw-term", BIN],
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
|
preexec_fn=_preexec_death(os.getpid()))
|
|
assert r.returncode != 0
|
|
assert b"not found in PATH" in (r.stdout + r.stderr)
|
|
|
|
|
|
def test_missing_arg_fails():
|
|
"""A command-line tab whose program is not on PATH aborts up front."""
|
|
r = subprocess.run(
|
|
[BIN, "definitely-no-such-cmd-xyz"],
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
|
preexec_fn=_preexec_death(os.getpid()))
|
|
assert r.returncode != 0
|
|
assert b"definitely-no-such-cmd-xyz" in (r.stdout + r.stderr)
|
|
assert b"not found in PATH" in (r.stdout + r.stderr)
|