"""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()