1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
|
# SPDX-License-Identifier: MIT
#
#
# This file is formatted with Python Black
from dbusmock import DBusMockObject
from typing import Dict, Any, NamedTuple, Optional
from itertools import count
from gi.repository import GLib # type: ignore
import dbus
import dbus.service
import logging
ASVType = Dict[str, Any]
logging.basicConfig(format="%(levelname).1s|%(name)s: %(message)s", level=logging.DEBUG)
logger = logging.getLogger("templates")
class MockParams:
"""
Helper class for storing template parameters. The Mock object passed into
``load()`` is shared between all templates. This makes it easy to have
per-template parameters by calling:
>>> params = MockParams.get(mock, MAIN_IFACE)
>>> params.version = 1
and later, inside a DBus method:
>>> params = MockParams.get(self, MAIN_IFACE)
>>> return params.version
"""
@classmethod
def get(cls, mock, interface_name):
params = getattr(mock, "params", {})
try:
return params[interface_name]
except KeyError:
c = cls()
params[interface_name] = c
mock.params = params
return c
class Response(NamedTuple):
response: int
results: ASVType
class Request:
_token_counter = count()
def __init__(
self, bus_name: dbus.service.BusName, sender: str, options: Optional[ASVType]
):
options = options or {}
sender_token = sender.removeprefix(":").replace(".", "_")
handle_token = options.get("handle_token", next(self._token_counter))
self.sender = sender
self.handle = (
f"/org/freedesktop/portal/desktop/request/{sender_token}/{handle_token}"
)
self.mock = DBusMockObject(
bus_name=bus_name,
path=self.handle,
interface="org.freedesktop.portal.Request",
props={},
)
self.mock.AddMethod("", "Close", "", "", "self.RemoveObject(self.path)")
logger.debug(f"Request created at {self.handle}")
def respond(self, response: Response, delay: int = 0):
def respond():
logger.debug(f"Request.Response on {self.handle}: {response}")
self.mock.EmitSignalDetailed(
"",
"Response",
"ua{sv}",
[dbus.UInt32(response.response), response.results],
details={"destination": self.sender},
)
if delay > 0:
GLib.timeout_add(delay, respond)
else:
respond()
class Session:
_token_counter = count()
def __init__(
self, bus_name: dbus.service.BusName, sender: str, options: Optional[ASVType]
):
options = options or {}
sender_token = sender.removeprefix(":").replace(".", "_")
handle_token = options.get("session_handle_token", next(self._token_counter))
self.sender = sender
self.handle = (
f"/org/freedesktop/portal/desktop/session/{sender_token}/{handle_token}"
)
self.mock = DBusMockObject(
bus_name=bus_name,
path=self.handle,
interface="org.freedesktop.portal.Session",
props={},
)
self.mock.AddMethod("", "Close", "", "", "self.RemoveObject(self.path)")
logger.debug(f"Session created at {self.handle}")
def close(self, details: ASVType, delay: int = 0):
def respond():
logger.debug(f"Session.Closed on {self.handle}: {details}")
self.mock.EmitSignalDetailed(
"", "Closed", "a{sv}", [details], destination=self.sender
)
if delay > 0:
GLib.timeout_add(delay, respond)
else:
respond()
|
