Writing tests for Clipboard portal

tests/__init__.py

@@ -367,6 +367,17 @@ def start_impl_portal(self, params=None, portal=None):
 
         self.start_dbus_monitor()
 
+    def add_template(self, portal, params: Dict[str, Any] = {}):
+        """
+        Add an additional template to the portal object
+        """
+
+        self.obj_portal.AddTemplate(
+            f"tests/templates/{portal.lower()}.py",
+            dbus.Dictionary(params, signature="sv"),
+            dbus_interface=dbusmock.MOCK_IFACE,
+        )
+
     def start_xdp(self):
         """
         Start the xdg-desktop-portal process

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;org.freedesktop.impl.portal.RemoteDesktop;
+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.RemoteDesktop;org.freedesktop.impl.portal.Clipboard;
 UseIn=test

tests/templates/clipboard.py

@@ -0,0 +1,151 @@
+# SPDX-License-Identifier: LGPL-2.1-or-later
+#
+# This file is formatted with Python Black
+
+from tests.templates import init_template_logger
+import dbus.service
+import dbus
+import tempfile
+
+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.Clipboard"
+VERSION = 1
+
+logger = init_template_logger(__name__)
+
+
+def load(mock, parameters=None):
+    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)
+
+    mock.AddProperties(
+        MAIN_IFACE,
+        dbus.Dictionary(
+            {
+                "version": dbus.UInt32(parameters.get("version", VERSION)),
+            }
+        ),
+    )
+
+
+@dbus.service.method(
+    MAIN_IFACE,
+    in_signature="oa{sv}",
+    out_signature="",
+    async_callbacks=("cb_success", "cb_error"),
+)
+def RequestClipboard(self, session_handle, options, cb_success, cb_error):
+    try:
+        logger.debug(f"RequestClipboard({session_handle}, {options})")
+
+        if self.expect_close:
+            cb_success()
+        else:
+            logger.debug(f"scheduling delay of {self.delay}")
+            GLib.timeout_add(self.delay, cb_success)
+    except Exception as e:
+        logger.critical(e)
+        cb_error(e)
+
+
+@dbus.service.method(
+    MAIN_IFACE,
+    in_signature="oa{sv}",
+    out_signature="",
+    async_callbacks=("cb_success", "cb_error"),
+)
+def SetSelection(self, session_handle, options, cb_success, cb_error):
+    try:
+        logger.debug(f"SetSelection({session_handle}, {options})")
+
+        if self.expect_close:
+            cb_success()
+        else:
+            logger.debug(f"scheduling delay of {self.delay}")
+            GLib.timeout_add(self.delay, cb_success)
+    except Exception as e:
+        logger.critical(e)
+        cb_error(e)
+
+
+@dbus.service.method(
+    MAIN_IFACE,
+    in_signature="oau",
+    out_signature="a{uh}",
+    async_callbacks=("cb_success", "cb_error"),
+)
+def SelectionWrite(self, session_handle, serials, cb_success, cb_error):
+    try:
+        logger.debug(f"SelectionWrite({session_handle}, {serials})")
+
+        request_fds = {}
+        for serial in serials:
+            temp_file = tempfile.TemporaryFile()
+            request_fds[serial] = dbus.types.UnixFd(temp_file.fileno())
+
+        if self.expect_close:
+            cb_success(request_fds)
+        else:
+
+            def reply():
+                cb_success(request_fds)
+
+            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(
+    MAIN_IFACE,
+    in_signature="oa{ub}",
+    out_signature="",
+    async_callbacks=("cb_success", "cb_error"),
+)
+def SelectionWriteDone(self, session_handle, request_successes, cb_success, cb_error):
+    try:
+        logger.debug(f"SelectionWriteDone({session_handle}, {request_successes})")
+
+        if self.expect_close:
+            cb_success()
+        else:
+            logger.debug(f"scheduling delay of {self.delay}")
+            GLib.timeout_add(self.delay, cb_success)
+    except Exception as e:
+        logger.critical(e)
+        cb_error(e)
+
+
+@dbus.service.method(
+    MAIN_IFACE,
+    in_signature="os",
+    out_signature="h",
+    async_callbacks=("cb_success", "cb_error"),
+)
+def SelectionRead(self, session_handle, mime_type, cb_success, cb_error):
+    try:
+        logger.debug(f"SelectionRead({session_handle}, {mime_type})")
+
+        temp_file = tempfile.TemporaryFile()
+        fd = dbus.types.UnixFd(temp_file.fileno())
+
+        if self.expect_close:
+            cb_success(fd)
+        else:
+
+            def reply():
+                cb_success(fd)
+
+            logger.debug(f"scheduling delay of {self.delay}")
+            GLib.timeout_add(self.delay, reply)
+    except Exception as e:
+        logger.critical(e)
+        cb_error(e)

tests/templates/remotedesktop.py

@@ -26,6 +26,9 @@ def load(mock, parameters):
     mock.response: int = parameters.get("response", 0)
     mock.expect_close: bool = parameters.get("expect-close", False)
     mock.force_close: int = parameters.get("force-close", 0)
+    mock.force_clipoboard_enabled: bool = parameters.get(
+        "force-clipboard-enabled", False
+    )
     mock.AddProperties(
         MAIN_IFACE,
         dbus.Dictionary(
@@ -80,3 +83,47 @@ def force_close():
     except Exception as e:
         logger.critical(e)
         cb_error(e)
+
+
+@dbus.service.method(
+    MAIN_IFACE,
+    in_signature="oossa{sv}",
+    out_signature="ua{sv}",
+    async_callbacks=("cb_success", "cb_error"),
+)
+def Start(
+    self, handle, session_handle, app_id, parent_window, options, cb_success, cb_error
+):
+    try:
+        logger.debug(
+            f"Start({handle}, {session_handle}, {parent_window}, {app_id}, {options})"
+        )
+
+        response = Response(self.response, {})
+
+        if self.force_clipoboard_enabled:
+            response.results["clipboard_enabled"] = True
+
+        request = ImplRequest(self, BUS_NAME, handle)
+
+        if self.expect_close:
+
+            def closed_callback():
+                response = Response(2, {})
+                logger.debug(f"Start Close() response {response}")
+                cb_success(response.response, response.results)
+
+            request.export(closed_callback)
+        else:
+            request.export()
+
+            def reply():
+                logger.debug(f"Start 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)

tests/test_clipboard.py

@@ -0,0 +1,101 @@
+# SPDX-License-Identifier: LGPL-2.1-or-later
+#
+# This file is formatted with Python Black
+
+from logging import debug
+from tests import Request, PortalTest, Session
+from gi.repository import GLib
+import dbus
+import os
+
+
+class TestClipboard(PortalTest):
+    def test_version(self):
+        self.check_version(1)
+
+    def start_session(self, params={}):
+        self.start_impl_portal(params=params)
+        self.add_template(portal="RemoteDesktop", params=params)
+        self.start_xdp()
+
+        remote_desktop_interface = self.get_dbus_interface("RemoteDesktop")
+        clipboard_interface = self.get_dbus_interface()
+
+        create_session_request = Request(self.dbus_con, remote_desktop_interface)
+        create_session_response = create_session_request.call(
+            "CreateSession", options={"session_handle_token": "1234"}
+        )
+        assert create_session_response.response == 0
+        assert str(create_session_response.results["session_handle"])
+
+        session = Session.from_response(self.dbus_con, create_session_response)
+
+        clipboard_interface.RequestClipboard(session.handle, {})
+
+        start_session_request = Request(self.dbus_con, remote_desktop_interface)
+        start_session_response = start_session_request.call(
+            "Start", session_handle=session.handle, parent_window="", options={}
+        )
+
+        assert start_session_response.response == 0
+
+        return (session, start_session_response.results.get("clipboard_enabled"))
+
+    def test_request_clipboard_and_start_session(self):
+        params = {"force-clipboard-enabled": True}
+        _, clipboard_enabled = self.start_session(params)
+
+        assert clipboard_enabled
+
+    def test_clipboard_checks_clipboard_enabled(self):
+        session, clipboard_enabled = self.start_session()
+        clipboard_interface = self.get_dbus_interface()
+
+        self.assertFalse(clipboard_enabled)
+
+        with self.assertRaises(dbus.exceptions.DBusException):
+            clipboard_interface.SetSelection(session.handle, {})
+
+    def test_clipboard_set_selection(self):
+        params = {"force-clipboard-enabled": True}
+        session, _ = self.start_session(params)
+        clipboard_interface = self.get_dbus_interface()
+
+        clipboard_interface.SetSelection(session.handle, {})
+
+    def test_clipboard_selection_write(self):
+        params = {"force-clipboard-enabled": True}
+        session, _ = self.start_session(params)
+        clipboard_interface = self.get_dbus_interface()
+
+        request_fds: dict[int, dbus.types.UnixFd] = clipboard_interface.SelectionWrite(
+            session.handle, [1234, 5678]
+        )
+        assert request_fds
+
+        for serial, fd_object in request_fds.items():
+            fd = fd_object.take()
+            assert fd
+
+            bytes_written = os.write(fd, b"Clipboard")
+            assert bytes_written > 0
+
+        clipboard_interface.SelectionWriteDone(
+            session.handle, {1234: True, 5678: False}
+        )
+
+    def test_clipboard_selection_read(self):
+        params = {"force-clipboard-enabled": True}
+        session, _ = self.start_session(params)
+        clipboard_interface = self.get_dbus_interface()
+
+        fd_object: dbus.types.UnixFd = clipboard_interface.SelectionRead(
+            session.handle, "mimetype"
+        )
+        assert fd_object
+
+        fd = fd_object.take()
+        assert fd
+
+        clipboard = os.read(fd, 1000)
+        assert str(clipboard)