feresaria/scripts/tests/test_xwayland_isolated.py

"""Verify real on-demand/eager XWayland startup and teardown in a nested session.

FERESE_TEST_XWAYLAND=1 python3 scripts/tests/test_xwayland_isolated.py
Requires a Wayland host, Xwayland, xdpyinfo and Satellite built with systemd support.
"""
import json
import os
from pathlib import Path
import shutil
import signal
import subprocess
import sys
import tempfile
import time
import unittest

from session_socket import ipc_environment, ipc_socket


@unittest.skipUnless(os.environ.get("FERESE_TEST_XWAYLAND") == "1", "opt-in nested XWayland test")
class XwaylandTest(unittest.TestCase):
    def test_startup_authorization_and_owned_process_cleanup(self):
        repo = Path(__file__).resolve().parents[2]
        binary = repo / os.environ.get("FERESE_TEST_BINARY", "target/release/ferese")
        ctl = repo / os.environ.get("FERESE_TEST_CTL", "target/release/feresectl")
        satellite = os.environ.get("FERESE_TEST_SATELLITE") or shutil.which("xwayland-satellite")
        self.assertIsNotNone(satellite, "Satellite must be installed; build it with --features systemd")
        satellite = str(Path(satellite).resolve())
        self.assertTrue(Path(satellite).is_file(), f"Satellite does not exist: {satellite}")
        for mode in ("on-demand", "eager"):
            with self.subTest(mode=mode), tempfile.TemporaryDirectory(prefix="ferese-xwayland-") 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(
                    f'xwayland {{ startup "{mode}"; path {json.dumps(satellite)}; }}\n')
                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_") or key.startswith("FERESE_TEST_")}
                for key in ("WAYLAND_SOCKET", "DISPLAY", "XAUTHORITY", "DBUS_SESSION_BUS_ADDRESS"):
                    env.pop(key, None)
                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"))
                # This managed child receives Ferese's reserved display and cookie.
                # Waiting for a gate lets us prove on-demand stays idle until an X client connects.
                client = root / "client.py"
                client.write_text('''import json, os
from pathlib import Path
import subprocess, time
root = Path(__file__).parent
def publish(name, value):
    temporary = root / (name + ".tmp")
    temporary.write_text(json.dumps(value))
    temporary.replace(root / name)
publish("endpoint.json", {
    "display": os.environ["DISPLAY"], "authority": os.environ["XAUTHORITY"]})
deadline = time.monotonic() + 30
while not (root / "connect").exists():
    if time.monotonic() >= deadline:
        raise SystemExit("X client was never released")
    time.sleep(.025)
result = subprocess.run(["xdpyinfo"], capture_output=True, text=True, timeout=15)
publish("client-result.json", {
    "code": result.returncode, "stdout": result.stdout, "stderr": result.stderr})
''')
                log_path = root / "compositor.log"
                with log_path.open("w") as log:
                    compositor = subprocess.Popen([
                        str(binary), "--backend", "nested", "--", sys.executable, str(client),
                    ], env=ipc_environment(env), stdout=log, stderr=log, start_new_session=True)

                    def wait_for(predicate, timeout=15):
                        deadline = time.monotonic() + timeout
                        while time.monotonic() < deadline:
                            value = predicate()
                            if value:
                                return value
                            if compositor.poll() is not None:
                                break
                            time.sleep(.025)
                        self.fail(log_path.read_text()[-12000:])

                    def command(*args):
                        return subprocess.run([str(ctl), "-j", *args], env=ipc_environment(env),
                                              check=True, capture_output=True, text=True, timeout=5)

                    def status_is(state):
                        status = json.loads(command("xwayland", "status").stdout)
                        self.assertNotEqual(status["state"], "failed", status["last_error"])
                        return status if status["state"] == state else None

                    try:
                        wait_for(lambda: ipc_socket(runtime).is_socket())
                        initial = wait_for(lambda: status_is("idle" if mode == "on-demand" else "running"))
                        self.assertEqual(initial["effective_startup"], mode)
                        wait_for(lambda: (root / "endpoint.json").exists())
                        endpoint = json.loads((root / "endpoint.json").read_text())
                        self.assertEqual(endpoint["display"], initial["display"])
                        self.assertTrue(Path(endpoint["authority"]).is_relative_to(runtime))
                        (root / "connect").touch()
                        running = wait_for(lambda: status_is("running"), timeout=10)
                        self.assertEqual(running["readiness"], "verified")
                        self.assertEqual(running["recent_failures"], 0)
                        wait_for(lambda: (root / "client-result.json").exists())
                        result = json.loads((root / "client-result.json").read_text())
                        self.assertEqual(result["code"], 0, result["stderr"])
                        self.assertIn("X.Org", result["stdout"])
                        # Record the complete owned service group before requesting a clean exit.
                        leader = running["satellite_pid"]
                        members = []
                        for proc in Path("/proc").iterdir():
                            if not proc.name.isdigit():
                                continue
                            try:
                                stat = (proc / "stat").read_text()
                                if int(stat[stat.rfind(")") + 1:].split()[2]) == leader:
                                    members.append(proc)
                            except (FileNotFoundError, ProcessLookupError, PermissionError):
                                continue
                        self.assertGreaterEqual(len(members), 2, "Satellite and Xwayland must both be running")
                        command("exit")
                        compositor.wait(timeout=10)
                        self.assertEqual(compositor.returncode, 0, log_path.read_text())
                        self.assertFalse(any(proc.exists() for proc in members), "owned X11 processes survived shutdown")
                        self.assertFalse(Path(endpoint["authority"]).exists(), "Xauthority survived shutdown")
                        slot = endpoint["display"].removeprefix(":")
                        self.assertFalse(Path(f"/tmp/.X11-unix/X{slot}").exists(), "X11 socket survived shutdown")
                        self.assertFalse(Path(f"/tmp/.X{slot}-lock").exists(), "X11 display lock survived shutdown")
                        self.assertNotIn("panicked at", log_path.read_text())
                        print(f"XWayland: {mode} startup, authorized client and complete teardown passed", flush=True)
                    finally:
                        if compositor.poll() is None:
                            os.killpg(compositor.pid, signal.SIGTERM)
                            try:
                                compositor.wait(timeout=10)
                            except subprocess.TimeoutExpired:
                                os.killpg(compositor.pid, signal.SIGKILL)
                                compositor.wait(timeout=5)
                        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"xwayland-{mode}.log")


if __name__ == "__main__":
    unittest.main()