Add python tests for all portals that have a C test

tests/.mypy.ini

@@ -4,7 +4,7 @@ check_untyped_defs=True
 files=.
 exclude = (?x)(
     templates/.*.py # template files are a bit special
-    |test-document-fuse.py$ # has issues with typing
+    |test_document_fuse.py$ # has issues with typing
   )
 
 [mypy-gi.*]

tests/README.md

@@ -15,6 +15,9 @@ but should not normally be set on production systems:
     reliable are skipped.
     Set this for automated QA testing, leave it unset during development.
 
+* `XDP_TEST_RUN_LONG`: If set (to any value), some tests will run more
+    iterations or otherwise test more thoroughly.
+
 * `XDP_VALIDATE_ICON_INSECURE`: If set (to any value), x-d-p doesn't
     sandbox the icon validator using **bwrap**(1), even if sandboxed
     validation was enabled at compile time.

tests/__init__.py

@@ -5,9 +5,9 @@
 # This file is formatted with Python Black
 
 from dbus.mainloop.glib import DBusGMainLoop
-from gi.repository import GLib
+from gi.repository import GLib, Gio
 from itertools import count
-from typing import Any, Dict, Optional, NamedTuple, Callable
+from typing import Any, Dict, Optional, NamedTuple, Callable, List
 
 import os
 import dbus
@@ -39,6 +39,10 @@ def is_in_container() -> bool:
     )
 
 
