Add a dbusmock-based Python test suite

For complex portals like the InputCapture portal, the current test suite
is difficult to use and adapt. A much simpler approach is to use
dbusmock to provide the impl.portal emulation, use Python to poke at the
actual DBus API of the portal and have xdg-desktop-portal sit in
between to negotiate the two.

This patch adds a test suite that provides enough infrastructure to add
tests for other portals in a similar manner - luckily the impl side is
relatively trivial to implement in dbusmock.

tests/__init__.py

@@ -0,0 +1,239 @@
+#!/usr/bin/env python3
+#
+# SPDX-License-Identifier: LGPL-2.1-or-later
+#
+# This file is formatted with Python Black
+#
+
+# Shared setup for portal tests. To test a portal, subclass TestPortal with
+# your portal's name (e.g. TestInputCapture). This will auto-fill your portal
+# name into some of the functions.
+#
+# Make sure you have a dbusmock template for the impl.portal of your portal in
+# tests/templates. See the dbusmock documentaiton for details on that.
+#
+# Environment variables:
+#   XDP_DBUS_MONITOR: if set, starts dbus_monitor on the custom bus, useful
+#                     for debugging
+#   XDP_UNINSTALLED: run from $PWD (with an emulation of the glib test behavior)
+#   LIBEXECDIR: run xdg-desktop-portal from that dir, required if
+#               XDP_UNINSTALLEDis unset
+#
+
+from dbus.mainloop.glib import DBusGMainLoop
+from gi.repository import GLib
+from itertools import count
+from typing import Any, Dict, Tuple
+from pathlib import Path
+
+import dbus
+import dbusmock
+import fcntl
+import logging
+import os
+import subprocess
+import time
+
+DBusGMainLoop(set_as_default=True)
+
+_counter = count()
+
+ASV = Dict[str, Any]
+
+logger = logging.getLogger("tests")
+
+
+class Request:
+    """
+    Helper class for executing methods that use Requests.
+    """
+    def __init__(self, bus: dbus.Bus, interface: dbus.Interface):
+        self.interface = interface
+        self._handle_token = f"request{next(_counter)}"
+        self.response = -1
+        self.results = None
+
+        def sanitize(name):
+            return name.lstrip(':').replace('.', '_')
+
+        sender_token = sanitize(bus.get_unique_name())
+        self.handle = f"/org/freedesktop/portal/desktop/request/{sender_token}/{self._handle_token}"
+        proxy = bus.get_object(
+            "org.freedesktop.portal.Desktop", self.handle
+        )
+        self.request_interface = dbus.Interface(proxy, "org.freedesktop.portal.Request")
+
+        def cb_response(response, results):
+            self.response = response
+            self.results = results
+
+        self.request_interface.connect_to_signal("Response", cb_response)
+
+    @property
+    def handle_token(self) -> dbus.String:
+        """
+        Returns the dbus-ready handle_token, ready to be put into the options
+        """
+        return dbus.String(self._handle_token, variant_level=1)
+
+    def call(self, methodname, **kwargs) -> Tuple[int, ASV]:
+        """
+        Synchronously call method ``methodname`` on the interface given in the
+        request's constructor. The kwargs must be specified in the order the
+        DBus method takes them but the handle_token is automatically filled
+        in.
+        """
+        try:
+            options = kwargs["options"]
+        except KeyError:
+            options = dbus.Dictionary({}, signature="sv")
+
+        if "handle_token" not in options:
+            options["handle_token"] = self.handle_token
+
+        method = getattr(self.interface, methodname)
+        assert method
+        handle = method(*list(kwargs.values()))
+        assert handle.endswith(self._handle_token)
+
+        # Wait 300ms for the Response signal
+        def timeout():
+            loop.quit()
+
+        loop = GLib.MainLoop()
+        GLib.timeout_add(300, timeout)
+        loop.run()
+
+        return self.response, self.results
+
+
+class TestPortal(dbusmock.DBusTestCase):
+    @classmethod
+    def setUpClass(cls):
+        logging.basicConfig(
+            format="%(levelname).1s|%(name)s: %(message)s", level=logging.DEBUG
+        )
+
+        cls.start_session_bus()
+        cls.dbus_con = cls.get_dbus(system_bus=False)
+        assert cls.dbus_con
+
+    def setUp(self):
+        self.p_mock = None
+        self.xdp = None
+        self._portal_interfaces = {}
+
+    def start_impl_portal(self, params=None, portal=None):
+        """
+        Start the impl.portal for the given portal name. If missing,
+        the portal name is derived from the class name of the test, e.g.
+        TestFoo will start impl.portal.Foo.
+        """
+        if portal is None:
+            portal = type(self).__name__.removeprefix("Test")
+        self.p_mock, self.obj_portal = self.spawn_server_template(
+            template=f"tests/templates/{portal.lower()}.py",
+            parameters=params,
+            stdout=subprocess.PIPE,
+        )
+        flags = fcntl.fcntl(self.p_mock.stdout, fcntl.F_GETFL)
+        fcntl.fcntl(self.p_mock.stdout, fcntl.F_SETFL, flags | os.O_NONBLOCK)
+        self.mock_interface = dbus.Interface(self.obj_portal, dbusmock.MOCK_IFACE)
+
+        self.start_dbus_monitor()
+
+    def start_xdp(self):
+        """
+        Start the xdg-desktop-portal process
+        """
+
+        # This roughly resembles test-portals.c and glib's test behavior
+        if os.getenv("XDP_UNINSTALLED", None):
+            builddir = Path(os.getenv("G_TEST_BUILDDIR") or "tests") / ".."
+        else:
+            builddir = os.getenv("LIBEXECDIR")
+            if not builddir:
+                raise NotImplementedError("LIBEXECDIR is not set")
+
+        distdir = os.getenv("G_TEST_SRCDIR") or "tests"
+        portal_dir = Path(distdir) / "portals"
+
+        argv = [Path(builddir) / "xdg-desktop-portal"]
+        env = os.environ.copy()
+        env["G_DEBUG"] = "fatal-criticals"
+        env["XDG_DESKTOP_PORTAL_DIR"] = portal_dir
+
+        xdp = subprocess.Popen(argv, env=env)
+
+        timeout = 500
+        while timeout > 0:
+            if self.dbus_con.name_has_owner("org.freedesktop.portal.Desktop"):
+                break
+            timeout -= 1
+            time.sleep(0.1)
+
+        assert timeout > 0, "Timeout while waiting for xdg-desktop-portal to claim the bus"
+
+        self.xdp = xdp
+
+    def start_dbus_monitor(self):
+        self.dbus_monitor = None
+        if not os.getenv("XDP_DBUS_MONITOR"):
+            return
+
+        argv = ['dbus-monitor', '--session']
+        self.dbus_monitor = subprocess.Popen(argv);
+
+    def tearDown(self):
+        if self.dbus_monitor:
+            self.dbus_monitor.terminate()
+            self.dbus_monitor.wait()
+
+        if self.xdp:
+            self.xdp.terminate()
+            self.xdp.wait()
+
+        if self.p_mock:
+            if self.p_mock.stdout:
+                out = (self.p_mock.stdout.read() or b"").decode("utf-8")
+                if out:
+                    print(out)
+                self.p_mock.stdout.close()
+            self.p_mock.terminate()
+            self.p_mock.wait()
+
+    def get_xdp_dbus_object(self) -> dbus.proxies.ProxyObject:
+        """
+        Return the object that is the org.freedesktop.portal.Desktop proxy
+        """
+        try:
+            return self._xdp_dbus_object
+        except AttributeError:
+            obj = self.dbus_con.get_object(
+                "org.freedesktop.portal.Desktop", "/org/freedesktop/portal/desktop"
+            )
+            # Useful for debugging:
+            # print(obj.Introspect(dbus_interface="org.freedesktop.DBus.Introspectable"))
+            assert obj
+            self._xdp_dbus_object = obj
+            return self._xdp_dbus_object
+
+    def get_dbus_interface(self, name=None) -> dbus.Interface:
+        """
+        Return the interface with the given name.
+
+        For portals, it's enough to specify the portal name (e.g. "InputCapture").
+        If no name is provided, guess from the test class name.
+        """
+        if name is None:
+            name = type(self).__name__.removeprefix("Test")
+        if "." not in name:
+            name = f"org.freedesktop.portal.{name}"
+
+        try:
+            intf = self._portal_interfaces[name]
+        except KeyError:
+            intf = dbus.Interface(self.get_xdp_dbus_object(), name)
+            assert intf
+            self._portal_interfaces[name] = intf
+        return intf

