From 2f9d453d796b62a44285c2d4087dde9753e1e3eb Mon Sep 17 00:00:00 2001 From: Max Booth Date: Tue, 7 Feb 2023 18:34:25 +0000 Subject: [PATCH] remote-desktop: Add clipboard tests --- tests/__init__.py | 11 +++ tests/portals/meson.build | 1 + tests/templates/clipboard.py | 149 +++++++++++++++++++++++++++++++ tests/templates/remotedesktop.py | 49 +++++++++- tests/test_clipboard.py | 98 ++++++++++++++++++++ tests/test_remotedesktop.py | 2 +- 6 files changed, 308 insertions(+), 2 deletions(-) create mode 100644 tests/templates/clipboard.py create mode 100644 tests/test_clipboard.py diff --git a/tests/__init__.py b/tests/__init__.py index 7377fc2cc..a7a4dfe5d 100644 --- a/tests/__init__.py +++ b/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 diff --git a/tests/portals/meson.build b/tests/portals/meson.build index dc7903aed..7bed78a1b 100644 --- a/tests/portals/meson.build +++ b/tests/portals/meson.build @@ -3,6 +3,7 @@ test_portals = [ 'org.freedesktop.impl.portal.Account', 'org.freedesktop.impl.portal.AppChooser', 'org.freedesktop.impl.portal.Background', + 'org.freedesktop.impl.portal.Clipboard', 'org.freedesktop.impl.portal.Email', 'org.freedesktop.impl.portal.FileChooser', 'org.freedesktop.impl.portal.Inhibit', diff --git a/tests/templates/clipboard.py b/tests/templates/clipboard.py new file mode 100644 index 000000000..8b55fc8d7 --- /dev/null +++ b/tests/templates/clipboard.py @@ -0,0 +1,149 @@ +# 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="ou", + out_signature="h", + async_callbacks=("cb_success", "cb_error"), +) +def SelectionWrite(self, session_handle, serial, cb_success, cb_error): + try: + logger.debug(f"SelectionWrite({session_handle}, {serial})") + + 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) + + +@dbus.service.method( + MAIN_IFACE, + in_signature="oub", + out_signature="", + async_callbacks=("cb_success", "cb_error"), +) +def SelectionWriteDone(self, session_handle, serial, success, cb_success, cb_error): + try: + logger.debug(f"SelectionWriteDone({session_handle}, {serial}, {success})") + + 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) diff --git a/tests/templates/remotedesktop.py b/tests/templates/remotedesktop.py index 2f5803aeb..8f59bf64b 100644 --- a/tests/templates/remotedesktop.py +++ b/tests/templates/remotedesktop.py @@ -13,7 +13,7 @@ MAIN_OBJ = "/org/freedesktop/portal/desktop" SYSTEM_BUS = False MAIN_IFACE = "org.freedesktop.impl.portal.RemoteDesktop" -VERSION = 1 +VERSION = 2 logger = init_template_logger(__name__) @@ -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) diff --git a/tests/test_clipboard.py b/tests/test_clipboard.py new file mode 100644 index 000000000..73047afee --- /dev/null +++ b/tests/test_clipboard.py @@ -0,0 +1,98 @@ +# 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() + + fd_object: dbus.types.UnixFd = clipboard_interface.SelectionWrite( + session.handle, 1234 + ) + assert fd_object + + fd = fd_object.take() + assert fd + + bytes_written = os.write(fd, b"Clipboard") + assert bytes_written > 0 + + clipboard_interface.SelectionWriteDone(session.handle, 1234, True) + + 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) diff --git a/tests/test_remotedesktop.py b/tests/test_remotedesktop.py index f25d84996..0e125473d 100644 --- a/tests/test_remotedesktop.py +++ b/tests/test_remotedesktop.py @@ -12,7 +12,7 @@ class TestRemoteDesktop(PortalTest): def test_version(self): - self.check_version(1) + self.check_version(2) def test_remote_desktop_create_close_session(self): self.start_impl_portal()