+def run_long_tests() -> bool:
+    return os.environ.get("XDP_TEST_RUN_LONG") is not None
+
+
 def wait(ms: int):
     """
     Waits for the specified amount of milliseconds.
@@ -429,3 +433,123 @@ def closed(self):
     @classmethod
     def from_response(cls, bus: dbus.Bus, response: Response) -> "Session":
         return cls(bus, response.results["session_handle"])
+
+
+class GDBusIfaceSignal:
+    """
+    Helper class which represents a connected signal on a GDBusIface and can be
+    used to disconnect from the signal.
+    """
+
+    def __init__(self, signal_id: int, proxy: Gio.DBusProxy):
+        self.signal_id = signal_id
+        self.proxy = proxy
+
+    def disconnect(self):
+        """
+        Disconnects the signal
+        """
+        self.proxy.disconnect(self.signal_id)
+
+
+class GDBusIface:
+    """
+    Helper class for calling dbus interfaces with complex arguments.
+    Usually you want to use python-dbus on the dbus_con fixture with
+    get_portal_iface , get_mock_iface or get_iface. This is convenient but
+    might not be sufficient for complex arguments or for asynchronously calling
+    a method.
+    """
+
+    def __init__(self, bus: str, obj: str, iface: str):
+        """
+        Creates a GDBusIface for a specific bus, object and interface on the
+        session bus.
+        """
+        address = Gio.dbus_address_get_for_bus_sync(Gio.BusType.SESSION, None)
+        session_bus = Gio.DBusConnection.new_for_address_sync(
+            address,
+            Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT
+            | Gio.DBusConnectionFlags.MESSAGE_BUS_CONNECTION,
+            None,
+            None,
+        )
+        assert session_bus
+        self._proxy = Gio.DBusProxy.new_sync(
+            session_bus,
+            Gio.DBusProxyFlags.NONE,
+            None,
+            bus,
+            obj,
+            iface,
+            None,
+        )
+
+    def _call(
+        self, method_name: str, args_variant: GLib.Variant, fds: List[int] = []
+    ) -> GLib.Variant:
+        """
+        Calls a method synchronously with the arguments passed in args_variant,
+        passing the file descriptors specified in fds.
+        Returns the result of the dbus call.
+        """
+        fdlist = Gio.UnixFDList.new()
+        for fd in fds:
+            fdlist.append(fd)
+
+        return self._proxy.call_with_unix_fd_list_sync(
+            method_name,
+            args_variant,
+            0,
+            -1,
+            fdlist,
+            None,
+        )
+
+    def _call_async(
+        self,
+        method_name: str,
+        args_variant: GLib.Variant,
+        fds: List[int] = [],
+        cb: Optional[Callable[[GLib.Variant], None]] = None,
+    ) -> None:
+        """
+        Calls a method asynchronously with the arguments passed in args_variant,
+        passing the file descriptors specified in fds.
+        Invokes the callback cb when the call finished.
+        """
+        fdlist = Gio.UnixFDList.new()
+        for fd in fds:
+            fdlist.append(fd)
+
+        def internal_cb(s, res, _):
+            res = s.call_finish(res)
+            if cb:
+                cb(res)
+
+        self._proxy.call_with_unix_fd_list(
+            method_name,
+            args_variant,
+            0,
+            -1,
+            fdlist,
+            None,
+            internal_cb,
+            None,
+        )
+
+    def connect_to_signal(
+        self, name: str, cb: Callable[[GLib.Variant], None]
+    ) -> GDBusIfaceSignal:
+        """
+        Connects to the dbus signal name to the callback cb. Returns an object
+        representing the connection which can be used to disconnect it again.
+        """
+
+        def internal_cb(proxy, sender_name, signal_name, parameters):
+            if signal_name != name:
+                return
+            cb(parameters)
+
+        signal_id = self._proxy.connect("g-signal", internal_cb)
+        return GDBusIfaceSignal(signal_id, self._proxy)

tests/meson.build

@@ -212,15 +212,6 @@ foreach p : portal_limited
     )
 endforeach
 
-if enable_installed_tests
-  install_data(
-    'session.conf.in',
-    'test-document-fuse.sh',
-    'test-document-fuse.py',
-    install_dir: installed_tests_dir
-  )
-endif
-
 test_permission_store = executable(
   'test-permission-store',
   'test-permission-store.c',
@@ -311,15 +302,27 @@ if enable_pytest
   endif
 
   pytest_files = [
+    'test_account.py',
+    'test_background.py',
+    'test_camera.py',
     'test_clipboard.py',
+    'test_document_fuse.py',
     'test_email.py',
+    'test_filechooser.py',
     'test_globalshortcuts.py',
+    'test_inhibit.py',
     'test_inputcapture.py',
     'test_location.py',
+    'test_notification.py',
+    'test_openuri.py',
+    'test_permission_store.py',
+    'test_print.py',
     'test_remotedesktop.py',
     'test_settings.py',
+    'test_screenshot.py',
     'test_trash.py',
     'test_usb.py',
+    'test_wallpaper.py',
   ]
   foreach pytest_file : pytest_files
     testname = pytest_file.replace('.py', '')
@@ -345,7 +348,6 @@ if enable_installed_tests
   testfiles = [
     'testdb',
     'test-doc-portal',
-    'test-document-fuse.sh',
     'test-permission-store',
     'test-xdp-utils',
   ]

tests/templates/access.py

@@ -0,0 +1,71 @@
+# SPDX-License-Identifier: LGPL-2.1-or-later
+#
+# This file is formatted with Python Black
+
+from tests.templates import Response, init_template_logger, ImplRequest
+
+import dbus.service
+from gi.repository import GLib
+
+
+BUS_NAME = "org.freedesktop.impl.portal.Test"
+MAIN_OBJ = "/org/freedesktop/portal/desktop"
+SYSTEM_BUS = False
+MAIN_IFACE = "org.freedesktop.impl.portal.Access"
+
+
+logger = init_template_logger(__name__)
+
+
+def load(mock, parameters={}):
+    logger.debug(f"Loading parameters: {parameters}")
+
+    mock.delay: int = parameters.get("delay", 200)
+    mock.response: int = parameters.get("response", 0)
+    mock.expect_close: bool = parameters.get("expect-close", False)
+
+
+@dbus.service.method(
+    MAIN_IFACE,
+    in_signature="osssssa{sv}",
+    out_signature="ua{sv}",
+    async_callbacks=("cb_success", "cb_error"),
+)
+def AccessDialog(
+    self,
+    handle,
+    app_id,
+    parent_window,
+    title,
+    subtitle,
+    body,
+    options,
+    cb_success,
+    cb_error,
+):
+    try:
+        logger.debug(
+            f"AccessDialog({handle}, {app_id}, {parent_window}, {title}, {subtitle}, {body}, {options})"
+        )
+
+        def closed_callback():
+            response = Response(2, {})
+            logger.debug(f"AccessDialog Close() response {response}")
+            cb_success(response.response, response.results)
+
+        def reply_callback():
+            response = Response(self.response, {})
+            logger.debug(f"AccessDialog with response {response}")
+            cb_success(response.response, response.results)
+
+        request = ImplRequest(self, BUS_NAME, handle)
+        if self.expect_close:
+            request.export(closed_callback)
+        else:
+            request.export()
+
+            logger.debug(f"scheduling delay of {self.delay}")
+            GLib.timeout_add(self.delay, reply_callback)
+    except Exception as e:
+        logger.critical(e)
+        cb_error(e)

tests/templates/account.py

@@ -0,0 +1,57 @@
+# SPDX-License-Identifier: LGPL-2.1-or-later
+#
+# This file is formatted with Python Black
+
+from tests.templates import Response, init_template_logger, ImplRequest
+import dbus.service
+import dbus
+from gi.repository import GLib
+
+BUS_NAME = "org.freedesktop.impl.portal.Test"
+MAIN_OBJ = "/org/freedesktop/portal/desktop"
+SYSTEM_BUS = False
+MAIN_IFACE = "org.freedesktop.impl.portal.Account"
+
+logger = init_template_logger(__name__)
+
+
+def load(mock, parameters={}):
+    logger.debug(f"Loading parameters: {parameters}")
+
+    mock.delay: int = parameters.get("delay", 200)
+    mock.response: int = parameters.get("response", 0)
+    mock.results: bool = parameters.get("results", {})
+    mock.expect_close: bool = parameters.get("expect-close", False)
+
+
+@dbus.service.method(
+    MAIN_IFACE,
+    in_signature="ossa{sv}",
+    out_signature="ua{sv}",
+    async_callbacks=("cb_success", "cb_error"),
+)
+def GetUserInformation(self, handle, app_id, window, options, cb_success, cb_error):
+    try:
+        logger.debug(f"GetUserInformation({handle}, {app_id}, {window}, {options})")
+
+        def closed_callback():
+            response = Response(2, {})
+            logger.debug(f"GetUserInformation Close() response {response}")
+            cb_success(response.response, response.results)
+
+        def reply_callback():
+            response = Response(self.response, self.results)
+            logger.debug(f"GetUserInformation with response {response}")
+            cb_success(response.response, response.results)
+
+        request = ImplRequest(self, BUS_NAME, handle)
+        if self.expect_close:
+            request.export(closed_callback)
+        else:
+            request.export()
+
+            logger.debug(f"scheduling delay of {self.delay}")
+            GLib.timeout_add(self.delay, reply_callback)
+    except Exception as e:
+        logger.critical(e)
+        cb_error(e)