"""polkitd mock template
This creates the expected methods and properties of the main
org.freedesktop.PolicyKit1 object. By default, all actions are rejected. You
can call AllowUnknown() and SetAllowed() on the mock D-Bus interface to control
which actions are allowed.
"""
# SPDX-License-Identifier: LGPL-3.0-or-later
[docs]
__author__ = "Martin Pitt"
[docs]
__copyright__ = """
(c) 2013-2021 Canonical Ltd.
(c) 2017 - 2022 Martin Pitt <martin@piware.de>
"""
import time
import dbus
from dbusmock import MOCK_IFACE
[docs]
BUS_NAME = "org.freedesktop.PolicyKit1"
[docs]
MAIN_OBJ = "/org/freedesktop/PolicyKit1/Authority"
[docs]
MAIN_IFACE = "org.freedesktop.PolicyKit1.Authority"
[docs]
def load(mock, _parameters):
mock.AddProperties(
MAIN_IFACE,
dbus.Dictionary(
{
"BackendName": "local",
"BackendVersion": "0.8.15",
"BackendFeatures": dbus.UInt32(1),
},
signature="sv",
),
)
# default state
mock.allow_unknown = False
mock.allowed = []
mock.delay = 0
mock.simulate_hang = False
mock.hanging_actions = []
mock.hanging_calls = []
@dbus.service.method(
MAIN_IFACE, in_signature="(sa{sv})sa{ss}us", out_signature="(bba{ss})", async_callbacks=("ok_cb", "_err_cb")
)
[docs]
def CheckAuthorization(self, _subject, action_id, _details, _flags, _cancellation_id, ok_cb, _err_cb):
time.sleep(self.delay)
allowed = action_id in self.allowed or self.allow_unknown
ret = (allowed, False, {"test": "test"})
if self.simulate_hang or action_id in self.hanging_actions:
self.hanging_calls.append((ok_cb, ret))
else:
ok_cb(ret)
@dbus.service.method(MAIN_IFACE, in_signature="(sa{sv})ss")
[docs]
def RegisterAuthenticationAgent(_self, _subject, _locale, _object_path):
pass
@dbus.service.method(MAIN_IFACE, in_signature="(sa{sv})ssa{sv}")
[docs]
def RegisterAuthenticationAgentWithOptions(_self, _subject, _locale, _object_path, _options):
pass
@dbus.service.method(MOCK_IFACE, in_signature="b", out_signature="")
[docs]
def AllowUnknown(self, default):
"""Control whether unknown actions are allowed
This controls the return value of CheckAuthorization for actions which were
not explicitly allowed by SetAllowed().
"""
self.allow_unknown = default
@dbus.service.method(MOCK_IFACE, in_signature="as", out_signature="")
[docs]
def SetAllowed(self, actions):
"""Set allowed actions"""
self.allowed = actions
@dbus.service.method(MOCK_IFACE, in_signature="d", out_signature="")
[docs]
def SetDelay(self, delay):
"""Makes the CheckAuthorization() method to delay"""
self.delay = delay
@dbus.service.method(MOCK_IFACE, in_signature="b", out_signature="")
[docs]
def SimulateHang(self, hang):
"""Makes the CheckAuthorization() method to hang"""
self.simulate_hang = hang
@dbus.service.method(MOCK_IFACE, in_signature="as", out_signature="")
[docs]
def SimulateHangActions(self, actions):
"""Makes the CheckAuthorization() method to hang on such actions"""
self.hanging_actions = actions
@dbus.service.method(MOCK_IFACE, in_signature="", out_signature="")
[docs]
def ReleaseHangingCalls(self):
"""Calls all the hanging callbacks"""
for cb, ret in self.hanging_calls:
cb(ret)
self.hanging_calls = []
@dbus.service.method(MOCK_IFACE, in_signature="", out_signature="b")
[docs]
def HaveHangingCalls(self):
"""Check if we've hangling calls"""
return len(self.hanging_calls)