Release layout slots for unbuffered XDG windows

scripts/tests/test_unmapped_toplevel_isolated.py

@@ -0,0 +1,128 @@
+"""Opt-in XDG buffer lifecycle test for a disposable nested compositor.
+
+FERESE_TEST_UNMAPPED=1 python3 scripts/tests/test_unmapped_toplevel_isolated.py
+Requires cc, pkg-config, wayland-scanner, and wayland-protocols.
+"""
+
+import json
+import os
+from pathlib import Path
+import select
+import subprocess
+import tempfile
+import time
+import unittest
+
+
+@unittest.skipUnless(os.environ.get("FERESE_TEST_UNMAPPED") == "1", "opt-in nested test")
+class UnmappedToplevelTest(unittest.TestCase):
+    def test_only_buffered_toplevels_take_layout_space(self):
+        repo = Path(__file__).resolve().parents[2]
+        with tempfile.TemporaryDirectory(prefix="ferese-unmapped-test-") 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 = subprocess.check_output(
+                ["pkg-config", "--cflags", "--libs", "wayland-client"], text=True
+            ).split()
+            subprocess.run(
+                [
+                    "cc", "-Wall", "-Wextra", "-I", str(root),
+                    str(repo / "scripts/tests/fixtures/unmapped-toplevel.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("\n")
+            host_display = os.environ["WAYLAND_DISPLAY"]
+            if not host_display.startswith("/"):
+                host_display = str(Path(os.environ["XDG_RUNTIME_DIR"]) / host_display)
+            env = dict(
+                os.environ,
+                XDG_RUNTIME_DIR=str(runtime),
+                XDG_CONFIG_HOME=str(root / "config"),
+                WAYLAND_DISPLAY=host_display,
+            )
+            compositor_binary = repo / os.environ.get("FERESE_TEST_BINARY", "target/debug/ferese")
+            ctl_binary = repo / os.environ.get("FERESE_TEST_CTL", "target/debug/feresectl")
+            with (root / "compositor.log").open("w") as log:
+                compositor = subprocess.Popen(
+                    [str(compositor_binary), "--backend", "nested"],
+                    env=env,
+                    stdout=log,
+                    stderr=subprocess.STDOUT,
+                )
+                client = None
+                try:
+                    for _ in range(100):
+                        if compositor.poll() is not None:
+                            self.fail((root / "compositor.log").read_text())
+                        sockets = [
+                            path for path in runtime.glob("wayland-*")
+                            if not path.name.endswith(".lock")
+                        ]
+                        if sockets:
+                            break
+                        time.sleep(0.1)
+                    self.assertTrue(sockets, "nested Wayland socket did not appear")
+                    client = subprocess.Popen(
+                        [str(root / "client")],
+                        env=dict(env, WAYLAND_DISPLAY=str(sockets[0])),
+                        stdin=subprocess.PIPE,
+                        stdout=subprocess.PIPE,
+                        text=True,
+                    )
+
+                    for phase, visible in [
+                        ("initial-empty", False),
+                        ("visible", True),
+                        ("detached", False),
+                        ("remapped", True),
+                    ]:
+                        ready, _, _ = select.select([client.stdout], [], [], 10)
+                        self.assertTrue(ready, f"client did not reach {phase}")
+                        self.assertEqual(client.stdout.readline().strip(), phase)
+                        deadline = time.monotonic() + 3
+                        while True:
+                            result = subprocess.run(
+                                [str(ctl_binary), "get-focused-window"],
+                                env=env,
+                                capture_output=True,
+                                text=True,
+                                check=True,
+                            )
+                            focused = json.loads(result.stdout)
+                            if (focused is not None and focused.get("id") is not None) == visible:
+                                break
+                            if time.monotonic() >= deadline:
+                                self.fail(f"{phase}: unexpected focused window {focused}")
+                            time.sleep(0.05)
+                        client.stdin.write("\n")
+                        client.stdin.flush()
+                    self.assertEqual(client.wait(timeout=5), 0)
+                finally:
+                    if client and client.poll() is None:
+                        client.terminate()
+                        client.wait(timeout=5)
+                    if client:
+                        client.stdin.close()
+                        client.stdout.close()
+                    compositor.terminate()
+                    compositor.wait(timeout=5)
+
+
+if __name__ == "__main__":
+    unittest.main()