96 lines
4.5 KiB
Python
96 lines
4.5 KiB
Python
"""Exercise shortcut connection lifetimes in an isolated nested compositor."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import signal
|
|
import socket
|
|
import struct
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
REPO = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
@unittest.skipUnless(os.environ.get("FERESE_TEST_SHORTCUTS") == "1", "requires a Wayland host and a built compositor")
|
|
class ShortcutConnections(unittest.TestCase):
|
|
def test_disconnect_conflicts_and_configuration_reload(self):
|
|
host_runtime = Path(os.environ["XDG_RUNTIME_DIR"])
|
|
host_display = Path(os.environ["WAYLAND_DISPLAY"])
|
|
if not host_display.is_absolute():
|
|
host_display = host_runtime / host_display
|
|
with tempfile.TemporaryDirectory(prefix="ferese-shortcut-test-") as directory:
|
|
root = Path(directory)
|
|
runtime = root / "runtime"
|
|
runtime.mkdir(mode=0o700)
|
|
config = root / "config/ferese/config.kdl"
|
|
config.parent.mkdir(parents=True)
|
|
config.write_text("")
|
|
log = (root / "compositor.log").open("w")
|
|
env = dict(os.environ, XDG_RUNTIME_DIR=str(runtime), XDG_CONFIG_HOME=str(root / "config"), WAYLAND_DISPLAY=str(host_display))
|
|
env.pop("WAYLAND_SOCKET", None)
|
|
env.pop("FERESE_SHELL_CONTROL_SOCKET", None)
|
|
process = subprocess.Popen([str(REPO / "target/debug/ferese"), "--backend=nested"], env=env,
|
|
stdout=log, stderr=log, start_new_session=True)
|
|
connections = []
|
|
try:
|
|
deadline = time.monotonic() + 15
|
|
while not (runtime / "ferese/control.sock").exists():
|
|
if process.poll() is not None or time.monotonic() >= deadline:
|
|
self.fail((root / "compositor.log").read_text())
|
|
time.sleep(0.02)
|
|
|
|
def connect():
|
|
connection = socket.socket(socket.AF_UNIX)
|
|
connection.settimeout(2)
|
|
connection.connect(str(runtime / "ferese/control.sock"))
|
|
connections.append(connection)
|
|
return connection
|
|
|
|
def call(connection, command, args=None):
|
|
data = json.dumps({"version": 1, "id": 1, "type": "command", "command": command, "args": args}).encode()
|
|
connection.sendall(struct.pack(">I", len(data)) + data)
|
|
|
|
def receive(length):
|
|
data = b""
|
|
while len(data) < length:
|
|
chunk = connection.recv(length - len(data))
|
|
if not chunk:
|
|
raise AssertionError("IPC connection closed")
|
|
data += chunk
|
|
return data
|
|
|
|
length, = struct.unpack(">I", receive(4))
|
|
self.assertLessEqual(length, 1024 * 1024)
|
|
return json.loads(receive(length))
|
|
|
|
first = connect()
|
|
second = connect()
|
|
shortcut = [{"id": "record", "trigger": "Ctrl+Alt+a"}]
|
|
self.assertNotIn("error", call(first, "portal-shortcuts-register", shortcut))
|
|
conflict = call(second, "portal-shortcuts-register", shortcut)
|
|
self.assertEqual(conflict["error"]["code"], "shortcut_conflict")
|
|
self.assertTrue(call(second, "portal-shortcuts-poll")["result"]["closed"])
|
|
first.close()
|
|
deadline = time.monotonic() + 3
|
|
while "error" in call(second, "portal-shortcuts-register", shortcut):
|
|
self.assertLess(time.monotonic(), deadline, "disconnected owner kept its binding")
|
|
time.sleep(0.02)
|
|
self.assertEqual(call(second, "portal-shortcuts-poll")["result"]["shortcuts"], shortcut)
|
|
config.write_text('binding keys="Ctrl+Alt+a" action="none"\n')
|
|
deadline = time.monotonic() + 5
|
|
while call(second, "portal-shortcuts-poll")["result"]["shortcuts"]:
|
|
self.assertLess(time.monotonic(), deadline, "configuration did not revoke conflicting shortcut")
|
|
time.sleep(0.05)
|
|
finally:
|
|
for connection in connections:
|
|
connection.close()
|
|
os.killpg(process.pid, signal.SIGTERM)
|
|
process.wait(timeout=5)
|
|
log.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|