Samuel Onoja
bffd538dd4
Queue popup, modal and notification material geometry and opacity until the matching content buffer commits. Remove event-handler opacity writes that could present material state ahead of content, while preserving compositor dismissal ownership. Add pending-presentation regression coverage and update the nested shell lifecycle check.
154 lines
8.8 KiB
Python
154 lines
8.8 KiB
Python
"""Exercise real shell frame-driven motion on a private nested desktop and bus.
|
|
|
|
FERESE_TEST_SHELL_MOTION=1 python3 scripts/tests/test_shell_motion_isolated.py
|
|
Uses release binaries by default. Requires a Wayland host, dbus-daemon and gdbus.
|
|
Protocol traces contain only this test's shell traffic, never the host desktop.
|
|
"""
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import signal
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
|
|
@unittest.skipUnless(os.environ.get("FERESE_TEST_SHELL_MOTION") == "1", "opt-in nested shell test")
|
|
class ShellMotionTest(unittest.TestCase):
|
|
def test_notifications_and_modals_follow_frames_and_finish_closing(self):
|
|
repo = Path(__file__).resolve().parents[2]
|
|
binary = repo / os.environ.get("FERESE_TEST_BINARY", "target/release/ferese")
|
|
shell = repo / os.environ.get("FERESE_TEST_SHELL", "target/release/ferese-shell")
|
|
ctl = repo / os.environ.get("FERESE_TEST_CTL", "target/release/feresectl")
|
|
children = []
|
|
with tempfile.TemporaryDirectory(prefix="ferese-shell-motion-") as temporary:
|
|
root = Path(temporary)
|
|
runtime = root / "runtime"
|
|
runtime.mkdir(mode=0o700)
|
|
config = root / "config/ferese"
|
|
config.mkdir(parents=True)
|
|
(config / "config.kdl").write_text(
|
|
'animations { speed 0.5; }\nstatus { keybinding-guide #false; }\n')
|
|
display = Path(os.environ["WAYLAND_DISPLAY"])
|
|
if not display.is_absolute():
|
|
display = Path(os.environ["XDG_RUNTIME_DIR"]) / display
|
|
env = dict(os.environ, XDG_RUNTIME_DIR=str(runtime), WAYLAND_DISPLAY=str(display),
|
|
XDG_CONFIG_HOME=str(root / "config"), XDG_STATE_HOME=str(root / "state"),
|
|
XDG_CACHE_HOME=str(root / "cache"), RUST_LOG="ferese=info,ferese::nested_input=debug")
|
|
env.pop("WAYLAND_SOCKET", None)
|
|
env.pop("FERESE_SHELL_CONTROL_SOCKET", None)
|
|
log_path = root / "session.log"
|
|
# No service directories: readiness checks must not auto-activate
|
|
# the host's notification daemon on this private bus.
|
|
bus_config = root / "bus.conf"
|
|
bus_config.write_text(
|
|
'<busconfig><type>session</type><listen>unix:tmpdir=/tmp</listen>'
|
|
'<policy context="default"><allow send_destination="*"/>'
|
|
'<allow receive_sender="*"/><allow own="*"/></policy></busconfig>')
|
|
|
|
def launch(command, **kwargs):
|
|
process = subprocess.Popen(command, env=env, start_new_session=True, **kwargs)
|
|
children.append(process)
|
|
return process
|
|
|
|
def wait_for(predicate, timeout=15):
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
value = predicate()
|
|
if value:
|
|
return value
|
|
time.sleep(.025)
|
|
trace = log_path.read_text()
|
|
relevant = '\n'.join(line for line in trace.splitlines() if any(
|
|
word in line for word in ('get_layer_surface', 'ERROR', 'WARN', 'notification')))
|
|
self.fail("Timed out:\n" + relevant[-16000:] + '\n' + trace[-4000:])
|
|
|
|
def dbus(method, *args, check=True):
|
|
return subprocess.run(["gdbus", "call", "--session", "--dest", "org.freedesktop.Notifications",
|
|
"--object-path", "/org/freedesktop/Notifications", "--method",
|
|
"org.freedesktop.Notifications." + method, *args],
|
|
env=env, text=True, capture_output=True, timeout=5, check=check)
|
|
|
|
def layer_surface(namespace):
|
|
matches = list(re.finditer(r'get_layer_surface\([^\n]*?wl_surface[@#](\d+)[^\n]*"' + namespace + '"', log_path.read_text()))
|
|
# Wayland object IDs can be reused after destruction. Restrict
|
|
# assertions to this surface's lifetime in the trace.
|
|
return (matches[-1].group(1), matches[-1].start()) if matches else None
|
|
|
|
def frames(surface):
|
|
identity, start = surface
|
|
return len(re.findall(r'wl_surface[@#]' + identity + r'\.frame\(', log_path.read_text()[start:]))
|
|
|
|
def commits(surface):
|
|
identity, start = surface
|
|
return len(re.findall(r'wl_surface[@#]' + identity + r'\.commit\(', log_path.read_text()[start:]))
|
|
|
|
def destroyed(surface):
|
|
identity, start = surface
|
|
return re.search(r'wl_surface[@#]' + identity + r'\.destroy\(', log_path.read_text()[start:]) is not None
|
|
|
|
with log_path.open("w") as log:
|
|
try:
|
|
bus = launch(["dbus-daemon", "--config-file=" + str(bus_config), "--nofork", "--print-address=1"],
|
|
stdout=subprocess.PIPE, stderr=log, text=True)
|
|
env["DBUS_SESSION_BUS_ADDRESS"] = bus.stdout.readline().strip()
|
|
compositor = launch([str(binary), "--backend", "nested", "--grant-effects", "--grant-shell-control",
|
|
"--", "env", "WAYLAND_DEBUG=client", "ICED_BACKEND=tiny-skia", str(shell)],
|
|
stdout=log, stderr=log)
|
|
wait_for(lambda: (runtime / "ferese/control.sock").is_socket())
|
|
wait_for(lambda: dbus("GetServerInformation", check=False).returncode == 0)
|
|
wait_for(lambda: layer_surface("ferese-shell-top-bar"))
|
|
time.sleep(.5)
|
|
reply = dbus("Notify", "Ferese motion test", "0", "", "Frame cadence", "Synthetic test notification", "[]", "{}", "0").stdout
|
|
notice = re.search(r'uint32 (\d+)', reply).group(1)
|
|
surface = wait_for(lambda: layer_surface("ferese-shell-notifications"))
|
|
time.sleep(1.5)
|
|
opening_frames = frames(surface)
|
|
self.assertGreater(opening_frames, 10, log_path.read_text()[-8000:])
|
|
time.sleep(1)
|
|
for attempt in range(3):
|
|
trace_start = len(log_path.read_text())
|
|
before = commits(surface)
|
|
time.sleep(2)
|
|
idle_frames = commits(surface) - before
|
|
if 'ferese::nested_input:' not in log_path.read_text()[trace_start:]:
|
|
break
|
|
else:
|
|
self.fail("Host input interrupted all three idle measurement intervals")
|
|
# Clock/status updates still redraw settled surfaces. Count
|
|
# commits: Iced can request two callbacks for one redraw.
|
|
self.assertLessEqual(idle_frames, 12, "settled toast kept requesting animation frames")
|
|
material_updates = log_path.read_text().count('.set_presentation(')
|
|
dbus("CloseNotification", notice)
|
|
wait_for(lambda: destroyed(surface), timeout=5)
|
|
self.assertGreater(log_path.read_text().count('.set_presentation(') - material_updates, 5,
|
|
"notification materials did not follow their fade")
|
|
print(f"Toast: {opening_frames} opening callback requests, {idle_frames / 2:g} settled commits/s; destroyed", flush=True)
|
|
|
|
subprocess.run([str(ctl), "toggle-keybinding-guide"], env=env, check=True, capture_output=True)
|
|
modal = wait_for(lambda: layer_surface("ferese-system-modal"))
|
|
wait_for(lambda: frames(modal) >= 4)
|
|
time.sleep(.3)
|
|
subprocess.run([str(ctl), "toggle-keybinding-guide"], env=env, check=True, capture_output=True)
|
|
wait_for(lambda: destroyed(modal), timeout=5)
|
|
self.assertGreater(frames(modal), 10, "modal close did not follow frames")
|
|
self.assertIsNone(compositor.poll(), log_path.read_text()[-8000:])
|
|
self.assertNotIn("panicked at", log_path.read_text())
|
|
print(f"Modal: {frames(modal)} callbacks; interrupted opening closed and destroyed", flush=True)
|
|
finally:
|
|
for child in reversed(children):
|
|
if child.poll() is None:
|
|
os.killpg(child.pid, signal.SIGTERM)
|
|
try:
|
|
child.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
os.killpg(child.pid, signal.SIGKILL)
|
|
child.wait(timeout=5)
|
|
if child.stdout is not None:
|
|
child.stdout.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|