2
# (C) 2011 Sebastian Heinlein
3
# (C) 2012 Canonical Ltd.
5
# Sebastian Heinlein <sebi@glatzor.de>
6
# Martin Pitt <martin.pitt@ubuntu.com>
8
# This program is free software; you can redistribute it and/or modify
9
# it under the terms of the GNU General Public License as published by
10
# the Free Software Foundation; either version 2 of the License, or
11
# (at your option) any later version.
13
'''Simple mock polkit daemon for test suites.
15
This also provides some convenience API for launching the daemon and for
16
writing unittest test cases involving polkit operations.
28
from gi.repository import GLib, Gio
30
# ----------------------------------------------------------------------------
32
class TestPolicyKitDaemon(dbus.service.Object):
33
def __init__(self, allowed_actions, on_bus=None, replace=False):
34
'''Initialize test polkit daemon.
36
@allowed_actions is a list of PolicyKit action IDs which will be
37
allowed (active/inactive sessions or user IDs will not be considered);
38
all actions not in that list will be denied. If 'all' is an element of
39
@allowed_actions, all actions will be allowed.
41
When @on_bus string is given, the daemon will run on that D-BUS
42
address, otherwise on the system D-BUS.
44
If @replace is True, this will replace an already running polkit daemon
47
self.allowed_actions = allowed_actions
49
bus = dbus.bus.BusConnection(on_bus)
51
bus = dbus.SystemBus()
52
bus_name = dbus.service.BusName('org.freedesktop.PolicyKit1',
53
bus, do_not_queue=True,
54
replace_existing=replace,
55
allow_replacement=True)
56
bus.add_signal_receiver(self.on_disconnected, signal_name='Disconnected')
58
dbus.service.Object.__init__(self, bus_name,
59
'/org/freedesktop/PolicyKit1/Authority')
60
self.loop = GLib.MainLoop()
65
@dbus.service.method('org.freedesktop.PolicyKit1.Authority',
66
in_signature='(sa{sv})sa{ss}us',
67
out_signature='(bba{ss})')
68
def CheckAuthorization(self, subject, action_id, details, flags,
70
if 'all' in self.allowed_actions:
73
allowed = action_id in self.allowed_actions
75
details = {'test': 'test'}
76
return (allowed, challenged, details)
78
@dbus.service.method('org.freedesktop.PolicyKit1.Authority',
79
in_signature='', out_signature='')
81
GLib.idle_add(self.loop.quit)
83
def on_disconnected(self):
84
print('disconnected from D-BUS, terminating')
87
# ----------------------------------------------------------------------------
89
class PolkitTestCase(unittest.TestCase):
90
'''Convenient test cases involving polkit.
92
Call start_polkitd() with the list of allowed actions in your test cases.
93
The daemon will be automatically terminated when the test case exits.
96
def __init__(self, methodName='runTest'):
97
unittest.TestCase.__init__(self, methodName)
98
self.polkit_pid = None
100
def start_polkitd(self, allowed_actions, on_bus=None):
101
'''Start test polkitd.
103
This should be called in your test cases before the exercised code
104
makes any polkit query. The daemon will be stopped automatically when
105
the test case ends (regardless of whether its successful or failed). If
106
you want to test multiple different action sets in one test case, you
107
have to call stop_polkitd() before starting a new one.
109
@allowed_actions is a list of PolicyKit action IDs which will be
110
allowed (active/inactive sessions or user IDs will not be considered);
111
all actions not in that list will be denied. If 'all' is an element of
112
@allowed_actions, all actions will be allowed.
114
When @on_bus string is given, the daemon will run on that D-BUS
115
address, otherwise on the system D-BUS.
117
assert self.polkit_pid is None, \
118
'can only launch one polkitd at a time; write a separate test case or call stop_polkitd()'
119
self.polkit_pid = spawn(allowed_actions, on_bus)
120
self.addCleanup(self.stop_polkitd)
122
def stop_polkitd(self):
123
'''Stop test polkitd.
125
This happens automatically when a test case ends, but is required when
126
you want to test multiple different action sets in one test case.
128
assert self.polkit_pid is not None, 'polkitd is not running'
129
os.kill(self.polkit_pid, signal.SIGTERM)
130
os.waitpid(self.polkit_pid, 0)
131
self.polkit_pid = None
133
# ----------------------------------------------------------------------------
135
def _run(allowed_actions, bus_address, replace=False):
136
# Set up the DBus main loop
137
import dbus.mainloop.glib
138
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
140
polkitd = TestPolicyKitDaemon(allowed_actions, bus_address, replace)
143
def spawn(allowed_actions, on_bus=None):
144
'''Run a TestPolicyKitDaemon instance in a separate process.
146
@allowed_actions is a list of PolicyKit action IDs which will be
147
allowed (active/inactive sessions or user IDs will not be considered);
148
all actions not in that list will be denied. If 'all' is an element of
149
@allowed_actions, all actions will be allowed.
151
When @on_bus string is given, the daemon will run on that D-BUS address,
152
otherwise on the system D-BUS.
154
The daemon will terminate automatically when the @on_bus D-BUS goes down.
155
If that does not happen (e. g. you test on the actual system/session bus),
156
you need to kill it manually.
158
Returns the process ID of the spawned daemon.
163
_run(allowed_actions, on_bus)
166
# wait until the daemon is up on the bus
168
bus = dbus.bus.BusConnection(on_bus)
169
elif 'DBUS_SYSTEM_BUS_ADDRESS' in os.environ:
170
# dbus.SystemBus() does not recognize this env var, so we have to
171
# handle that manually
172
bus = dbus.bus.BusConnection(os.environ['DBUS_SYSTEM_BUS_ADDRESS'])
174
bus = dbus.SystemBus()
176
while timeout > 0 and not bus.name_has_owner('org.freedesktop.PolicyKit1'):
179
assert timeout > 0, 'test polkitd failed to start up'
184
parser = argparse.ArgumentParser(description='Simple mock polkit daemon for test suites')
185
parser.add_argument('-a', '--allowed-actions', metavar='ACTION[,ACTION,...]',
186
default='', help='Comma separated list of allowed action ids')
187
parser.add_argument('-b', '--bus-address',
188
help='D-BUS address to listen on (if not given, listen on system D-BUS)')
189
parser.add_argument('-r', '--replace', action='store_true',
190
help='Replace existing polkit daemon on the bus')
191
args = parser.parse_args()
193
_run(args.allowed_actions.split(','), args.bus_address, args.replace)
195
if __name__ == '__main__':
added
removed
src/tests/test_polkitd.py
