tests: add tests for RemoteDesktop.ConnectToEIS

Would be nice to use @pytest.mark.parametrize here but dbusmock forces
us to use python unittest, so duplication it is.

tests/templates/remotedesktop.py

@@ -5,6 +5,7 @@
 from tests.templates import Response, init_template_logger, ImplRequest, ImplSession
 import dbus
 import dbus.service
+import socket
 
 from gi.repository import GLib
 
@@ -26,6 +27,7 @@ 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.fail_connect_to_eis: bool = parameters.get("fail-connect-to-eis", False)
     mock.AddProperties(
         MAIN_IFACE,
         dbus.Dictionary(
@@ -138,3 +140,27 @@ def reply():
     except Exception as e:
         logger.critical(e)
         cb_error(e)
+
+
+@dbus.service.method(
+    MAIN_IFACE,
+    in_signature="osa{sv}",
+    out_signature="h",
+)
+def ConnectToEIS(self, session_handle, app_id, options):
+    try:
+        logger.debug(f"ConnectToEIS({session_handle}, {app_id}, {options})")
+
+        assert session_handle in self.sessions
+
+        if self.fail_connect_to_eis:
+            raise dbus.exceptions.DBusException(f"Purposely failing ConnectToEIS")
+
+        sockets = socket.socketpair()
+        self.eis_socket = sockets[0]
+        assert self.eis_socket.send(b"HELLO") == 5
+
+        return dbus.types.UnixFd(sockets[1])
+    except Exception as e:
+        logger.critical(e)
+        raise e

tests/test_remotedesktop.py

@@ -8,6 +8,8 @@
 
 import dbus
 import time
+import socket
+import pytest
 
 
 class TestRemoteDesktop(PortalTest):
@@ -78,3 +80,92 @@ def test_remote_desktop_create_session_signal_closed(self):
         mainloop.run()
 
         assert session.closed
+
+    def test_remote_desktop_connect_to_eis(self):
+        self.start_impl_portal()
+        self.start_xdp()
+
+        rd_intf = self.get_dbus_interface()
+        request = Request(self.dbus_con, rd_intf)
+        options = {
+            "session_handle_token": "session_token0",
+        }
+        response = request.call(
+            "CreateSession",
+            options=options,
+        )
+
+        assert response.response == 0
+
+        session = Session.from_response(self.dbus_con, response)
+        request = Request(self.dbus_con, rd_intf)
+        options = {
+            "types": dbus.UInt32(0x3),
+        }
+        response = request.call(
+            "SelectDevices",
+            session_handle=session.handle,
+            options=options,
+        )
+        assert response.response == 0
+
+        session = Session.from_response(self.dbus_con, response)
+        request = Request(self.dbus_con, rd_intf)
+        options = {}
+        response = request.call(
+            "Start",
+            session_handle=session.handle,
+            parent_window="",
+            options=options,
+        )
+        assert response.response == 0
+
+        fd = rd_intf.ConnectToEIS(session.handle, dbus.Dictionary({}, signature="sv"))
+        eis_socket = socket.fromfd(fd.take(), socket.AF_UNIX, socket.SOCK_STREAM)
+        assert eis_socket.recv(10) == b"HELLO"
+
+    def test_remote_desktop_connect_to_eis_fail(self):
+        params = {"fail-connect-to-eis": True}
+        self.start_impl_portal(params=params)
+        self.start_xdp()
+
+        rd_intf = self.get_dbus_interface()
+        request = Request(self.dbus_con, rd_intf)
+        options = {
+            "session_handle_token": "session_token0",
+        }
+        response = request.call(
+            "CreateSession",
+            options=options,
+        )
+
+        assert response.response == 0
+
+        session = Session.from_response(self.dbus_con, response)
+        request = Request(self.dbus_con, rd_intf)
+        options = {
+            "types": dbus.UInt32(0x3),
+        }
+        response = request.call(
+            "SelectDevices",
+            session_handle=session.handle,
+            options=options,
+        )
+        assert response.response == 0
+
+        session = Session.from_response(self.dbus_con, response)
+        request = Request(self.dbus_con, rd_intf)
+        options = {}
+        response = request.call(
+            "Start",
+            session_handle=session.handle,
+            parent_window="",
+            options=options,
+        )
+        assert response.response == 0
+
+        with self.assertRaises(dbus.exceptions.DBusException) as cm:
+            fd = rd_intf.ConnectToEIS(
+                session.handle, dbus.Dictionary({}, signature="sv")
+            )
+        assert "Purposely failing ConnectToEIS" in cm.exception.get_dbus_message()