global shortcuts activation token test

portal-test.py

#!/bin/env python3

import dbus
import dbus.mainloop.glib
from dbus.mainloop.glib import DBusGMainLoop
from typing import Any, Dict, Optional, NamedTuple, Callable, List
from gi.repository import GLib
import logging
from itertools import count


_counter = count()

ASV = Dict[str, Any]

def init_logger(name: str) -> logging.Logger:
    """
    Common logging setup for tests. Use as:

        >>> import tests.xdp_utils as xdp
        >>> logger = xdp.init_logger(__name__)
        >>> logger.debug("foo")

    """
    logging.basicConfig(
        format="%(levelname).1s|%(name)s: %(message)s", level=logging.DEBUG
    )
    logger = logging.getLogger(f"xdp.{name}")
    logger.setLevel(logging.DEBUG)
    return logger


logger = init_logger("utils")


class Response(NamedTuple):
    """
    Response as returned by a completed :class:`Request`
    """

    response: int
    results: ASV


class ResponseTimeout(Exception):
    """
    Exception raised by :meth:`Request.call` if the Request did not receive a
    Response in time.
    """

    pass


class Closable:
    """
    Parent class for both Session and Request. Both of these have a Close()
    method.
    """

    def __init__(self, bus: dbus.Bus, objpath: str):
        self.objpath = objpath
        # GLib makes assertions in callbacks impossible, so we wrap all
        # callbacks into a try: except and store the error on the request to
        # be raised later when we're back in the main context
        self.error: Optional[Exception] = None

        self._mainloop: Optional[GLib.MainLoop] = None
        self._impl_closed = False
        self._bus = bus

        self._closable = type(self).__name__
        assert self._closable in ("Request", "Session")
        proxy = bus.get_object("org.freedesktop.portal.Desktop", objpath)
        self._closable_interface = dbus.Interface(
            proxy, f"org.freedesktop.portal.{self._closable}"
        )

    @property
    def bus(self) -> dbus.Bus:
        return self._bus

    @property
    def closed(self) -> bool:
        """
        True if the impl.portal was closed
        """
        return self._impl_closed

    def close(self) -> None:
        signal_match = None

        def cb_impl_closed_by_portal(handle) -> None:
            if handle == self.objpath:
                logger.debug(f"Impl{self._closable} {self.objpath} was closed")
                signal_match.remove()  # type: ignore
                self._impl_closed = True
                if self.closed and self._mainloop:
                    self._mainloop.quit()

        # See :class:`ImplRequest`, this signal is a side-channel for the
        # impl.portal template to notify us when the impl.Request was really
        # closed by the portal.
        signal_match = self._bus.add_signal_receiver(
            cb_impl_closed_by_portal,
            f"{self._closable}Closed",
            dbus_interface="org.freedesktop.impl.portal.Mock",
        )

        logger.debug(f"Closing {self._closable} {self.objpath}")
        self._closable_interface.Close()

    def schedule_close(self, timeout_ms=300):
        """
        Schedule an automatic Close() on the given timeout in milliseconds.
        """
        GLib.timeout_add(timeout_ms)


class Request(Closable):
    """
    Helper class for executing methods that use Requests. This calls takes
    care of subscribing to the signals and invokes the method on the
    interface with the expected behaviors. A typical invocation is:

            >>> response = Request(connection, interface).call("Foo", bar="bar")
            >>> assert response.response == 0

    Requests can only be used once, to call a second method you must
    instantiate a new Request object.
    """

    def __init__(self, bus: dbus.Bus, interface: dbus.Interface):
        def sanitize(name):
            return name.lstrip(":").replace(".", "_")

        sender_token = sanitize(bus.get_unique_name())
        self._handle_token = f"request{next(_counter)}"
        self.handle = f"/org/freedesktop/portal/desktop/request/{sender_token}/{self._handle_token}"
        self.timeout = -1
        # The Closable
        super().__init__(bus, self.handle)

        self.interface = interface
        self.response: Optional[Response] = None
        self.used = False
        # GLib makes assertions in callbacks impossible, so we wrap all
        # callbacks into a try: except and store the error on the request to
        # be raised later when we're back in the main context
        self.error: Optional[Exception] = None

        proxy = bus.get_object("org.freedesktop.portal.Desktop", self.handle)

        def cb_response(response: int, results: ASV) -> None:
            try:
                logger.debug(f"Response received on {self.handle}")
                assert self.response is None
                self.response = Response(response, results)
                if self._mainloop:
                    self._mainloop.quit()
            except Exception as e:
                self.error = e

        self.request_interface = dbus.Interface(proxy, "org.freedesktop.portal.Request")
        self.request_interface.connect_to_signal("Response", cb_response)

    @property
    def handle_token(self) -> dbus.String:
        """
        Returns the dbus-ready handle_token, ready to be put into the options
        """
        return dbus.String(self._handle_token, variant_level=1)

    def call(self, methodname: str, **kwargs) -> Optional[Response]:
        """
        Semi-synchronously call method ``methodname`` on the interface given
        in the Request's constructor. The kwargs must be specified in the
        order the DBus method takes them but the handle_token is automatically
        filled in.

            >>> response = Request(connection, interface).call("Foo", bar="bar")
            >>> if response.response != 0:
            ...     print("some error occured")

        The DBus call itself is asynchronous (required for signals to work)
        but this method does not return until the Response is received, the
        Request is closed or an error occurs. If the Request is closed, the
        Response is None.

        If the "reply_handler" and "error_handler" keywords are present, those
        callbacks are called just like they would be as dbus.service.ProxyObject.
        """
        assert not self.used
        self.used = True

        # Make sure options exists and has the handle_token set
        try:
            options = kwargs["options"]
        except KeyError:
            options = dbus.Dictionary({}, signature="sv")

        if "handle_token" not in options:
            options["handle_token"] = self.handle_token

        # Anything that takes longer than 5s needs to fail
        self._mainloop = GLib.MainLoop()
        if self.timeout >= 0:
            GLib.timeout_add(self.timeout, self._mainloop.quit)

        method = getattr(self.interface, methodname)
        assert method

        reply_handler = kwargs.pop("reply_handler", None)
        error_handler = kwargs.pop("error_handler", None)

        # Handle the normal method reply which returns is the Request object
        # path. We don't exit the mainloop here, we're waiting for either the
        # Response signal on the Request itself or the Close() handling
        def reply_cb(handle):
            try:
                logger.debug(f"Reply to {methodname} with {self.handle}")
                assert handle == self.handle

                if reply_handler:
                    reply_handler(handle)
            except Exception as e:
                self.error = e

        # Handle any exceptions during the actual method call (not the Request
        # handling itself). Can exit the mainloop if that happens
        def error_cb(error):
            try:
                logger.debug(f"Error after {methodname} with {error}")
                if error_handler:
                    error_handler(error)
                self.error = error
            except Exception as e:
                self.error = e
            finally:
                if self._mainloop:
                    self._mainloop.quit()

        # Method is invoked async, otherwise we can't mix and match signals
        # and other calls. It's still sync as seen by the caller in that we
        # have a mainloop that waits for us to finish though.
        method(
            *list(kwargs.values()),
            reply_handler=reply_cb,
            error_handler=error_cb,
        )

        self._mainloop.run()

        if self.error:
            raise self.error
        elif not self.closed and self.response is None:
            raise ResponseTimeout(f"Timed out waiting for response from {methodname}")

        return self.response


