To get this branch, use:
bzr branch
http://bzr.recompile.se/loggerhead/mandos/trunk
« back to all changes in this revision
Viewing changes to mandos-ctl
- browse files at revision 1089
- compare with another revision
- download diff
- download tarball
- view history from revision 1089
- Committer: Teddy Hogeborn
- Date: 2019-03-23 16:00:26 UTC
- Revision ID: teddy@recompile.se-20190323160026-u82ba6lkh7jeq0zj
mandos-ctl: Eliminate unnecessary D-Bus call
* mandos-ctl (command.Base.run): Don't set self.mandos.
(command.Remove): Implement .run() instead of .run_on_one_client()
to avoid the command.Base.run() unnecessarily
connecting to a client object.
* mandos-ctl (command.Base.run): Don't set self.mandos.
(command.Remove): Implement .run() instead of .run_on_one_client()
to avoid the command.Base.run() unnecessarily
connecting to a client object.
- files modified:
- intro.xml
added
removed
mandos-ctl
45
45
import io
46
46
import tempfile
47
47
import contextlib
48
import abc
49
48
50
import dbus as dbus_python
49
import dbus
51
50
52
51
# Show warnings by default
53
52
if not sys.warnoptions:
67
66
68
67
locale.setlocale(locale.LC_ALL, "")
69
68
69
dbus_busname_domain = "se.recompile"
70
dbus_busname = dbus_busname_domain + ".Mandos"
71
server_dbus_path = "/"
72
server_dbus_interface = dbus_busname_domain + ".Mandos"
73
client_dbus_interface = dbus_busname_domain + ".Mandos.Client"
74
del dbus_busname_domain
70
75
version = "1.8.3"
71
76
72
77
78
try:
79
dbus.OBJECT_MANAGER_IFACE
80
except AttributeError:
81
dbus.OBJECT_MANAGER_IFACE = "org.freedesktop.DBus.ObjectManager"
82
83
73
84
def main():
74
85
parser = argparse.ArgumentParser()
75
86
add_command_line_options(parser)
82
93
if options.debug:
83
94
log.setLevel(logging.DEBUG)
84
95
85
bus = dbus_python_adapter.CachingBus(dbus_python)
86
87
try:
88
all_clients = bus.get_clients_and_properties()
89
except dbus.ConnectFailed as e:
90
log.critical("Could not connect to Mandos server: %s", e)
91
sys.exit(1)
92
except dbus.Error as e:
93
log.critical(
94
"Failed to access Mandos server through D-Bus:\n%s", e)
95
sys.exit(1)
96
bus = dbus.SystemBus()
97
98
mandos_dbus_object = get_mandos_dbus_object(bus)
99
100
mandos_serv = dbus.Interface(
101
mandos_dbus_object, dbus_interface=server_dbus_interface)
102
mandos_serv_object_manager = dbus.Interface(
103
mandos_dbus_object, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
104
105
managed_objects = get_managed_objects(mandos_serv_object_manager)
106
107
all_clients = {}
108
for path, ifs_and_props in managed_objects.items():
109
try:
110
all_clients[path] = ifs_and_props[client_dbus_interface]
111
except KeyError:
112
pass
96
113
97
114
# Compile dict of (clientpath: properties) to process
98
115
if not clientnames:
111
128
commands = commands_from_options(options)
112
129
113
130
for command in commands:
114
command.run(clients, bus)
131
command.run(clients, bus, mandos_serv)
115
132
116
133
117
134
def add_command_line_options(parser):
407
424
options.remove = True
408
425
409
426
410
411
class dbus(object):
412
413
class SystemBus(object):
414
415
object_manager_iface = "org.freedesktop.DBus.ObjectManager"
416
def get_managed_objects(self, busname, objectpath):
417
return self.call_method("GetManagedObjects", busname,
418
objectpath,
419
self.object_manager_iface)
420
421
properties_iface = "org.freedesktop.DBus.Properties"
422
def set_property(self, busname, objectpath, interface, key,
423
value):
424
self.call_method("Set", busname, objectpath,
425
self.properties_iface, interface, key,
426
value)
427
428
429
class MandosBus(SystemBus):
430
busname_domain = "se.recompile"
431
busname = busname_domain + ".Mandos"
432
server_path = "/"
433
server_interface = busname_domain + ".Mandos"
434
client_interface = busname_domain + ".Mandos.Client"
435
del busname_domain
436
437
def get_clients_and_properties(self):
438
managed_objects = self.get_managed_objects(
439
self.busname, self.server_path)
440
return {objpath: properties[self.client_interface]
441
for objpath, properties in managed_objects.items()
442
if self.client_interface in properties}
443
444
def set_client_property(self, objectpath, key, value):
445
return self.set_property(self.busname, objectpath,
446
self.client_interface, key,
447
value)
448
449
def call_client_method(self, objectpath, method, *args):
450
return self.call_method(method, self.busname, objectpath,
451
self.client_interface, *args)
452
453
def call_server_method(self, method, *args):
454
return self.call_method(method, self.busname,
455
self.server_path,
456
self.server_interface, *args)
457
458
class Error(Exception):
459
pass
460
461
class ConnectFailed(Error):
462
pass
463
464
465
class dbus_python_adapter(object):
466
467
class SystemBus(dbus.MandosBus):
468
"""Use dbus-python"""
469
470
def __init__(self, module=dbus_python):
471
self.dbus_python = module
472
self.bus = self.dbus_python.SystemBus()
473
474
@contextlib.contextmanager
475
def convert_exception(self, exception_class=dbus.Error):
476
try:
477
yield
478
except self.dbus_python.exceptions.DBusException as e:
479
# This does what "raise from" would do
480
exc = exception_class(*e.args)
481
exc.__cause__ = e
482
raise exc
483
484
def call_method(self, methodname, busname, objectpath,
485
interface, *args):
486
proxy_object = self.get_object(busname, objectpath)
487
log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,
488
interface, methodname,
489
", ".join(repr(a) for a in args))
490
method = getattr(proxy_object, methodname)
491
with self.convert_exception():
492
with dbus_python_adapter.SilenceLogger(
493
"dbus.proxies"):
494
value = method(*args, dbus_interface=interface)
495
return self.type_filter(value)
496
497
def get_object(self, busname, objectpath):
498
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
499
busname, objectpath)
500
with self.convert_exception(dbus.ConnectFailed):
501
return self.bus.get_object(busname, objectpath)
502
503
def type_filter(self, value):
504
"""Convert the most bothersome types to Python types"""
505
if isinstance(value, self.dbus_python.Boolean):
506
return bool(value)
507
if isinstance(value, self.dbus_python.ObjectPath):
508
return str(value)
509
# Also recurse into dictionaries
510
if isinstance(value, self.dbus_python.Dictionary):
511
return {self.type_filter(key):
512
self.type_filter(subval)
513
for key, subval in value.items()}
514
return value
515
516
517
class SilenceLogger(object):
518
"Simple context manager to silence a particular logger"
519
def __init__(self, loggername):
520
self.logger = logging.getLogger(loggername)
521
522
def __enter__(self):
523
self.logger.addFilter(self.nullfilter)
524
525
class NullFilter(logging.Filter):
526
def filter(self, record):
527
return False
528
529
nullfilter = NullFilter()
530
531
def __exit__(self, exc_type, exc_val, exc_tb):
532
self.logger.removeFilter(self.nullfilter)
533
534
535
class CachingBus(SystemBus):
536
"""A caching layer for dbus_python_adapter.SystemBus"""
537
def __init__(self, *args, **kwargs):
538
self.object_cache = {}
539
super(dbus_python_adapter.CachingBus,
540
self).__init__(*args, **kwargs)
541
def get_object(self, busname, objectpath):
542
try:
543
return self.object_cache[(busname, objectpath)]
544
except KeyError:
545
new_object = super(
546
dbus_python_adapter.CachingBus,
547
self).get_object(busname, objectpath)
548
self.object_cache[(busname, objectpath)] = new_object
549
return new_object
427
def get_mandos_dbus_object(bus):
428
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
429
dbus_busname, server_dbus_path)
430
with if_dbus_exception_log_with_exception_and_exit(
431
"Could not connect to Mandos server: %s"):
432
mandos_dbus_object = bus.get_object(dbus_busname,
433
server_dbus_path)
434
return mandos_dbus_object
435
436
437
@contextlib.contextmanager
438
def if_dbus_exception_log_with_exception_and_exit(*args, **kwargs):
439
try:
440
yield
441
except dbus.exceptions.DBusException as e:
442
log.critical(*(args + (e,)), **kwargs)
443
sys.exit(1)
444
445
446
def get_managed_objects(object_manager):
447
log.debug("D-Bus: %s:%s:%s.GetManagedObjects()", dbus_busname,
448
server_dbus_path, dbus.OBJECT_MANAGER_IFACE)
449
with if_dbus_exception_log_with_exception_and_exit(
450
"Failed to access Mandos server through D-Bus:\n%s"):
451
with SilenceLogger("dbus.proxies"):
452
managed_objects = object_manager.GetManagedObjects()
453
return managed_objects
454
455
456
class SilenceLogger(object):
457
"Simple context manager to silence a particular logger"
458
def __init__(self, loggername):
459
self.logger = logging.getLogger(loggername)
460
461
def __enter__(self):
462
self.logger.addFilter(self.nullfilter)
463
464
class NullFilter(logging.Filter):
465
def filter(self, record):
466
return False
467
468
nullfilter = NullFilter()
469
470
def __exit__(self, exc_type, exc_val, exc_tb):
471
self.logger.removeFilter(self.nullfilter)
550
472
551
473
552
474
def commands_from_options(options):
629
551
630
552
class Base(object):
631
553
"""Abstract base class for commands"""
632
def run(self, clients, bus=None):
554
def run(self, clients, bus=None, mandos=None):
633
555
"""Normal commands should implement run_on_one_client(),
634
556
but commands which want to operate on all clients at the same time can
635
557
override this run() method instead.
636
558
"""
637
self.bus = bus
638
for client, properties in clients.items():
559
for clientpath, properties in clients.items():
560
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
561
dbus_busname, str(clientpath))
562
client = bus.get_object(dbus_busname, clientpath)
639
563
self.run_on_one_client(client, properties)
640
564
641
565
642
566
class IsEnabled(Base):
643
def run(self, clients, bus=None):
644
properties = next(iter(clients.values()))
645
if properties["Enabled"]:
567
def run(self, clients, bus=None, mandos=None):
568
client, properties = next(iter(clients.items()))
569
if self.is_enabled(client, properties):
646
570
sys.exit(0)
647
571
sys.exit(1)
572
def is_enabled(self, client, properties):
573
return properties["Enabled"]
648
574
649
575
650
576
class Approve(Base):
651
577
def run_on_one_client(self, client, properties):
652
self.bus.call_client_method(client, "Approve", True)
578
log.debug("D-Bus: %s:%s:%s.Approve(True)", dbus_busname,
579
client.__dbus_object_path__,
580
client_dbus_interface)
581
client.Approve(dbus.Boolean(True),
582
dbus_interface=client_dbus_interface)
653
583
654
584
655
585
class Deny(Base):
656
586
def run_on_one_client(self, client, properties):
657
self.bus.call_client_method(client, "Approve", False)
587
log.debug("D-Bus: %s:%s:%s.Approve(False)", dbus_busname,
588
client.__dbus_object_path__,
589
client_dbus_interface)
590
client.Approve(dbus.Boolean(False),
591
dbus_interface=client_dbus_interface)
658
592
659
593
660
594
class Remove(Base):
661
def run(self, clients, bus):
662
for clientpath in frozenset(clients.keys()):
663
bus.call_server_method("RemoveClient", clientpath)
595
def run(self, clients, bus, mandos):
596
for clientpath in clients.keys():
597
log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)",
598
dbus_busname, server_dbus_path,
599
server_dbus_interface, clientpath)
600
mandos.RemoveClient(clientpath)
664
601
665
602
666
603
class Output(Base):
676
613
677
614
678
615
class DumpJSON(Output):
679
def run(self, clients, bus=None):
680
data = {properties["Name"]:
681
{key: properties[key]
616
def run(self, clients, bus=None, mandos=None):
617
data = {client["Name"]:
618
{key: self.dbus_boolean_to_bool(client[key])
682
619
for key in self.all_keywords}
683
for properties in clients.values()}
620
for client in clients.values()}
684
621
print(json.dumps(data, indent=4, separators=(',', ': ')))
685
622
623
@staticmethod
624
def dbus_boolean_to_bool(value):
625
if isinstance(value, dbus.Boolean):
626
value = bool(value)
627
return value
628
686
629
687
630
class PrintTable(Output):
688
631
def __init__(self, verbose=False):
689
632
self.verbose = verbose
690
633
691
def run(self, clients, bus=None):
634
def run(self, clients, bus=None, mandos=None):
692
635
default_keywords = ("Name", "Enabled", "Timeout",
693
636
"LastCheckedOK")
694
637
keywords = default_keywords
755
698
756
699
@classmethod
757
700
def valuetostring(cls, value, keyword):
758
if isinstance(value, bool):
701
if isinstance(value, dbus.Boolean):
759
702
return "Yes" if value else "No"
760
703
if keyword in ("Timeout", "Interval", "ApprovalDelay",
761
704
"ApprovalDuration", "ExtendedTimeout"):
784
727
class PropertySetter(Base):
785
728
"Abstract class for Actions for setting one client property"
786
729
787
def run_on_one_client(self, client, properties=None):
730
def run_on_one_client(self, client, properties):
788
731
"""Set the Client's D-Bus property"""
789
self.bus.set_client_property(client, self.propname,
790
self.value_to_set)
732
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", dbus_busname,
733
client.__dbus_object_path__,
734
dbus.PROPERTIES_IFACE, client_dbus_interface,
735
self.propname, self.value_to_set
736
if not isinstance(self.value_to_set,
737
dbus.Boolean)
738
else bool(self.value_to_set))
739
client.Set(client_dbus_interface, self.propname,
740
self.value_to_set,
741
dbus_interface=dbus.PROPERTIES_IFACE)
791
742
792
743
@property
793
744
def propname(self):
796
747
797
748
class Enable(PropertySetter):
798
749
propname = "Enabled"
799
value_to_set = True
750
value_to_set = dbus.Boolean(True)
800
751
801
752
802
753
class Disable(PropertySetter):
803
754
propname = "Enabled"
804
value_to_set = False
755
value_to_set = dbus.Boolean(False)
805
756
806
757
807
758
class BumpTimeout(PropertySetter):
811
762
812
763
class StartChecker(PropertySetter):
813
764
propname = "CheckerRunning"
814
value_to_set = True
765
value_to_set = dbus.Boolean(True)
815
766
816
767
817
768
class StopChecker(PropertySetter):
818
769
propname = "CheckerRunning"
819
value_to_set = False
770
value_to_set = dbus.Boolean(False)
820
771
821
772
822
773
class ApproveByDefault(PropertySetter):
823
774
propname = "ApprovedByDefault"
824
value_to_set = True
775
value_to_set = dbus.Boolean(True)
825
776
826
777
827
778
class DenyByDefault(PropertySetter):
828
779
propname = "ApprovedByDefault"
829
value_to_set = False
780
value_to_set = dbus.Boolean(False)
830
781
831
782
832
783
class PropertySetterValue(PropertySetter):
928
879
"output"))
929
880
930
881
931
class Unique(object):
932
"""Class for objects which exist only to be unique objects, since
933
unittest.mock.sentinel only exists in Python 3.3"""
934
935
936
882
class Test_string_to_delta(TestCaseWithAssertLogs):
937
883
# Just test basic RFC 3339 functionality here, the doc string for
938
884
# rfc3339_duration_to_delta() already has more comprehensive
939
# tests, which are run by doctest.
885
# tests, which is run by doctest.
940
886
941
887
def test_rfc3339_zero_seconds(self):
942
888
self.assertEqual(datetime.timedelta(),
1126
1072
self.check_option_syntax(options)
1127
1073
1128
1074
1129
class Test_dbus_exceptions(unittest.TestCase):
1130
1131
def test_dbus_ConnectFailed_is_Error(self):
1132
with self.assertRaises(dbus.Error):
1133
raise dbus.ConnectFailed()
1134
1135
1136
class Test_dbus_MandosBus(unittest.TestCase):
1137
1138
class MockMandosBus(dbus.MandosBus):
1139
def __init__(self):
1140
self._name = "se.recompile.Mandos"
1141
self._server_path = "/"
1142
self._server_interface = "se.recompile.Mandos"
1143
self._client_interface = "se.recompile.Mandos.Client"
1144
self.calls = []
1145
self.call_method_return = Unique()
1146
1147
def call_method(self, methodname, busname, objectpath,
1148
interface, *args):
1149
self.calls.append((methodname, busname, objectpath,
1150
interface, args))
1151
return self.call_method_return
1152
1153
def setUp(self):
1154
self.bus = self.MockMandosBus()
1155
1156
def test_set_client_property(self):
1157
self.bus.set_client_property("objectpath", "key", "value")
1158
expected_call = ("Set", self.bus._name, "objectpath",
1159
"org.freedesktop.DBus.Properties",
1160
(self.bus._client_interface, "key", "value"))
1161
self.assertIn(expected_call, self.bus.calls)
1162
1163
def test_call_client_method(self):
1164
ret = self.bus.call_client_method("objectpath", "methodname")
1165
self.assertIs(self.bus.call_method_return, ret)
1166
expected_call = ("methodname", self.bus._name, "objectpath",
1167
self.bus._client_interface, ())
1168
self.assertIn(expected_call, self.bus.calls)
1169
1170
def test_call_client_method_with_args(self):
1171
args = (Unique(), Unique())
1172
ret = self.bus.call_client_method("objectpath", "methodname",
1173
*args)
1174
self.assertIs(self.bus.call_method_return, ret)
1175
expected_call = ("methodname", self.bus._name, "objectpath",
1176
self.bus._client_interface,
1177
(args[0], args[1]))
1178
self.assertIn(expected_call, self.bus.calls)
1179
1180
def test_get_clients_and_properties(self):
1181
managed_objects = {
1182
"objectpath": {
1183
self.bus._client_interface: {
1184
"key": "value",
1185
"bool": True,
1186
},
1187
"irrelevant_interface": {
1188
"key": "othervalue",
1189
"bool": False,
1190
},
1191
},
1192
"other_objectpath": {
1193
"other_irrelevant_interface": {
1194
"key": "value 3",
1195
"bool": None,
1196
},
1197
},
1198
}
1199
expected_clients_and_properties = {
1200
"objectpath": {
1201
"key": "value",
1202
"bool": True,
1203
}
1204
}
1205
self.bus.call_method_return = managed_objects
1206
ret = self.bus.get_clients_and_properties()
1207
self.assertDictEqual(expected_clients_and_properties, ret)
1208
expected_call = ("GetManagedObjects", self.bus._name,
1209
self.bus._server_path,
1210
"org.freedesktop.DBus.ObjectManager", ())
1211
self.assertIn(expected_call, self.bus.calls)
1212
1213
def test_call_server_method(self):
1214
ret = self.bus.call_server_method("methodname")
1215
self.assertIs(self.bus.call_method_return, ret)
1216
expected_call = ("methodname", self.bus._name,
1217
self.bus._server_path,
1218
self.bus._server_interface, ())
1219
self.assertIn(expected_call, self.bus.calls)
1220
1221
def test_call_server_method_with_args(self):
1222
args = (Unique(), Unique())
1223
ret = self.bus.call_server_method("methodname", *args)
1224
self.assertIs(self.bus.call_method_return, ret)
1225
expected_call = ("methodname", self.bus._name,
1226
self.bus._server_path,
1227
self.bus._server_interface,
1228
(args[0], args[1]))
1229
self.assertIn(expected_call, self.bus.calls)
1230
1231
1232
class Test_dbus_python_adapter_SystemBus(TestCaseWithAssertLogs):
1233
1234
def MockDBusPython_func(self, func):
1235
class mock_dbus_python(object):
1236
"""mock dbus-python module"""
1237
class exceptions(object):
1238
"""Pseudo-namespace"""
1239
class DBusException(Exception):
1240
pass
1241
class SystemBus(object):
1242
@staticmethod
1243
def get_object(busname, objectpath):
1244
DBusObject = collections.namedtuple(
1245
"DBusObject", ("methodname",))
1246
def method(*args, **kwargs):
1247
self.assertEqual({"dbus_interface":
1248
"interface"},
1249
kwargs)
1250
return func(*args)
1251
return DBusObject(methodname=method)
1252
class Boolean(object):
1253
def __init__(self, value):
1254
self.value = bool(value)
1255
def __bool__(self):
1256
return self.value
1257
if sys.version_info.major == 2:
1258
__nonzero__ = __bool__
1259
class ObjectPath(str):
1260
pass
1261
class Dictionary(dict):
1262
pass
1263
return mock_dbus_python
1264
1265
def call_method(self, bus, methodname, busname, objectpath,
1266
interface, *args):
1267
with self.assertLogs(log, logging.DEBUG):
1268
return bus.call_method(methodname, busname, objectpath,
1269
interface, *args)
1270
1271
def test_call_method_returns(self):
1272
expected_method_return = Unique()
1273
method_args = (Unique(), Unique())
1274
def func(*args):
1275
self.assertEqual(len(method_args), len(args))
1276
for marg, arg in zip(method_args, args):
1277
self.assertIs(marg, arg)
1278
return expected_method_return
1279
mock_dbus_python = self.MockDBusPython_func(func)
1280
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1281
ret = self.call_method(bus, "methodname", "busname",
1282
"objectpath", "interface",
1283
*method_args)
1284
self.assertIs(ret, expected_method_return)
1285
1286
def test_call_method_filters_bool_true(self):
1287
def func():
1288
return method_return
1289
mock_dbus_python = self.MockDBusPython_func(func)
1290
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1291
method_return = mock_dbus_python.Boolean(True)
1292
ret = self.call_method(bus, "methodname", "busname",
1293
"objectpath", "interface")
1294
self.assertTrue(ret)
1295
self.assertNotIsInstance(ret, mock_dbus_python.Boolean)
1296
1297
def test_call_method_filters_bool_false(self):
1298
def func():
1299
return method_return
1300
mock_dbus_python = self.MockDBusPython_func(func)
1301
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1302
method_return = mock_dbus_python.Boolean(False)
1303
ret = self.call_method(bus, "methodname", "busname",
1304
"objectpath", "interface")
1305
self.assertFalse(ret)
1306
self.assertNotIsInstance(ret, mock_dbus_python.Boolean)
1307
1308
def test_call_method_filters_objectpath(self):
1309
def func():
1310
return method_return
1311
mock_dbus_python = self.MockDBusPython_func(func)
1312
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1313
method_return = mock_dbus_python.ObjectPath("objectpath")
1314
ret = self.call_method(bus, "methodname", "busname",
1315
"objectpath", "interface")
1316
self.assertEqual("objectpath", ret)
1317
self.assertIsNot("objectpath", ret)
1318
self.assertNotIsInstance(ret, mock_dbus_python.ObjectPath)
1319
1320
def test_call_method_filters_booleans_in_dict(self):
1321
def func():
1322
return method_return
1323
mock_dbus_python = self.MockDBusPython_func(func)
1324
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1325
method_return = mock_dbus_python.Dictionary(
1326
{mock_dbus_python.Boolean(True):
1327
mock_dbus_python.Boolean(False),
1328
mock_dbus_python.Boolean(False):
1329
mock_dbus_python.Boolean(True)})
1330
ret = self.call_method(bus, "methodname", "busname",
1331
"objectpath", "interface")
1332
expected_method_return = {True: False,
1333
False: True}
1334
self.assertEqual(expected_method_return, ret)
1335
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1336
1337
def test_call_method_filters_objectpaths_in_dict(self):
1338
def func():
1339
return method_return
1340
mock_dbus_python = self.MockDBusPython_func(func)
1341
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1342
method_return = mock_dbus_python.Dictionary(
1343
{mock_dbus_python.ObjectPath("objectpath_key_1"):
1344
mock_dbus_python.ObjectPath("objectpath_value_1"),
1345
mock_dbus_python.ObjectPath("objectpath_key_2"):
1346
mock_dbus_python.ObjectPath("objectpath_value_2")})
1347
ret = self.call_method(bus, "methodname", "busname",
1348
"objectpath", "interface")
1349
expected_method_return = {str(key): str(value)
1350
for key, value in
1351
method_return.items()}
1352
self.assertEqual(expected_method_return, ret)
1353
self.assertIsInstance(ret, dict)
1354
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1355
1356
def test_call_method_filters_dict_in_dict(self):
1357
def func():
1358
return method_return
1359
mock_dbus_python = self.MockDBusPython_func(func)
1360
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1361
method_return = mock_dbus_python.Dictionary(
1362
{"key1": mock_dbus_python.Dictionary({"key11": "value11",
1363
"key12": "value12"}),
1364
"key2": mock_dbus_python.Dictionary({"key21": "value21",
1365
"key22": "value22"})})
1366
ret = self.call_method(bus, "methodname", "busname",
1367
"objectpath", "interface")
1368
expected_method_return = {
1369
"key1": {"key11": "value11",
1370
"key12": "value12"},
1371
"key2": {"key21": "value21",
1372
"key22": "value22"},
1373
}
1374
self.assertEqual(expected_method_return, ret)
1375
self.assertIsInstance(ret, dict)
1376
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1377
for key, value in ret.items():
1378
self.assertIsInstance(value, dict)
1379
self.assertEqual(expected_method_return[key], value)
1380
self.assertNotIsInstance(value,
1381
mock_dbus_python.Dictionary)
1382
1383
def test_call_method_filters_dict_three_deep(self):
1384
def func():
1385
return method_return
1386
mock_dbus_python = self.MockDBusPython_func(func)
1387
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1388
method_return = mock_dbus_python.Dictionary(
1389
{"key1":
1390
mock_dbus_python.Dictionary(
1391
{"key2":
1392
mock_dbus_python.Dictionary(
1393
{"key3":
1394
mock_dbus_python.Boolean(True),
1395
}),
1396
}),
1397
})
1398
ret = self.call_method(bus, "methodname", "busname",
1399
"objectpath", "interface")
1400
expected_method_return = {"key1": {"key2": {"key3": True}}}
1401
self.assertEqual(expected_method_return, ret)
1402
self.assertIsInstance(ret, dict)
1403
self.assertNotIsInstance(ret, mock_dbus_python.Dictionary)
1404
self.assertIsInstance(ret["key1"], dict)
1405
self.assertNotIsInstance(ret["key1"],
1406
mock_dbus_python.Dictionary)
1407
self.assertIsInstance(ret["key1"]["key2"], dict)
1408
self.assertNotIsInstance(ret["key1"]["key2"],
1409
mock_dbus_python.Dictionary)
1410
self.assertTrue(ret["key1"]["key2"]["key3"])
1411
self.assertNotIsInstance(ret["key1"]["key2"]["key3"],
1412
mock_dbus_python.Boolean)
1413
1414
def test_call_method_handles_exception(self):
1075
class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):
1076
def test_calls_and_returns_get_object_on_bus(self):
1077
class MockBus(object):
1078
called = False
1079
def get_object(mockbus_self, busname, dbus_path):
1080
# Note that "self" is still the testcase instance,
1081
# this MockBus instance is in "mockbus_self".
1082
self.assertEqual(dbus_busname, busname)
1083
self.assertEqual(server_dbus_path, dbus_path)
1084
mockbus_self.called = True
1085
return mockbus_self
1086
1087
mockbus = get_mandos_dbus_object(bus=MockBus())
1088
self.assertIsInstance(mockbus, MockBus)
1089
self.assertTrue(mockbus.called)
1090
1091
def test_logs_and_exits_on_dbus_error(self):
1092
class FailingBusStub(object):
1093
def get_object(self, busname, dbus_path):
1094
raise dbus.exceptions.DBusException("Test")
1095
1096
with self.assertLogs(log, logging.CRITICAL):
1097
with self.assertRaises(SystemExit) as e:
1098
bus = get_mandos_dbus_object(bus=FailingBusStub())
1099
1100
if isinstance(e.exception.code, int):
1101
self.assertNotEqual(0, e.exception.code)
1102
else:
1103
self.assertIsNotNone(e.exception.code)
1104
1105
1106
class Test_get_managed_objects(TestCaseWithAssertLogs):
1107
def test_calls_and_returns_GetManagedObjects(self):
1108
managed_objects = {"/clients/client": { "Name": "client"}}
1109
class ObjectManagerStub(object):
1110
def GetManagedObjects(self):
1111
return managed_objects
1112
retval = get_managed_objects(ObjectManagerStub())
1113
self.assertDictEqual(managed_objects, retval)
1114
1115
def test_logs_and_exits_on_dbus_error(self):
1415
1116
dbus_logger = logging.getLogger("dbus.proxies")
1416
1117
1417
def func():
1418
dbus_logger.error("Test")
1419
raise mock_dbus_python.exceptions.DBusException()
1420
1421
mock_dbus_python = self.MockDBusPython_func(func)
1422
bus = dbus_python_adapter.SystemBus(mock_dbus_python)
1118
class ObjectManagerFailingStub(object):
1119
def GetManagedObjects(self):
1120
dbus_logger.error("Test")
1121
raise dbus.exceptions.DBusException("Test")
1423
1122
1424
1123
class CountingHandler(logging.Handler):
1425
1124
count = 0
1431
1130
dbus_logger.addHandler(counting_handler)
1432
1131
1433
1132
try:
1434
with self.assertRaises(dbus.Error) as e:
1435
self.call_method(bus, "methodname", "busname",
1436
"objectpath", "interface")
1133
with self.assertLogs(log, logging.CRITICAL) as watcher:
1134
with self.assertRaises(SystemExit) as e:
1135
get_managed_objects(ObjectManagerFailingStub())
1437
1136
finally:
1438
1137
dbus_logger.removeFilter(counting_handler)
1439
1138
1440
self.assertNotIsInstance(e, dbus.ConnectFailed)
1441
1442
1139
# Make sure the dbus logger was suppressed
1443
1140
self.assertEqual(0, counting_handler.count)
1444
1141
1445
def test_get_object_converts_to_correct_exception(self):
1446
bus = dbus_python_adapter.SystemBus(
1447
self.fake_dbus_python_raises_exception_on_connect)
1448
with self.assertRaises(dbus.ConnectFailed):
1449
self.call_method(bus, "methodname", "busname",
1450
"objectpath", "interface")
1451
1452
class fake_dbus_python_raises_exception_on_connect(object):
1453
"""fake dbus-python module"""
1454
class exceptions(object):
1455
"""Pseudo-namespace"""
1456
class DBusException(Exception):
1457
pass
1458
1459
@classmethod
1460
def SystemBus(cls):
1461
def get_object(busname, objectpath):
1462
raise cls.exceptions.DBusException()
1463
Bus = collections.namedtuple("Bus", ["get_object"])
1464
return Bus(get_object=get_object)
1465
1466
1467
class Test_dbus_python_adapter_CachingBus(unittest.TestCase):
1468
class mock_dbus_python(object):
1469
"""mock dbus-python modules"""
1470
class SystemBus(object):
1471
@staticmethod
1472
def get_object(busname, objectpath):
1473
return Unique()
1474
1475
def setUp(self):
1476
self.bus = dbus_python_adapter.CachingBus(
1477
self.mock_dbus_python)
1478
1479
def test_returns_distinct_objectpaths(self):
1480
obj1 = self.bus.get_object("busname", "objectpath1")
1481
self.assertIsInstance(obj1, Unique)
1482
obj2 = self.bus.get_object("busname", "objectpath2")
1483
self.assertIsInstance(obj2, Unique)
1484
self.assertIsNot(obj1, obj2)
1485
1486
def test_returns_distinct_busnames(self):
1487
obj1 = self.bus.get_object("busname1", "objectpath")
1488
self.assertIsInstance(obj1, Unique)
1489
obj2 = self.bus.get_object("busname2", "objectpath")
1490
self.assertIsInstance(obj2, Unique)
1491
self.assertIsNot(obj1, obj2)
1492
1493
def test_returns_distinct_both(self):
1494
obj1 = self.bus.get_object("busname1", "objectpath")
1495
self.assertIsInstance(obj1, Unique)
1496
obj2 = self.bus.get_object("busname2", "objectpath")
1497
self.assertIsInstance(obj2, Unique)
1498
self.assertIsNot(obj1, obj2)
1499
1500
def test_returns_same(self):
1501
obj1 = self.bus.get_object("busname", "objectpath")
1502
self.assertIsInstance(obj1, Unique)
1503
obj2 = self.bus.get_object("busname", "objectpath")
1504
self.assertIsInstance(obj2, Unique)
1505
self.assertIs(obj1, obj2)
1506
1507
def test_returns_same_old(self):
1508
obj1 = self.bus.get_object("busname1", "objectpath1")
1509
self.assertIsInstance(obj1, Unique)
1510
obj2 = self.bus.get_object("busname2", "objectpath2")
1511
self.assertIsInstance(obj2, Unique)
1512
obj1b = self.bus.get_object("busname1", "objectpath1")
1513
self.assertIsInstance(obj1b, Unique)
1514
self.assertIsNot(obj1, obj2)
1515
self.assertIsNot(obj2, obj1b)
1516
self.assertIs(obj1, obj1b)
1142
# Test that the dbus_logger still works
1143
with self.assertLogs(dbus_logger, logging.ERROR):
1144
dbus_logger.error("Test")
1145
1146
if isinstance(e.exception.code, int):
1147
self.assertNotEqual(0, e.exception.code)
1148
else:
1149
self.assertIsNotNone(e.exception.code)
1517
1150
1518
1151
1519
1152
class Test_commands_from_options(unittest.TestCase):
1520
1521
1153
def setUp(self):
1522
1154
self.parser = argparse.ArgumentParser()
1523
1155
add_command_line_options(self.parser)
1738
1370
class TestCommand(unittest.TestCase):
1739
1371
"""Abstract class for tests of command classes"""
1740
1372
1741
class FakeMandosBus(dbus.MandosBus):
1742
def __init__(self, testcase):
1743
self.client_properties = {
1744
"Name": "foo",
1745
"KeyID": ("92ed150794387c03ce684574b1139a65"
1746
"94a34f895daaaf09fd8ea90a27cddb12"),
1747
"Secret": b"secret",
1748
"Host": "foo.example.org",
1749
"Enabled": True,
1750
"Timeout": 300000,
1751
"LastCheckedOK": "2019-02-03T00:00:00",
1752
"Created": "2019-01-02T00:00:00",
1753
"Interval": 120000,
1754
"Fingerprint": ("778827225BA7DE539C5A"
1755
"7CFA59CFF7CDBD9A5920"),
1756
"CheckerRunning": False,
1757
"LastEnabled": "2019-01-03T00:00:00",
1758
"ApprovalPending": False,
1759
"ApprovedByDefault": True,
1760
"LastApprovalRequest": "",
1761
"ApprovalDelay": 0,
1762
"ApprovalDuration": 1000,
1763
"Checker": "fping -q -- %(host)s",
1764
"ExtendedTimeout": 900000,
1765
"Expires": "2019-02-04T00:00:00",
1766
"LastCheckerStatus": 0,
1767
}
1768
self.other_client_properties = {
1769
"Name": "barbar",
1770
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
1771
"6ab612cff5ad227247e46c2b020f441c"),
1772
"Secret": b"secretbar",
1773
"Host": "192.0.2.3",
1774
"Enabled": True,
1775
"Timeout": 300000,
1776
"LastCheckedOK": "2019-02-04T00:00:00",
1777
"Created": "2019-01-03T00:00:00",
1778
"Interval": 120000,
1779
"Fingerprint": ("3E393AEAEFB84C7E89E2"
1780
"F547B3A107558FCA3A27"),
1781
"CheckerRunning": True,
1782
"LastEnabled": "2019-01-04T00:00:00",
1783
"ApprovalPending": False,
1784
"ApprovedByDefault": False,
1785
"LastApprovalRequest": "2019-01-03T00:00:00",
1786
"ApprovalDelay": 30000,
1787
"ApprovalDuration": 93785000,
1788
"Checker": ":",
1789
"ExtendedTimeout": 900000,
1790
"Expires": "2019-02-05T00:00:00",
1791
"LastCheckerStatus": -2,
1792
}
1793
self.clients = collections.OrderedDict(
1794
[
1795
("client_objectpath", self.client_properties),
1796
("other_client_objectpath",
1797
self.other_client_properties),
1798
])
1799
self.one_client = {"client_objectpath":
1800
self.client_properties}
1801
self.testcase = testcase
1802
self.calls = []
1803
1804
def call_method(self, methodname, busname, objectpath,
1805
interface, *args):
1806
self.testcase.assertEqual("se.recompile.Mandos", busname)
1807
self.calls.append((methodname, busname, objectpath,
1808
interface, args))
1809
if interface == "org.freedesktop.DBus.Properties":
1810
if methodname == "Set":
1811
self.testcase.assertEqual(3, len(args))
1812
interface, key, value = args
1813
self.testcase.assertEqual(
1814
"se.recompile.Mandos.Client", interface)
1815
self.clients[objectpath][key] = value
1816
return
1817
elif interface == "se.recompile.Mandos":
1818
self.testcase.assertEqual("RemoveClient", methodname)
1819
self.testcase.assertEqual(1, len(args))
1820
clientpath = args[0]
1821
del self.clients[clientpath]
1822
return
1823
elif interface == "se.recompile.Mandos.Client":
1824
if methodname == "Approve":
1825
self.testcase.assertEqual(1, len(args))
1826
return
1827
raise ValueError()
1828
1829
1373
def setUp(self):
1830
self.bus = self.FakeMandosBus(self)
1374
testcase = self
1375
class MockClient(object):
1376
def __init__(self, name, **attributes):
1377
self.__dbus_object_path__ = "/clients/{}".format(name)
1378
self.attributes = attributes
1379
self.attributes["Name"] = name
1380
self.calls = []
1381
def Set(self, interface, propname, value, dbus_interface):
1382
testcase.assertEqual(client_dbus_interface, interface)
1383
testcase.assertEqual(dbus.PROPERTIES_IFACE,
1384
dbus_interface)
1385
self.attributes[propname] = value
1386
def Approve(self, approve, dbus_interface):
1387
testcase.assertEqual(client_dbus_interface,
1388
dbus_interface)
1389
self.calls.append(("Approve", (approve,
1390
dbus_interface)))
1391
self.client = MockClient(
1392
"foo",
1393
KeyID=("92ed150794387c03ce684574b1139a65"
1394
"94a34f895daaaf09fd8ea90a27cddb12"),
1395
Secret=b"secret",
1396
Host="foo.example.org",
1397
Enabled=dbus.Boolean(True),
1398
Timeout=300000,
1399
LastCheckedOK="2019-02-03T00:00:00",
1400
Created="2019-01-02T00:00:00",
1401
Interval=120000,
1402
Fingerprint=("778827225BA7DE539C5A"
1403
"7CFA59CFF7CDBD9A5920"),
1404
CheckerRunning=dbus.Boolean(False),
1405
LastEnabled="2019-01-03T00:00:00",
1406
ApprovalPending=dbus.Boolean(False),
1407
ApprovedByDefault=dbus.Boolean(True),
1408
LastApprovalRequest="",
1409
ApprovalDelay=0,
1410
ApprovalDuration=1000,
1411
Checker="fping -q -- %(host)s",
1412
ExtendedTimeout=900000,
1413
Expires="2019-02-04T00:00:00",
1414
LastCheckerStatus=0)
1415
self.other_client = MockClient(
1416
"barbar",
1417
KeyID=("0558568eedd67d622f5c83b35a115f79"
1418
"6ab612cff5ad227247e46c2b020f441c"),
1419
Secret=b"secretbar",
1420
Host="192.0.2.3",
1421
Enabled=dbus.Boolean(True),
1422
Timeout=300000,
1423
LastCheckedOK="2019-02-04T00:00:00",
1424
Created="2019-01-03T00:00:00",
1425
Interval=120000,
1426
Fingerprint=("3E393AEAEFB84C7E89E2"
1427
"F547B3A107558FCA3A27"),
1428
CheckerRunning=dbus.Boolean(True),
1429
LastEnabled="2019-01-04T00:00:00",
1430
ApprovalPending=dbus.Boolean(False),
1431
ApprovedByDefault=dbus.Boolean(False),
1432
LastApprovalRequest="2019-01-03T00:00:00",
1433
ApprovalDelay=30000,
1434
ApprovalDuration=93785000,
1435
Checker=":",
1436
ExtendedTimeout=900000,
1437
Expires="2019-02-05T00:00:00",
1438
LastCheckerStatus=-2)
1439
self.clients = collections.OrderedDict(
1440
[
1441
(self.client.__dbus_object_path__,
1442
self.client.attributes),
1443
(self.other_client.__dbus_object_path__,
1444
self.other_client.attributes),
1445
])
1446
self.one_client = {self.client.__dbus_object_path__:
1447
self.client.attributes}
1448
1449
@property
1450
def bus(self):
1451
class MockBus(object):
1452
@staticmethod
1453
def get_object(client_bus_name, path):
1454
self.assertEqual(dbus_busname, client_bus_name)
1455
# Note: "self" here is the TestCmd instance, not the
1456
# MockBus instance, since this is a static method!
1457
if path == self.client.__dbus_object_path__:
1458
return self.client
1459
elif path == self.other_client.__dbus_object_path__:
1460
return self.other_client
1461
return MockBus()
1831
1462
1832
1463
1833
1464
class TestBaseCommands(TestCommand):
1834
1465
1835
1466
def test_IsEnabled_exits_successfully(self):
1836
1467
with self.assertRaises(SystemExit) as e:
1837
command.IsEnabled().run(self.bus.one_client)
1468
command.IsEnabled().run(self.one_client)
1838
1469
if e.exception.code is not None:
1839
1470
self.assertEqual(0, e.exception.code)
1840
1471
else:
1841
1472
self.assertIsNone(e.exception.code)
1842
1473
1843
1474
def test_IsEnabled_exits_with_failure(self):
1844
self.bus.client_properties["Enabled"] = False
1475
self.client.attributes["Enabled"] = dbus.Boolean(False)
1845
1476
with self.assertRaises(SystemExit) as e:
1846
command.IsEnabled().run(self.bus.one_client)
1477
command.IsEnabled().run(self.one_client)
1847
1478
if isinstance(e.exception.code, int):
1848
1479
self.assertNotEqual(0, e.exception.code)
1849
1480
else:
1850
1481
self.assertIsNotNone(e.exception.code)
1851
1482
1852
1483
def test_Approve(self):
1853
busname = "se.recompile.Mandos"
1854
client_interface = "se.recompile.Mandos.Client"
1855
command.Approve().run(self.bus.clients, self.bus)
1856
for clientpath in self.bus.clients:
1857
self.assertIn(("Approve", busname, clientpath,
1858
client_interface, (True,)), self.bus.calls)
1484
command.Approve().run(self.clients, self.bus)
1485
for clientpath in self.clients:
1486
client = self.bus.get_object(dbus_busname, clientpath)
1487
self.assertIn(("Approve", (True, client_dbus_interface)),
1488
client.calls)
1859
1489
1860
1490
def test_Deny(self):
1861
busname = "se.recompile.Mandos"
1862
client_interface = "se.recompile.Mandos.Client"
1863
command.Deny().run(self.bus.clients, self.bus)
1864
for clientpath in self.bus.clients:
1865
self.assertIn(("Approve", busname, clientpath,
1866
client_interface, (False,)),
1867
self.bus.calls)
1491
command.Deny().run(self.clients, self.bus)
1492
for clientpath in self.clients:
1493
client = self.bus.get_object(dbus_busname, clientpath)
1494
self.assertIn(("Approve", (False, client_dbus_interface)),
1495
client.calls)
1868
1496
1869
1497
def test_Remove(self):
1870
command.Remove().run(self.bus.clients, self.bus)
1871
for clientpath in self.bus.clients:
1872
self.assertIn(("RemoveClient", dbus_busname,
1873
dbus_server_path, dbus_server_interface,
1874
(clientpath,)), self.bus.calls)
1498
class MandosSpy(object):
1499
def __init__(self):
1500
self.calls = []
1501
def RemoveClient(self, dbus_path):
1502
self.calls.append(("RemoveClient", (dbus_path,)))
1503
mandos = MandosSpy()
1504
command.Remove().run(self.clients, self.bus, mandos)
1505
for clientpath in self.clients:
1506
self.assertIn(("RemoveClient", (clientpath,)),
1507
mandos.calls)
1875
1508
1876
1509
expected_json = {
1877
1510
"foo": {
1926
1559
1927
1560
def test_DumpJSON_normal(self):
1928
1561
with self.capture_stdout_to_buffer() as buffer:
1929
command.DumpJSON().run(self.bus.clients)
1562
command.DumpJSON().run(self.clients)
1930
1563
json_data = json.loads(buffer.getvalue())
1931
1564
self.assertDictEqual(self.expected_json, json_data)
1932
1565
1943
1576
1944
1577
def test_DumpJSON_one_client(self):
1945
1578
with self.capture_stdout_to_buffer() as buffer:
1946
command.DumpJSON().run(self.bus.one_client)
1579
command.DumpJSON().run(self.one_client)
1947
1580
json_data = json.loads(buffer.getvalue())
1948
1581
expected_json = {"foo": self.expected_json["foo"]}
1949
1582
self.assertDictEqual(expected_json, json_data)
1950
1583
1951
1584
def test_PrintTable_normal(self):
1952
1585
with self.capture_stdout_to_buffer() as buffer:
1953
command.PrintTable().run(self.bus.clients)
1586
command.PrintTable().run(self.clients)
1954
1587
expected_output = "\n".join((
1955
1588
"Name Enabled Timeout Last Successful Check",
1956
1589
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1960
1593
1961
1594
def test_PrintTable_verbose(self):
1962
1595
with self.capture_stdout_to_buffer() as buffer:
1963
command.PrintTable(verbose=True).run(self.bus.clients)
1596
command.PrintTable(verbose=True).run(self.clients)
1964
1597
columns = (
1965
1598
(
1966
1599
"Name ",
2056
1689
2057
1690
def test_PrintTable_one_client(self):
2058
1691
with self.capture_stdout_to_buffer() as buffer:
2059
command.PrintTable().run(self.bus.one_client)
1692
command.PrintTable().run(self.one_client)
2060
1693
expected_output = "\n".join((
2061
1694
"Name Enabled Timeout Last Successful Check",
2062
1695
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
2066
1699
2067
1700
class TestPropertySetterCmd(TestCommand):
2068
1701
"""Abstract class for tests of command.PropertySetter classes"""
2069
2070
1702
def runTest(self):
2071
1703
if not hasattr(self, "command"):
2072
return # Abstract TestCase class
1704
return
2073
1705
values_to_get = getattr(self, "values_to_get",
2074
1706
self.values_to_set)
2075
1707
for value_to_set, value_to_get in zip(self.values_to_set,
2076
1708
values_to_get):
2077
for clientpath in self.bus.clients:
2078
self.bus.clients[clientpath][self.propname] = Unique()
2079
self.run_command(value_to_set, self.bus.clients)
2080
for clientpath in self.bus.clients:
2081
value = self.bus.clients[clientpath][self.propname]
2082
self.assertNotIsInstance(value, Unique)
1709
for clientpath in self.clients:
1710
client = self.bus.get_object(dbus_busname, clientpath)
1711
old_value = client.attributes[self.propname]
1712
client.attributes[self.propname] = self.Unique()
1713
self.run_command(value_to_set, self.clients)
1714
for clientpath in self.clients:
1715
client = self.bus.get_object(dbus_busname, clientpath)
1716
value = client.attributes[self.propname]
1717
self.assertNotIsInstance(value, self.Unique)
2083
1718
self.assertEqual(value_to_get, value)
2084
1719
1720
class Unique(object):
1721
"""Class for objects which exist only to be unique objects,
1722
since unittest.mock.sentinel only exists in Python 3.3"""
1723
2085
1724
def run_command(self, value, clients):
2086
1725
self.command().run(clients, self.bus)
2087
1726
2089
1728
class TestEnableCmd(TestPropertySetterCmd):
2090
1729
command = command.Enable
2091
1730
propname = "Enabled"
2092
values_to_set = [True]
1731
values_to_set = [dbus.Boolean(True)]
2093
1732
2094
1733
2095
1734
class TestDisableCmd(TestPropertySetterCmd):
2096
1735
command = command.Disable
2097
1736
propname = "Enabled"
2098
values_to_set = [False]
1737
values_to_set = [dbus.Boolean(False)]
2099
1738
2100
1739
2101
1740
class TestBumpTimeoutCmd(TestPropertySetterCmd):
2107
1746
class TestStartCheckerCmd(TestPropertySetterCmd):
2108
1747
command = command.StartChecker
2109
1748
propname = "CheckerRunning"
2110
values_to_set = [True]
1749
values_to_set = [dbus.Boolean(True)]
2111
1750
2112
1751
2113
1752
class TestStopCheckerCmd(TestPropertySetterCmd):
2114
1753
command = command.StopChecker
2115
1754
propname = "CheckerRunning"
2116
values_to_set = [False]
1755
values_to_set = [dbus.Boolean(False)]
2117
1756
2118
1757
2119
1758
class TestApproveByDefaultCmd(TestPropertySetterCmd):
2120
1759
command = command.ApproveByDefault
2121
1760
propname = "ApprovedByDefault"
2122
values_to_set = [True]
1761
values_to_set = [dbus.Boolean(True)]
2123
1762
2124
1763
2125
1764
class TestDenyByDefaultCmd(TestPropertySetterCmd):
2126
1765
command = command.DenyByDefault
2127
1766
propname = "ApprovedByDefault"
2128
values_to_set = [False]
1767
values_to_set = [dbus.Boolean(False)]
2129
1768
2130
1769
2131
1770
class TestPropertySetterValueCmd(TestPropertySetterCmd):
2132
1771
"""Abstract class for tests of PropertySetterValueCmd classes"""
2133
1772
1773
def runTest(self):
1774
if type(self) is TestPropertySetterValueCmd:
1775
return
1776
return super(TestPropertySetterValueCmd, self).runTest()
1777
2134
1778
def run_command(self, value, clients):
2135
1779
self.command(value).run(clients, self.bus)
2136
1780
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0