tests/usb: Test the USB portal via pytest, dbusmock and umockdev

.github/workflows/Containerfile

@@ -47,7 +47,11 @@ RUN apt install -y --no-install-recommends \
     python3-pytest \
     python3-pytest-xdist \
     python3-dbusmock \
-    python3-dbus
+    python3-dbus \
+    libumockdev0 \
+    libumockdev-dev \
+    umockdev \
+    gir1.2-umockdev-1.0
 
 # Install pip
 RUN apt install -y --no-install-recommends python3-pip

.github/workflows/container.yml

@@ -1,5 +1,5 @@
 env:
-  IMAGE_TAG: 20241024-1
+  IMAGE_TAG: 20241203-1
 
 on:
   workflow_call:

meson.build

@@ -133,6 +133,7 @@ libportal_dep = dependency(
 pipewire_dep = dependency('libpipewire-0.3', version: '>= 0.2.90')
 libsystemd_dep = dependency('libsystemd', required: get_option('systemd'))
 gudev_dep = dependency('gudev-1.0', required: get_option('gudev'))
+umockdev_dep = dependency('umockdev-1.0')
 
 bwrap = find_program('bwrap', required: get_option('sandboxed-image-validation').allowed() or get_option('sandboxed-sound-validation').allowed())
 

tests/meson.build

@@ -277,7 +277,11 @@ python = pymod.find_installation(
   required: get_option('pytest'),
 )
 
-enable_pytest = pytest.found() and python.found() and python.language_version().version_compare('>=3.9')
+enable_pytest = \
+  pytest.found() and \
+  python.found() and \
+  python.language_version().version_compare('>=3.9') and \
+  umockdev_dep.found()
 
 if enable_pytest
   subdir('templates')
@@ -304,6 +308,7 @@ if enable_pytest
     'test_location.py',
     'test_remotedesktop.py',
     'test_trash.py',
+    'test_usb.py',
   ]
   foreach pytest_file : pytest_files
     configure_file(

tests/portals/meson.build

@@ -15,6 +15,7 @@ test_portals = [
   'org.freedesktop.impl.portal.RemoteDesktop',
   'org.freedesktop.impl.portal.Screenshot',
   'org.freedesktop.impl.portal.Settings',
+  'org.freedesktop.impl.portal.Usb',
   'org.freedesktop.impl.portal.Wallpaper',
 ]
 

tests/templates/meson.build

@@ -6,6 +6,7 @@ template_files = [
   'globalshortcuts.py',
   'inputcapture.py',
   'remotedesktop.py',
+  'usb.py',
 ]
 foreach template_file : template_files
   configure_file(

tests/templates/usb.py

@@ -0,0 +1,129 @@
+# 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
+import dbus.service
+from dbusmock import MOCK_IFACE
+
+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.Usb"
+VERSION = 1
+
+
+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.filters = parameters.get("filters", {})
+    mock.AddProperties(
+        MAIN_IFACE,
+        dbus.Dictionary(
+            {
+                "version": dbus.UInt32(parameters.get("version", VERSION)),
+            }
+        ),
+    )
+
+
+@dbus.service.method(
+    MAIN_IFACE,
+    in_signature="ossa(sa{sv}a{sv})a{sv}",
+    out_signature="ua{sv}",
+    async_callbacks=("cb_success", "cb_error"),
+)
+def AcquireDevices(
+    self,
+    handle,
+    parent_window,
+    app_id,
+    devices,
+    options,
+    cb_success,
+    cb_error,
+):
+    try:
+        logger.debug(
+            f"AcquireDevices({handle}, {parent_window}, {app_id}, {devices}, {options})"
+        )
+
+        # no options supported
+        assert not options
+        devices_out = []
+
+        for device in devices:
+            (id, info, access_options) = device
+            props = info["properties"]
+
+            allows_writable = self.filters.get("writable", True)
+            needs_writable = access_options.get("writable", False)
+            if needs_writable and not allows_writable:
+                logger.debug(f"Skipping device {id} because it requires writable")
+                continue
+
+            needs_vendor = self.filters.get("vendor", None)
+            needs_vendor = int(needs_vendor, 16) if needs_vendor else None
+
+            vendor = props.get("ID_VENDOR_ID", None)
+            vendor = int(vendor, 16) if vendor else None
+
+            if needs_vendor is not None and needs_vendor != vendor:
+                logger.debug(
+                    f"Skipping device {id} because it does not belong to vendor {needs_vendor:02x}"
+                )
+                continue
+
+            needs_model = self.filters.get("model", None)
+            needs_model = int(needs_model, 16) if needs_model else None
+
+            model = props.get("ID_MODEL_ID", None)
+            model = int(model, 16) if model else None
+
+            if needs_model is not None and needs_model != model:
+                logger.debug(
+                    f"Skipping device {id} because it is not a model {needs_model:02x}"
+                )
+                continue
+
+            devices_out.append(
+                dbus.Struct([id, access_options], signature="sa{sv}", variant_level=1)
+            )
+
+        response = Response(
+            self.response,
+            {"devices": dbus.Array(devices_out, signature="(sa{sv})", variant_level=1)},
+        )
+        request = ImplRequest(self, BUS_NAME, handle)
+        request.export()
+
+        def reply():
+            logger.debug(f"AcquireDevices with response {response}")
+            cb_success(response.response, response.results)
+
+        logger.debug(f"scheduling delay of {self.delay}")
+        GLib.timeout_add(self.delay, reply)
+
+    except Exception as e:
+        logger.critical(e)
+        cb_error(e)
+
+
+@dbus.service.method(
+    MOCK_IFACE,
+    in_signature="a{sv}",
+    out_signature="",
+)
+def SetSelectionFilters(self, filters):
+    logger.debug(f"SetSelectionFilters({filters})")
+
+    self.filters = filters