Samuel Onoja
91e2f227d2
* Coordinate config writes and gate releases on desktop validation * Use toolkit rc.2 with painting and session lock fixes * Ignore retired desktop widget settings in existing configs
314 lines
19 KiB
Python
314 lines
19 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
|
|
Set FERESE_TEST_BAR_LAYOUT=islands to also test island masks and live switching.
|
|
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
|
|
from session_socket import ipc_environment, ipc_socket
|
|
import re
|
|
import signal
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
|
|
def panel_config(preset):
|
|
surface, group_surface = ("solid", "none") if preset == "continuous" else ("none", "island")
|
|
return f'''panel main {{
|
|
surface "{surface}"
|
|
group-surface "{group_surface}"
|
|
start {{ group navigation {{ island-padding 12; item overview kind=overview; item spaces kind=workspaces; }}; }}
|
|
center {{ group title {{ island-padding 12; item title kind=focused-window; }}; }}
|
|
end {{ group status {{ island-padding 12; item media kind=media; item quick kind=quick-settings; item network kind=network; item audio kind=audio; item recording kind=recording; item notifications kind=notifications; item battery kind=battery; item clock kind=clock; item display kind=display-mode; }}; }}
|
|
}}
|
|
'''
|
|
|
|
|
|
@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)
|
|
bar_layout = os.environ.get("FERESE_TEST_BAR_LAYOUT", "continuous")
|
|
self.assertIn(bar_layout, ("continuous", "islands"))
|
|
config_file = config / "config.kdl"
|
|
config_file.write_text(
|
|
'animations { speed 0.5; }\n'
|
|
'status { keybinding-guide #false; }\n' + panel_config(bar_layout))
|
|
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=ipc_environment(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=ipc_environment(env), text=True, capture_output=True, timeout=5, check=check)
|
|
|
|
def layer_surface(namespace):
|
|
matches = layer_surfaces(namespace)
|
|
# Wayland object IDs can be reused after destruction. Restrict
|
|
# assertions to this surface's lifetime in the trace.
|
|
return matches[-1] if matches else None
|
|
|
|
def layer_surfaces(namespace):
|
|
return [(match.group(1), match.start()) for match in re.finditer(
|
|
r'get_layer_surface\([^\n]*?wl_surface[@#](\d+)[^\n]*"' + namespace + '"',
|
|
log_path.read_text())]
|
|
|
|
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
|
|
|
|
def bar_region_count(surface):
|
|
identity, start = surface
|
|
trace = log_path.read_text()[start:]
|
|
effects = re.findall(
|
|
r'get_surface_effects\(new id ferese_surface_effects_v1[@#](\d+), wl_surface[@#]'
|
|
+ identity + r'\)', trace)
|
|
if not effects:
|
|
return None
|
|
|
|
updates = re.findall(
|
|
r'ferese_surface_effects_v1[@#]' + effects[-1] + r'\.set_presentation\(\d+, array\[(\d+)\],', trace)
|
|
return int(updates[-1]) // 20 if updates else None
|
|
|
|
def bar_input_rectangles(surface):
|
|
identity, start = surface
|
|
trace = log_path.read_text()[start:]
|
|
updates = list(re.finditer(
|
|
r'wl_surface[@#]' + identity + r'\.set_input_region\(wl_region[@#](\d+)\)', trace))
|
|
if not updates:
|
|
return []
|
|
|
|
update = updates[-1]
|
|
region = update.group(1)
|
|
creations = list(re.finditer(r'create_region\(new id wl_region[@#]' + region + r'\)', trace[:update.start()]))
|
|
if not creations:
|
|
return []
|
|
|
|
rectangles = [tuple(map(int, values)) for values in re.findall(
|
|
r'wl_region[@#]' + region + r'\.add\((-?\d+), (-?\d+), (\d+), (\d+)\)',
|
|
trace[creations[-1].start():update.start()])]
|
|
# Wayland regions are unions. Decorated groups can include
|
|
# redundant child control rectangles without changing coverage.
|
|
def contains(outer, inner):
|
|
x, y, width, height = outer
|
|
ix, iy, iw, ih = inner
|
|
return x <= ix and y <= iy and x + width >= ix + iw and y + height >= iy + ih
|
|
return [rectangle for index, rectangle in enumerate(rectangles)
|
|
if not any(other != rectangle and contains(other, rectangle)
|
|
or other == rectangle and other_index < index
|
|
for other_index, other in enumerate(rectangles))]
|
|
|
|
|
|
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: (ipc_socket(runtime)).is_socket())
|
|
wait_for(lambda: dbus("GetServerInformation", check=False).returncode == 0)
|
|
bar = wait_for(lambda: layer_surface("ferese-shell-top-bar"))
|
|
if bar_layout == "islands":
|
|
wait_for(lambda: bar_region_count(bar) == 2)
|
|
original_input = wait_for(lambda: bar_input_rectangles(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)
|
|
# Require multiple animation frames, without imposing a
|
|
# hardware-dependent frame rate on software-rendered CI.
|
|
self.assertGreaterEqual(opening_frames, 4, 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, 1,
|
|
"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=ipc_environment(env), check=True, capture_output=True, timeout=5)
|
|
modal = wait_for(lambda: layer_surface("ferese-system-modal"))
|
|
wait_for(lambda: frames(modal) >= 4)
|
|
time.sleep(.3)
|
|
before_close = frames(modal)
|
|
subprocess.run([str(ctl), "toggle-keybinding-guide"], env=ipc_environment(env), check=True, capture_output=True, timeout=5)
|
|
wait_for(lambda: destroyed(modal), timeout=5)
|
|
self.assertGreater(frames(modal), before_close, "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)
|
|
|
|
# A toast can resize or start closing before its previous
|
|
# software buffer has caught up with the new viewport.
|
|
for index in range(40):
|
|
reply = dbus("Notify", "Ferese motion test", "0", "", "Rapid toast",
|
|
"Short body", "[]", "{}", "0").stdout
|
|
notice = re.search(r'uint32 (\d+)', reply).group(1)
|
|
if index % 2 == 0:
|
|
dbus("Notify", "Ferese motion test", notice, "", "Updated toast",
|
|
"A longer notification body that wraps over several lines. " * 6,
|
|
"[]", "{}", "0")
|
|
dbus("CloseNotification", notice)
|
|
time.sleep(.015)
|
|
time.sleep(.5)
|
|
self.assertIsNone(compositor.poll(), log_path.read_text()[-8000:])
|
|
self.assertEqual(dbus("GetServerInformation", check=False).returncode, 0)
|
|
self.assertNotIn("panicked at", log_path.read_text())
|
|
print("Toast: 40 rapid open/update/close cycles completed", flush=True)
|
|
|
|
for _ in range(40):
|
|
subprocess.run([str(ctl), "toggle-keybinding-guide"],
|
|
env=ipc_environment(env), check=True, capture_output=True, timeout=5)
|
|
time.sleep(.03)
|
|
subprocess.run([str(ctl), "toggle-keybinding-guide"],
|
|
env=ipc_environment(env), check=True, capture_output=True, timeout=5)
|
|
time.sleep(.06)
|
|
# A final toggle may leave the owned modal closing. Let
|
|
# that motion finish before deciding whether a live modal
|
|
# needs another toggle; toggling during closing reopens it.
|
|
deadline = time.monotonic() + 5
|
|
while True:
|
|
remaining = [surface for surface in layer_surfaces("ferese-system-modal")
|
|
if not destroyed(surface)]
|
|
if not remaining or time.monotonic() >= deadline:
|
|
break
|
|
time.sleep(.05)
|
|
self.assertLessEqual(len(remaining), 1, "cancelled modals retained native surfaces")
|
|
if remaining:
|
|
subprocess.run([str(ctl), "toggle-keybinding-guide"],
|
|
env=ipc_environment(env), check=True, capture_output=True, timeout=5)
|
|
wait_for(lambda: destroyed(remaining[0]))
|
|
print("Modal: rapid toggles released all native surfaces", flush=True)
|
|
|
|
if bar_layout == "islands":
|
|
for layout_name, expected_count in [("continuous", 1), ("islands", None)]:
|
|
config_file.write_text('animations { speed 0.5; }\nstatus { keybinding-guide #false; }\n' + panel_config(layout_name))
|
|
subprocess.run([str(ctl), "reload-config"], env=ipc_environment(env), check=True, capture_output=True, timeout=5)
|
|
wait_for(lambda: bar_region_count(bar) == (expected_count or 2))
|
|
|
|
self.assertIsNone(compositor.poll(), log_path.read_text()[-8000:])
|
|
self.assertNotIn("panicked at", log_path.read_text())
|
|
print("Bar: separate material regions; continuous/islands live switches passed", flush=True)
|
|
|
|
for padding in [4, 0]:
|
|
source = config_file.read_text()
|
|
config_file.write_text(re.sub(
|
|
r'island-padding \d+', f'island-padding {padding}', source))
|
|
subprocess.run([str(ctl), "reload-config"], env=ipc_environment(env), check=True, capture_output=True, timeout=5)
|
|
expected_width = original_input[0][2] - 2 * (12 - padding)
|
|
wait_for(lambda: len(bar_input_rectangles(bar)) == 2
|
|
and bar_input_rectangles(bar)[0][2] == expected_width)
|
|
|
|
print("Bar: live padding reductions updated the input regions", flush=True)
|
|
saved = config_file.read_text()
|
|
config_file.write_text('panel main { surface "invalid-surface"; }\n')
|
|
rejected = subprocess.run([str(ctl), "reload-config"], env=ipc_environment(env),
|
|
capture_output=True, text=True, timeout=5)
|
|
self.assertNotEqual(rejected.returncode, 0, "invalid reload was accepted")
|
|
self.assertIn("invalid_config", rejected.stderr)
|
|
self.assertFalse(destroyed(bar), "invalid reload destroyed the current panel")
|
|
self.assertEqual(dbus("GetServerInformation", check=False).returncode, 0)
|
|
config_file.write_text(saved)
|
|
subprocess.run([str(ctl), "reload-config"], env=ipc_environment(env),
|
|
check=True, capture_output=True, timeout=5)
|
|
subprocess.run([str(ctl), "toggle-keybinding-guide"], env=ipc_environment(env),
|
|
check=True, capture_output=True, timeout=5)
|
|
recovered = wait_for(lambda: (surface := layer_surface("ferese-system-modal"))
|
|
and not destroyed(surface) and surface)
|
|
subprocess.run([str(ctl), "toggle-keybinding-guide"], env=ipc_environment(env),
|
|
check=True, capture_output=True, timeout=5)
|
|
wait_for(lambda: destroyed(recovered))
|
|
self.assertIsNone(compositor.poll(), log_path.read_text()[-8000:])
|
|
self.assertNotIn("panicked at", log_path.read_text())
|
|
print("Reload: invalid config preserved the panel; repaired config reopened and released a modal", 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 directory := os.environ.get("FERESE_TEST_LOG_DIR"):
|
|
logs = Path(directory)
|
|
logs.mkdir(parents=True, exist_ok=True)
|
|
shutil.copyfile(log_path, logs / f"shell-{bar_layout}.log")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|