tests/portals/test.portal

@@ -1,4 +1,4 @@
 [portal]
 DBusName=org.freedesktop.impl.portal.Test
-Interfaces=org.freedesktop.impl.portal.Account;org.freedesktop.impl.portal.Email;org.freedesktop.impl.portal.FileChooser;org.freedesktop.impl.portal.Screenshot;org.freedesktop.impl.portal.Lockdown;org.freedesktop.impl.portal.Print;org.freedesktop.impl.portal.Access;org.freedesktop.impl.portal.Inhibit;org.freedesktop.impl.portal.AppChooser;org.freedesktop.impl.portal.Wallpaper;org.freedesktop.impl.portal.Background;org.freedesktop.impl.portal.Notification;org.freedesktop.impl.portal.Settings;
+Interfaces=org.freedesktop.impl.portal.Account;org.freedesktop.impl.portal.Email;org.freedesktop.impl.portal.FileChooser;org.freedesktop.impl.portal.Screenshot;org.freedesktop.impl.portal.Lockdown;org.freedesktop.impl.portal.Print;org.freedesktop.impl.portal.Access;org.freedesktop.impl.portal.Inhibit;org.freedesktop.impl.portal.AppChooser;org.freedesktop.impl.portal.Wallpaper;org.freedesktop.impl.portal.Background;org.freedesktop.impl.portal.Notification;org.freedesktop.impl.portal.Settings;org.freedesktop.impl.portal.InputCapture;
 UseIn=test

tests/templates/__init__.py

@@ -0,0 +1,7 @@
+# SPDX-License-Identifier: LGPL-2.1-or-later
+#
+# This file is formatted with Python Black
+
+import logging
+
+logging.basicConfig(format="%(levelname).1s|%(name)s: %(message)s", level=logging.DEBUG)