Samuel Onoja
979afbd213
Avoid background title updates when no shell window list is visible, and skip output and capture invalidation for titles outside compositor decorations. Preserve focused titles, fresh window lists and capture privacy. Request high EGL context priority with fallback, cover rendering and native title behavior in CI, and keep audio recording flowing while video capture is idle.
184 lines
9.6 KiB
Python
184 lines
9.6 KiB
Python
"""Keep a real GPU panel idle during background title changes.
|
|
|
|
FERESE_TEST_SHELL_TITLES=1 python3 scripts/tests/test_shell_title_updates_isolated.py
|
|
Requires a Wayland host, cc, wayland-scanner, and dbus-daemon.
|
|
"""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import select
|
|
import shlex
|
|
import signal
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
from session_socket import ipc_environment, ipc_socket
|
|
from test_shell_motion_isolated import panel_config
|
|
|
|
|
|
@unittest.skipUnless(os.environ.get("FERESE_TEST_SHELL_TITLES") == "1", "opt-in nested shell test")
|
|
class ShellTitleUpdatesTest(unittest.TestCase):
|
|
def test_background_titles_only_redraw_visible_window_lists(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-titles-") as temporary:
|
|
root = Path(temporary)
|
|
protocols = subprocess.check_output(
|
|
["pkg-config", "--variable=pkgdatadir", "wayland-protocols"], text=True).strip()
|
|
xml = str(Path(protocols) / "stable/xdg-shell/xdg-shell.xml")
|
|
for kind, output in [("client-header", "xdg-shell-client-protocol.h"),
|
|
("private-code", "xdg-shell-protocol.c")]:
|
|
subprocess.run(["wayland-scanner", kind, xml, str(root / output)], check=True)
|
|
flags = shlex.split(subprocess.check_output(
|
|
["pkg-config", "--cflags", "--libs", "wayland-client"], text=True))
|
|
subprocess.run(["cc", "-Wall", "-Wextra", "-Werror", "-I", str(root),
|
|
str(repo / "scripts/tests/fixtures/window-titles.c"),
|
|
str(root / "xdg-shell-protocol.c"), "-o", str(root / "client"), *flags], check=True)
|
|
runtime = root / "runtime"
|
|
runtime.mkdir(mode=0o700)
|
|
config = root / "config/ferese"
|
|
config.mkdir(parents=True)
|
|
(config / "config.kdl").write_text(
|
|
'status { keybinding-guide #false; }\n'
|
|
'window-rule app-id="ferese.test.titles" floating=#true\n' + panel_config("flat"))
|
|
display = Path(os.environ["WAYLAND_DISPLAY"])
|
|
if not display.is_absolute():
|
|
display = Path(os.environ["XDG_RUNTIME_DIR"]) / display
|
|
env = {key: value for key, value in os.environ.items() if not key.startswith("FERESE_")}
|
|
env.update(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"), XDG_DATA_HOME=str(root / "data"),
|
|
FERESE_COMPOSITOR_WALLPAPER="1", RUST_LOG="ferese=info")
|
|
env.pop("WAYLAND_SOCKET", None)
|
|
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>')
|
|
log_path = root / "session.log"
|
|
|
|
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):
|
|
deadline = time.monotonic() + 15
|
|
while time.monotonic() < deadline:
|
|
value = predicate()
|
|
if value:
|
|
return value
|
|
time.sleep(.025)
|
|
self.fail("Timed out:\n" + log_path.read_text()[-8000:])
|
|
|
|
def control(*args):
|
|
return subprocess.check_output([str(ctl), *args], env=ipc_environment(env), text=True, timeout=5)
|
|
|
|
def snapshots():
|
|
return log_path.read_text().count(".snapshot_end(")
|
|
|
|
def title(index, value):
|
|
client.stdin.write(f"{index} {value}\n")
|
|
client.stdin.flush()
|
|
self.assertTrue(select.select([client.stdout], [], [], 5)[0], "title client did not reply")
|
|
self.assertEqual(client.stdout.readline().strip(), "changed")
|
|
|
|
def burst(index, prefix):
|
|
for number in range(20):
|
|
title(index, f"{prefix} {number}")
|
|
time.sleep(.05)
|
|
time.sleep(.2)
|
|
|
|
def wait_title(value):
|
|
wait_for(lambda: any(window["title"] == value for window in json.loads(control("-j", "windows"))))
|
|
|
|
def focused_title():
|
|
return next((window["title"] for window in json.loads(control("-j", "windows"))
|
|
if window["focused"]), None)
|
|
|
|
def watch(enabled):
|
|
matches = re.findall(r"\.set_background_title_updates\(([01])\)", log_path.read_text())
|
|
return matches and matches[-1] == str(int(enabled))
|
|
|
|
try:
|
|
with log_path.open("w") as log:
|
|
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=wgpu", str(shell)],
|
|
stdout=log, stderr=log)
|
|
wait_for(lambda: ipc_socket(runtime).is_socket() and watch(False))
|
|
child_display = wait_for(lambda: next((p for p in runtime.glob("wayland-*") if p.is_socket()), None))
|
|
env["WAYLAND_DISPLAY"] = str(child_display)
|
|
client = launch([str(root / "client")], stdin=subprocess.PIPE, stdout=subprocess.PIPE,
|
|
stderr=log, text=True)
|
|
self.assertTrue(select.select([client.stdout], [], [], 10)[0], "window client did not map")
|
|
self.assertEqual(client.stdout.readline().strip(), "ready")
|
|
wait_for(lambda: len(json.loads(control("-j", "windows"))) == 2)
|
|
wait_for(lambda: focused_title() == "Focused window")
|
|
time.sleep(.5)
|
|
before = snapshots()
|
|
burst(0, "Hidden spinner")
|
|
wait_title("Hidden spinner 19")
|
|
self.assertEqual(snapshots(), before, "background titles repainted the closed panel")
|
|
print("Background: 20 actual title changes produced no shell snapshots", flush=True)
|
|
|
|
before = snapshots()
|
|
title(1, "Current focused title")
|
|
wait_title("Current focused title")
|
|
wait_for(lambda: snapshots() > before)
|
|
self.assertIn('"Current focused title"', log_path.read_text())
|
|
|
|
control("focus-last-window")
|
|
wait_for(lambda: focused_title() == "Hidden spinner 19")
|
|
wait_for(lambda: '"Hidden spinner 19"' in log_path.read_text())
|
|
control("focus-last-window")
|
|
wait_for(lambda: focused_title() == "Current focused title")
|
|
time.sleep(.2)
|
|
|
|
before = snapshots()
|
|
control("launcher", "toggle")
|
|
wait_for(lambda: watch(True) and snapshots() > before)
|
|
self.assertIn('"Hidden spinner 19"', log_path.read_text())
|
|
before = snapshots()
|
|
burst(0, "Visible list")
|
|
wait_title("Visible list 19")
|
|
wait_for(lambda: snapshots() > before)
|
|
self.assertIn('"Visible list 19"', log_path.read_text())
|
|
print("Launcher: opening fetched the latest title; live updates continued", flush=True)
|
|
|
|
control("launcher", "toggle")
|
|
wait_for(lambda: watch(False))
|
|
time.sleep(.5)
|
|
before = snapshots()
|
|
burst(0, "Closed list")
|
|
wait_title("Closed list 19")
|
|
self.assertEqual(snapshots(), before, "closing the launcher left a title subscription active")
|
|
self.assertIsNone(compositor.poll(), log_path.read_text()[-8000:])
|
|
self.assertNotIn("panicked at", log_path.read_text())
|
|
finally:
|
|
if destination := os.environ.get("FERESE_TEST_LOG"):
|
|
Path(destination).write_text(log_path.read_text() if log_path.exists() else "")
|
|
for process in reversed(children):
|
|
if process.poll() is None:
|
|
os.killpg(process.pid, signal.SIGTERM)
|
|
try:
|
|
process.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
os.killpg(process.pid, signal.SIGKILL)
|
|
process.wait()
|
|
for stream in (process.stdin, process.stdout, process.stderr):
|
|
if stream is not None:
|
|
stream.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|