gxde-ai-see/tests/portal/run.py

#!/usr/bin/env python3
# SPDX-FileCopyrightText: 2026 GXDE OS Team
# SPDX-License-Identifier: GPL-3.0-or-later
"""End-to-end test for the GXDE portal (native drop into a window).

Run inside a private session bus:

    dbus-run-session -- python3 tests/portal/run.py [compositor]

It starts a nested GXDE compositor, a GTK drag source and a GTK drop target,
drives the pointer through wlr-virtual-pointer, asks the compositor to open the
portal overview and asserts that the drop lands in the target window.
"""

import glob
import json
import os
import subprocess
import sys
import time

HERE = os.path.dirname(os.path.abspath(__file__))
RUNTIME = f"/run/user/{os.getuid()}"
PORTAL = ["org.gxde.Portal", "/org/gxde/Portal", "org.gxde.Portal"]


def log(message):
    print(f"[test] {message}", flush=True)


def wait_for(predicate, timeout=20.0, interval=0.1):
    deadline = time.time() + timeout
    while time.time() < deadline:
        value = predicate()
        if value:
            return value
        time.sleep(interval)
    return None


def portal_call(method, *args):
    address = os.environ.get("DBUS_SESSION_BUS_ADDRESS", "")
    cmd = ["busctl", f"--address={address}", "--json=short", "call", *PORTAL, method, *args]
    out = subprocess.check_output(cmd, text=True)
    return json.loads(out)


def list_windows():
    data = portal_call("ListWindows")["data"]
    rows = data[0] if data else []
    return [
        {
            "uuid": row[0],
            "title": row[1],
            "app": row[2],
            "x": row[3],
            "y": row[4],
            "w": row[5],
            "h": row[6],
        }
        for row in rows
    ]


def layout():
    data = portal_call("GetLayout")["data"]
    rows = data[0] if data else []
    return [
        {
            "uuid": row[0],
            "title": row[1],
            "x": row[3],
            "y": row[4],
            "w": row[5],
            "h": row[6],
        }
        for row in rows
    ]


def find_by_title(windows, title):
    for window in windows:
        if window["title"] == title:
            return window
    return None


def main():
    compositor = (
        sys.argv[1]
        if len(sys.argv) > 1
        else os.path.expanduser("~/playground/flake-wm/build-portal/flakewm")
    )
    parent_display = os.environ.get("WAYLAND_DISPLAY", "wayland-0")

    before = set(glob.glob(f"{RUNTIME}/wayland-*"))
    env = os.environ.copy()
    env["WAYLAND_DISPLAY"] = parent_display

    compositor_log = open("/tmp/pt-comp.log", "wb")
    compositor_proc = subprocess.Popen(
        [compositor], env=env, stdout=compositor_log, stderr=subprocess.STDOUT
    )

    procs = [compositor_proc]
    pointer = None
    try:
        def nested_display():
            for path in set(glob.glob(f"{RUNTIME}/wayland-*")) - before:
                if not path.endswith(".lock"):
                    return os.path.basename(path)
            return None

        display = wait_for(nested_display, 25.0)
        assert display, "nested compositor did not create a Wayland socket"
        log(f"nested display: {display}")
        env["WAYLAND_DISPLAY"] = display

        target_log = open("/tmp/pt-target.log", "wb")
        source_log = open("/tmp/pt-source.log", "wb")
        target = subprocess.Popen(
            [sys.executable, f"{HERE}/drop_target.py"],
            env=env, stdout=target_log, stderr=subprocess.STDOUT,
        )
        source = subprocess.Popen(
            [sys.executable, f"{HERE}/drag_source.py"],
            env=env, stdout=source_log, stderr=subprocess.STDOUT,
        )
        procs += [target, source]

        def portal_ready():
            try:
                portal_call("ListWindows")
                return True
            except subprocess.CalledProcessError:
                return False

        assert wait_for(portal_ready, 20.0, 0.3), "org.gxde.Portal never appeared"
        log("portal D-Bus ready")

        windows = wait_for(
            lambda: list_windows()
            if find_by_title(list_windows(), "PortalDragSource")
            and find_by_title(list_windows(), "PortalDropTarget")
            else None,
            20.0,
            0.2,
        )
        assert windows, "windows did not appear"
        log(f"windows: {json.dumps(windows)}")

        src = find_by_title(windows, "PortalDragSource")
        dst = find_by_title(windows, "PortalDropTarget")

        pointer = subprocess.Popen(
            [f"{HERE}/pointer"], env=env, stdin=subprocess.PIPE,
            stdout=subprocess.PIPE, text=True, bufsize=1,
        )
        procs.append(pointer)

        def pcmd(line):
            pointer.stdin.write(line + "\n")
            pointer.stdin.flush()
            response = pointer.stdout.readline().strip()
            if not (response.startswith("ok") or response.startswith("size")):
                raise RuntimeError(f"pointer failed: {response!r}")
            return response

        log(f"pointer {pcmd('size')}")

        sx = src["x"] + src["w"] / 2
        sy = src["y"] + src["h"] / 2
        pcmd(f"move {sx:.0f} {sy:.0f}")
        time.sleep(0.2)
        pcmd("press")
        time.sleep(0.2)
        pcmd(f"move {sx + 60:.0f} {sy:.0f}")
        time.sleep(0.8)

        if "DRAG-BEGIN" not in open("/tmp/pt-source.log").read():
            raise RuntimeError("GTK drag did not start; see /tmp/pt-source.log")
        log("drag started")

        started = portal_call("BeginPortal")["data"][0]
        assert started, "BeginPortal returned false"
        log("portal opened")

        time.sleep(0.7)  # let the overview animation settle
        tiles = wait_for(lambda: layout() or None, 5.0, 0.2)
        assert tiles, "portal layout is empty"
        log(f"layout: {json.dumps(tiles)}")

        tile = find_by_title(tiles, "PortalDropTarget") or find_by_title(tiles, "PortalDragSource")
        assert tile, "target window not in the portal layout"

        # If we only see the source tile, the drop still validates retargeting.
        tx = tile["x"] + tile["w"] / 2
        ty = tile["y"] + tile["h"] / 2
        pcmd(f"move {tx:.0f} {ty:.0f}")
        time.sleep(0.6)

        subprocess.run(["grim", "/tmp/pt-before-drop.png"], env=env, check=False)
        pcmd("release")
        time.sleep(1.0)
        subprocess.run(["grim", "/tmp/pt-after-drop.png"], env=env, check=False)

        target_text = open("/tmp/pt-target.log").read()
        log(f"target log: {target_text.strip()!r}")
        assert "DROP" in target_text, "the drop did not reach the target window"
        assert "portal-payload-42" in target_text, "wrong payload"

        log("PASS: native drop landed in the target window")
        return 0
    except Exception as exc:  # noqa: BLE001
        log(f"FAIL: {exc}")
        for path in ("/tmp/pt-comp.log", "/tmp/pt-source.log", "/tmp/pt-target.log"):
            try:
                content = open(path).read().strip()
            except OSError:
                continue
            if content:
                log(f"--- {path} ---\n{content[-2000:]}")
        return 1
    finally:
        if pointer is not None:
            try:
                pointer.stdin.write("quit\n")
                pointer.stdin.flush()
            except Exception:  # noqa: BLE001
                pass
        for proc in reversed(procs):
            if proc.poll() is None:
                proc.terminate()
        for proc in reversed(procs):
            try:
                proc.wait(timeout=5)
            except Exception:  # noqa: BLE001
                proc.kill()


if __name__ == "__main__":
    sys.exit(main())