Skip to content

Commit

Permalink
tests/usb: Test the USB portal via pytest, dbusmock and umockdev
Browse files Browse the repository at this point in the history
  • Loading branch information
swick committed Oct 17, 2024
1 parent b20808f commit ed4d272
Show file tree
Hide file tree
Showing 5 changed files with 505 additions and 0 deletions.
1 change: 1 addition & 0 deletions tests/meson.build
Original file line number Diff line number Diff line change
Expand Up @@ -307,6 +307,7 @@ if enable_pytest
'test_location.py',
'test_remotedesktop.py',
'test_trash.py',
'test_usb.py',
]
foreach pytest_file : pytest_files
configure_file(
Expand Down
1 change: 1 addition & 0 deletions tests/portals/meson.build
Original file line number Diff line number Diff line change
Expand Up @@ -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',
]

Expand Down
1 change: 1 addition & 0 deletions tests/templates/meson.build
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ template_files = [
'globalshortcuts.py',
'inputcapture.py',
'remotedesktop.py',
'usb.py',
]
foreach template_file : template_files
configure_file(
Expand Down
135 changes: 135 additions & 0 deletions tests/templates/usb.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,135 @@
# SPDX-License-Identifier: LGPL-2.1-or-later
#
# This file is formatted with Python Black

from tests.templates import Response, init_template_logger, ImplRequest, ImplSession
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 != 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 != 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
Loading

0 comments on commit ed4d272

Please sign in to comment.