class Session(Closable):
    """
    Helper class for a Session created by a portal. This class takes care of
    subscribing to the `Closed` signals. A typical invocation is:

        >>> response = Request(connection, interface).call("CreateSession")
        >>> session = Session.from_response(response)
        # Now run the main loop and do other stuff
        # Check if the session was closed
        >>> if session.closed:
        ...    pass
        # or close the session explicitly
        >>> session.close()  # to close the session or
    """

    def __init__(self, bus: dbus.Bus, handle: str):
        assert handle
        super().__init__(bus, handle)

        self.handle = handle
        self.details = None
        # GLib makes assertions in callbacks impossible, so we wrap all
        # callbacks into a try: except and store the error on the request to
        # be raised later when we're back in the main context
        self.error = None
        self._closed_sig_received = False

        def cb_closed(details: ASV) -> None:
            try:
                logger.debug(f"Session.Closed received on {self.handle}")
                assert not self._closed_sig_received
                self._closed_sig_received = True
                self.details = details
                if self._mainloop:
                    self._mainloop.quit()
            except Exception as e:
                self.error = e

        proxy = bus.get_object("org.freedesktop.portal.Desktop", handle)
        self.session_interface = dbus.Interface(proxy, "org.freedesktop.portal.Session")
        self.session_interface.connect_to_signal("Closed", cb_closed)

    @property
    def closed(self):
        """
        Returns True if the session was closed by the backend
        """
        return self._closed_sig_received or super().closed

    @classmethod
    def from_response(cls, bus: dbus.Bus, response: Response) -> "Session":
        return cls(bus, response.results["session_handle"])

     
# This sets up a global shortcuts session, registers a shortcut, listens for it to activate
# It then tries to activate org.gnome.TextEditor
# If the activation token made it all the way to org.gnome.TextEditor, it should become focused
# Otherwise, gnome-shell will show the "An app is ready" notification instead

loop = GLib.MainLoop()
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

bus = dbus.SessionBus()
portal = bus.get_object("org.freedesktop.portal.Desktop", "/org/freedesktop/portal/desktop")

registry_iface = dbus.Interface(portal, "org.freedesktop.host.portal.Registry")
registry_iface.Register("org.gnome.Tecla", {})

globalshortcuts_iface = dbus.Interface(portal, "org.freedesktop.portal.GlobalShortcuts")
request = Request(bus, globalshortcuts_iface)
response = request.call("CreateSession", options={
    "session_handle_token": "test",
})
assert response.response == 0
session = Session.from_response(bus, response)

request = Request(bus, globalshortcuts_iface)
response = request.call("BindShortcuts",
    session_handle=session.handle,
    shortcuts=[
        ("s2", {"description": "test shortcut", "preferred_trigger": "<Ctrl>h"}),
    ],
    parent_window="",
    options={},
)
assert response.response == 0


def activated_handler(session_handle, shortcut_id, timestamp, options):
    print("got activation")
    print(options)

    platform_data = {}
    if "activation-token" in options:
        platform_data["activation-token"] = options["activation-token"]

    text_editor = bus.get_object("org.gnome.TextEditor", "/org/gnome/TextEditor")
    application_iface = dbus.Interface(text_editor, "org.freedesktop.Application")
    application_iface.Activate(platform_data)

    loop.quit()

globalshortcuts_iface.connect_to_signal("Activated", activated_handler)
loop.run()