236
def print_clients(clients, keywords):
237
def valuetostring(value, keyword):
238
if type(value) is dbus.Boolean:
239
return "Yes" if value else "No"
240
if keyword in ("Timeout", "Interval", "ApprovalDelay",
241
"ApprovalDuration", "ExtendedTimeout"):
242
return milliseconds_to_string(value)
245
# Create format string to print table rows
246
format_string = " ".join("{{{key}:{width}}}".format(
247
width=max(len(tablewords[key]),
248
max(len(valuetostring(client[key], key))
249
for client in clients)),
253
print(format_string.format(**tablewords))
254
for client in clients:
256
.format(**{key: valuetostring(client[key], key)
257
for key in keywords}))
271
## Classes for commands.
273
# Abstract classes first
274
class Command(object):
275
"""Abstract class for commands"""
276
def run(self, mandos, clients):
277
"""Normal commands should implement run_on_one_client(), but
278
commands which want to operate on all clients at the same time
279
can override this run() method instead."""
281
for client in clients:
282
self.run_on_one_client(client)
284
class PrintCmd(Command):
285
"""Abstract class for commands printing client details"""
286
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
287
"Created", "Interval", "Host", "KeyID",
288
"Fingerprint", "CheckerRunning", "LastEnabled",
289
"ApprovalPending", "ApprovedByDefault",
290
"LastApprovalRequest", "ApprovalDelay",
291
"ApprovalDuration", "Checker", "ExtendedTimeout",
292
"Expires", "LastCheckerStatus")
293
def run(self, mandos, clients):
294
print(self.output(clients))
296
class PropertyCmd(Command):
297
"""Abstract class for Actions for setting one client property"""
298
def run_on_one_client(self, client):
299
"""Set the Client's D-Bus property"""
300
client.Set(client_interface, self.property, self.value_to_set,
301
dbus_interface=dbus.PROPERTIES_IFACE)
303
class ValueArgumentMixIn(object):
304
"""Mixin class for commands taking a value as argument"""
305
def __init__(self, value):
306
self.value_to_set = value
308
class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):
309
"""Mixin class for commands taking a value argument as
312
def value_to_set(self):
315
def value_to_set(self, value):
316
"""When setting, convert value to a datetime.timedelta"""
317
self._vts = string_to_delta(value).total_seconds() * 1000
319
# Actual (non-abstract) command classes
321
class PrintTableCmd(PrintCmd):
322
def __init__(self, verbose=False):
323
self.verbose = verbose
325
def output(self, clients):
327
keywords = self.all_keywords
329
keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
330
return str(self.TableOfClients(clients.values(), keywords))
332
class TableOfClients(object):
335
"Enabled": "Enabled",
336
"Timeout": "Timeout",
337
"LastCheckedOK": "Last Successful Check",
338
"LastApprovalRequest": "Last Approval Request",
339
"Created": "Created",
340
"Interval": "Interval",
342
"Fingerprint": "Fingerprint",
344
"CheckerRunning": "Check Is Running",
345
"LastEnabled": "Last Enabled",
346
"ApprovalPending": "Approval Is Pending",
347
"ApprovedByDefault": "Approved By Default",
348
"ApprovalDelay": "Approval Delay",
349
"ApprovalDuration": "Approval Duration",
350
"Checker": "Checker",
351
"ExtendedTimeout": "Extended Timeout",
352
"Expires": "Expires",
353
"LastCheckerStatus": "Last Checker Status",
356
def __init__(self, clients, keywords, tableheaders=None):
357
self.clients = clients
358
self.keywords = keywords
359
if tableheaders is not None:
360
self.tableheaders = tableheaders
363
return "\n".join(self.rows())
365
if sys.version_info.major == 2:
366
__unicode__ = __str__
368
return str(self).encode(locale.getpreferredencoding())
371
format_string = self.row_formatting_string()
372
rows = [self.header_line(format_string)]
373
rows.extend(self.client_line(client, format_string)
374
for client in self.clients)
377
def row_formatting_string(self):
378
"Format string used to format table rows"
379
return " ".join("{{{key}:{width}}}".format(
380
width=max(len(self.tableheaders[key]),
381
*(len(self.string_from_client(client, key))
382
for client in self.clients)),
384
for key in self.keywords)
386
def string_from_client(self, client, key):
387
return self.valuetostring(client[key], key)
390
def valuetostring(value, keyword):
391
if isinstance(value, dbus.Boolean):
392
return "Yes" if value else "No"
393
if keyword in ("Timeout", "Interval", "ApprovalDelay",
394
"ApprovalDuration", "ExtendedTimeout"):
395
return milliseconds_to_string(value)
398
def header_line(self, format_string):
399
return format_string.format(**self.tableheaders)
401
def client_line(self, client, format_string):
402
return format_string.format(
403
**{key: self.string_from_client(client, key)
404
for key in self.keywords})
408
class DumpJSONCmd(PrintCmd):
409
def output(self, clients):
410
data = {client["Name"]:
411
{key: self.dbus_boolean_to_bool(client[key])
412
for key in self.all_keywords}
413
for client in clients.values()}
414
return json.dumps(data, indent=4, separators=(',', ': '))
416
def dbus_boolean_to_bool(value):
417
if isinstance(value, dbus.Boolean):
421
class IsEnabledCmd(Command):
422
def run_on_one_client(self, client):
423
if self.is_enabled(client):
426
def is_enabled(self, client):
427
return client.Get(client_interface, "Enabled",
428
dbus_interface=dbus.PROPERTIES_IFACE)
430
class RemoveCmd(Command):
431
def run_on_one_client(self, client):
432
self.mandos.RemoveClient(client.__dbus_object_path__)
434
class ApproveCmd(Command):
435
def run_on_one_client(self, client):
436
client.Approve(dbus.Boolean(True),
437
dbus_interface=client_interface)
439
class DenyCmd(Command):
440
def run_on_one_client(self, client):
441
client.Approve(dbus.Boolean(False),
442
dbus_interface=client_interface)
444
class EnableCmd(PropertyCmd):
446
value_to_set = dbus.Boolean(True)
448
class DisableCmd(PropertyCmd):
450
value_to_set = dbus.Boolean(False)
452
class BumpTimeoutCmd(PropertyCmd):
453
property = "LastCheckedOK"
456
class StartCheckerCmd(PropertyCmd):
457
property = "CheckerRunning"
458
value_to_set = dbus.Boolean(True)
460
class StopCheckerCmd(PropertyCmd):
461
property = "CheckerRunning"
462
value_to_set = dbus.Boolean(False)
464
class ApproveByDefaultCmd(PropertyCmd):
465
property = "ApprovedByDefault"
466
value_to_set = dbus.Boolean(True)
468
class DenyByDefaultCmd(PropertyCmd):
469
property = "ApprovedByDefault"
470
value_to_set = dbus.Boolean(False)
472
class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):
475
class SetHostCmd(PropertyCmd, ValueArgumentMixIn):
478
class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):
481
class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
484
class SetExtendedTimeoutCmd(PropertyCmd,
485
MillisecondsValueArgumentMixIn):
486
property = "ExtendedTimeout"
488
class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
489
property = "Interval"
491
class SetApprovalDelayCmd(PropertyCmd,
492
MillisecondsValueArgumentMixIn):
493
property = "ApprovalDelay"
495
class SetApprovalDurationCmd(PropertyCmd,
496
MillisecondsValueArgumentMixIn):
497
property = "ApprovalDuration"
260
499
def has_actions(options):
261
500
return any((options.enable,
325
570
parser.add_argument("-s", "--secret",
326
571
type=argparse.FileType(mode="rb"),
327
572
help="Set password blob (file) for client")
328
parser.add_argument("-A", "--approve", action="store_true",
329
help="Approve any current client request")
330
parser.add_argument("-D", "--deny", action="store_true",
331
help="Deny any current client request")
573
approve_deny = parser.add_mutually_exclusive_group()
574
approve_deny.add_argument(
575
"-A", "--approve", action="store_true",
576
help="Approve any current client request")
577
approve_deny.add_argument("-D", "--deny", action="store_true",
578
help="Deny any current client request")
332
579
parser.add_argument("--check", action="store_true",
333
580
help="Run self-test")
334
581
parser.add_argument("client", nargs="*", help="Client name")
584
def commands_and_clients_from_options(options):
588
if options.dump_json:
589
commands.append(DumpJSONCmd())
592
commands.append(EnableCmd())
595
commands.append(DisableCmd())
597
if options.bump_timeout:
598
commands.append(BumpTimeoutCmd(options.bump_timeout))
600
if options.start_checker:
601
commands.append(StartCheckerCmd())
603
if options.stop_checker:
604
commands.append(StopCheckerCmd())
606
if options.is_enabled:
607
commands.append(IsEnabledCmd())
610
commands.append(RemoveCmd())
612
if options.checker is not None:
613
commands.append(SetCheckerCmd())
615
if options.timeout is not None:
616
commands.append(SetTimeoutCmd(options.timeout))
618
if options.extended_timeout:
620
SetExtendedTimeoutCmd(options.extended_timeout))
622
if options.interval is not None:
623
command.append(SetIntervalCmd(options.interval))
625
if options.approved_by_default is not None:
626
if options.approved_by_default:
627
command.append(ApproveByDefaultCmd())
629
command.append(DenyByDefaultCmd())
631
if options.approval_delay is not None:
632
command.append(SetApprovalDelayCmd(options.approval_delay))
634
if options.approval_duration is not None:
636
SetApprovalDurationCmd(options.approval_duration))
638
if options.host is not None:
639
command.append(SetHostCmd(options.host))
641
if options.secret is not None:
642
command.append(SetSecretCmd(options.secret))
645
commands.append(ApproveCmd())
648
commands.append(DenyCmd())
650
# If no command option has been given, show table of clients,
651
# optionally verbosely
653
commands.append(PrintTableCmd(verbose=options.verbose))
655
return commands, options.client
659
parser = argparse.ArgumentParser()
661
add_command_line_options(parser)
335
663
options = parser.parse_args()
337
665
if has_actions(options) and not (options.client or options.all):
361
688
mandos_serv_object_manager = dbus.Interface(
362
689
mandos_dbus_objc, dbus_interface=dbus.OBJECT_MANAGER_IFACE)
364
# block stderr since dbus library prints to stderr
365
null = os.open(os.path.devnull, os.O_RDWR)
366
stderrcopy = os.dup(sys.stderr.fileno())
367
os.dup2(null, sys.stderr.fileno())
691
# Filter out log message from dbus module
692
dbus_logger = logging.getLogger("dbus.proxies")
693
class NullFilter(logging.Filter):
694
def filter(self, record):
696
dbus_filter = NullFilter()
371
mandos_clients = {path: ifs_and_props[client_interface]
372
for path, ifs_and_props in
373
mandos_serv_object_manager
374
.GetManagedObjects().items()
375
if client_interface in ifs_and_props}
378
os.dup2(stderrcopy, sys.stderr.fileno())
698
dbus_logger.addFilter(dbus_filter)
699
mandos_clients = {path: ifs_and_props[client_interface]
700
for path, ifs_and_props in
701
mandos_serv_object_manager
702
.GetManagedObjects().items()
703
if client_interface in ifs_and_props}
380
704
except dbus.exceptions.DBusException as e:
381
print("Access denied: "
382
"Accessing mandos server through D-Bus: {}".format(e),
705
log.critical("Failed to access Mandos server through D-Bus:"
709
# restore dbus logger
710
dbus_logger.removeFilter(dbus_filter)
386
712
# Compile dict of (clients: properties) to process
389
if options.all or not options.client:
390
716
clients = {bus.get_object(busname, path): properties
391
717
for path, properties in mandos_clients.items()}
393
for name in options.client:
719
for name in clientnames:
394
720
for path, client in mandos_clients.items():
395
721
if client["Name"] == name:
396
722
client_objc = bus.get_object(busname, path)
397
723
clients[client_objc] = client
400
print("Client not found on server: {!r}"
401
.format(name), file=sys.stderr)
726
log.critical("Client not found on server: %r", name)
404
if not has_actions(options) and clients:
405
if options.verbose or options.dump_json:
406
keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
407
"Created", "Interval", "Host", "Fingerprint",
408
"CheckerRunning", "LastEnabled",
409
"ApprovalPending", "ApprovedByDefault",
410
"LastApprovalRequest", "ApprovalDelay",
411
"ApprovalDuration", "Checker",
412
"ExtendedTimeout", "Expires",
415
keywords = defaultkeywords
417
if options.dump_json:
418
json.dump({client["Name"]: {key:
420
if isinstance(client[key],
424
for client in clients.values()},
425
fp=sys.stdout, indent=4,
426
separators=(',', ': '))
429
print_clients(clients.values(), keywords)
431
# Process each client in the list by all selected options
432
for client in clients:
434
def set_client_prop(prop, value):
435
"""Set a Client D-Bus property"""
436
client.Set(client_interface, prop, value,
437
dbus_interface=dbus.PROPERTIES_IFACE)
439
def set_client_prop_ms(prop, value):
440
"""Set a Client D-Bus property, converted
441
from a string to milliseconds."""
442
set_client_prop(prop,
443
string_to_delta(value).total_seconds()
447
mandos_serv.RemoveClient(client.__dbus_object_path__)
449
set_client_prop("Enabled", dbus.Boolean(True))
451
set_client_prop("Enabled", dbus.Boolean(False))
452
if options.bump_timeout:
453
set_client_prop("LastCheckedOK", "")
454
if options.start_checker:
455
set_client_prop("CheckerRunning", dbus.Boolean(True))
456
if options.stop_checker:
457
set_client_prop("CheckerRunning", dbus.Boolean(False))
458
if options.is_enabled:
459
if client.Get(client_interface, "Enabled",
460
dbus_interface=dbus.PROPERTIES_IFACE):
464
if options.checker is not None:
465
set_client_prop("Checker", options.checker)
466
if options.host is not None:
467
set_client_prop("Host", options.host)
468
if options.interval is not None:
469
set_client_prop_ms("Interval", options.interval)
470
if options.approval_delay is not None:
471
set_client_prop_ms("ApprovalDelay",
472
options.approval_delay)
473
if options.approval_duration is not None:
474
set_client_prop_ms("ApprovalDuration",
475
options.approval_duration)
476
if options.timeout is not None:
477
set_client_prop_ms("Timeout", options.timeout)
478
if options.extended_timeout is not None:
479
set_client_prop_ms("ExtendedTimeout",
480
options.extended_timeout)
481
if options.secret is not None:
482
set_client_prop("Secret",
483
dbus.ByteArray(options.secret.read()))
484
if options.approved_by_default is not None:
485
set_client_prop("ApprovedByDefault",
487
.approved_by_default))
489
client.Approve(dbus.Boolean(True),
490
dbus_interface=client_interface)
492
client.Approve(dbus.Boolean(False),
493
dbus_interface=client_interface)
729
# Run all commands on clients
730
for command in commands:
731
command.run(mandos_serv, clients)
734
class Test_milliseconds_to_string(unittest.TestCase):
736
self.assertEqual(milliseconds_to_string(93785000),
738
def test_no_days(self):
739
self.assertEqual(milliseconds_to_string(7385000), "02:03:05")
740
def test_all_zero(self):
741
self.assertEqual(milliseconds_to_string(0), "00:00:00")
742
def test_no_fractional_seconds(self):
743
self.assertEqual(milliseconds_to_string(400), "00:00:00")
744
self.assertEqual(milliseconds_to_string(900), "00:00:00")
745
self.assertEqual(milliseconds_to_string(1900), "00:00:01")
747
class Test_string_to_delta(unittest.TestCase):
748
def test_handles_basic_rfc3339(self):
749
self.assertEqual(string_to_delta("PT2H"),
750
datetime.timedelta(0, 7200))
751
def test_falls_back_to_pre_1_6_1_with_warning(self):
752
# assertLogs only exists in Python 3.4
753
if hasattr(self, "assertLogs"):
754
with self.assertLogs(log, logging.WARNING):
755
value = string_to_delta("2h")
757
class WarningFilter(logging.Filter):
758
"""Don't show, but record the presence of, warnings"""
759
def filter(self, record):
760
is_warning = record.levelno >= logging.WARNING
761
self.found = is_warning or getattr(self, "found",
763
return not is_warning
764
warning_filter = WarningFilter()
765
log.addFilter(warning_filter)
767
value = string_to_delta("2h")
769
log.removeFilter(warning_filter)
770
self.assertTrue(getattr(warning_filter, "found", False))
771
self.assertEqual(value, datetime.timedelta(0, 7200))
774
class TestCmd(unittest.TestCase):
775
"""Abstract class for tests of command classes"""
778
class MockClient(object):
779
def __init__(self, name, **attributes):
780
self.__dbus_object_path__ = "objpath_{}".format(name)
781
self.attributes = attributes
782
self.attributes["Name"] = name
784
def Set(self, interface, property, value, dbus_interface):
785
testcase.assertEqual(interface, client_interface)
786
testcase.assertEqual(dbus_interface,
787
dbus.PROPERTIES_IFACE)
788
self.attributes[property] = value
789
self.calls.append(("Set", (interface, property, value,
791
def Get(self, interface, property, dbus_interface):
792
testcase.assertEqual(interface, client_interface)
793
testcase.assertEqual(dbus_interface,
794
dbus.PROPERTIES_IFACE)
795
self.calls.append(("Get", (interface, property,
797
return self.attributes[property]
798
def __getitem__(self, key):
799
return self.attributes[key]
800
def __setitem__(self, key, value):
801
self.attributes[key] = value
802
self.clients = collections.OrderedDict([
806
KeyID=("92ed150794387c03ce684574b1139a65"
807
"94a34f895daaaf09fd8ea90a27cddb12"),
809
Host="foo.example.org",
810
Enabled=dbus.Boolean(True),
812
LastCheckedOK="2019-02-03T00:00:00",
813
Created="2019-01-02T00:00:00",
815
Fingerprint=("778827225BA7DE539C5A"
816
"7CFA59CFF7CDBD9A5920"),
817
CheckerRunning=dbus.Boolean(False),
818
LastEnabled="2019-01-03T00:00:00",
819
ApprovalPending=dbus.Boolean(False),
820
ApprovedByDefault=dbus.Boolean(True),
821
LastApprovalRequest="",
823
ApprovalDuration=1000,
824
Checker="fping -q -- %(host)s",
825
ExtendedTimeout=900000,
826
Expires="2019-02-04T00:00:00",
827
LastCheckerStatus=0)),
831
KeyID=("0558568eedd67d622f5c83b35a115f79"
832
"6ab612cff5ad227247e46c2b020f441c"),
835
Enabled=dbus.Boolean(True),
837
LastCheckedOK="2019-02-04T00:00:00",
838
Created="2019-01-03T00:00:00",
840
Fingerprint=("3E393AEAEFB84C7E89E2"
841
"F547B3A107558FCA3A27"),
842
CheckerRunning=dbus.Boolean(True),
843
LastEnabled="2019-01-04T00:00:00",
844
ApprovalPending=dbus.Boolean(False),
845
ApprovedByDefault=dbus.Boolean(False),
846
LastApprovalRequest="2019-01-03T00:00:00",
848
ApprovalDuration=1000,
850
ExtendedTimeout=900000,
851
Expires="2019-02-05T00:00:00",
852
LastCheckerStatus=-2)),
855
class TestPrintTableCmd(TestCmd):
856
def test_normal(self):
857
output = PrintTableCmd().output(self.clients)
858
expected_output = """
859
Name Enabled Timeout Last Successful Check
860
foo Yes 00:05:00 2019-02-03T00:00:00
861
barbar Yes 00:05:00 2019-02-04T00:00:00
863
self.assertEqual(output, expected_output)
864
def test_verbose(self):
865
output = PrintTableCmd(verbose=True).output(self.clients)
866
expected_output = """
867
Name Enabled Timeout Last Successful Check Created Interval Host Key ID Fingerprint Check Is Running Last Enabled Approval Is Pending Approved By Default Last Approval Request Approval Delay Approval Duration Checker Extended Timeout Expires Last Checker Status
868
foo Yes 00:05:00 2019-02-03T00:00:00 2019-01-02T00:00:00 00:02:00 foo.example.org 92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8ea90a27cddb12 778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 No 2019-01-03T00:00:00 No Yes 00:00:00 00:00:01 fping -q -- %(host)s 00:15:00 2019-02-04T00:00:00 0
869
barbar Yes 00:05:00 2019-02-04T00:00:00 2019-01-03T00:00:00 00:02:00 192.0.2.3 0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e46c2b020f441c 3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 Yes 2019-01-04T00:00:00 No No 2019-01-03T00:00:00 00:00:30 00:00:01 : 00:15:00 2019-02-05T00:00:00 -2
871
self.assertEqual(output, expected_output)
872
def test_one_client(self):
873
output = PrintTableCmd().output({"foo": self.clients["foo"]})
874
expected_output = """
875
Name Enabled Timeout Last Successful Check
876
foo Yes 00:05:00 2019-02-03T00:00:00
878
self.assertEqual(output, expected_output)
880
class TestDumpJSONCmd(TestCmd):
882
self.expected_json = {
885
"KeyID": ("92ed150794387c03ce684574b1139a65"
886
"94a34f895daaaf09fd8ea90a27cddb12"),
887
"Host": "foo.example.org",
890
"LastCheckedOK": "2019-02-03T00:00:00",
891
"Created": "2019-01-02T00:00:00",
893
"Fingerprint": ("778827225BA7DE539C5A"
894
"7CFA59CFF7CDBD9A5920"),
895
"CheckerRunning": False,
896
"LastEnabled": "2019-01-03T00:00:00",
897
"ApprovalPending": False,
898
"ApprovedByDefault": True,
899
"LastApprovalRequest": "",
901
"ApprovalDuration": 1000,
902
"Checker": "fping -q -- %(host)s",
903
"ExtendedTimeout": 900000,
904
"Expires": "2019-02-04T00:00:00",
905
"LastCheckerStatus": 0,
909
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
910
"6ab612cff5ad227247e46c2b020f441c"),
914
"LastCheckedOK": "2019-02-04T00:00:00",
915
"Created": "2019-01-03T00:00:00",
917
"Fingerprint": ("3E393AEAEFB84C7E89E2"
918
"F547B3A107558FCA3A27"),
919
"CheckerRunning": True,
920
"LastEnabled": "2019-01-04T00:00:00",
921
"ApprovalPending": False,
922
"ApprovedByDefault": False,
923
"LastApprovalRequest": "2019-01-03T00:00:00",
924
"ApprovalDelay": 30000,
925
"ApprovalDuration": 1000,
927
"ExtendedTimeout": 900000,
928
"Expires": "2019-02-05T00:00:00",
929
"LastCheckerStatus": -2,
932
return super(TestDumpJSONCmd, self).setUp()
933
def test_normal(self):
934
json_data = json.loads(DumpJSONCmd().output(self.clients))
935
self.assertDictEqual(json_data, self.expected_json)
936
def test_one_client(self):
937
clients = {"foo": self.clients["foo"]}
938
json_data = json.loads(DumpJSONCmd().output(clients))
939
expected_json = {"foo": self.expected_json["foo"]}
940
self.assertDictEqual(json_data, expected_json)
942
class TestIsEnabledCmd(TestCmd):
943
def test_is_enabled(self):
944
self.assertTrue(all(IsEnabledCmd().is_enabled(client)
945
for client in self.clients.values()))
946
def test_is_enabled_does_get_attribute(self):
947
client = self.clients["foo"]
948
self.assertTrue(IsEnabledCmd().is_enabled(client))
949
self.assertListEqual(client.calls,
951
("se.recompile.Mandos.Client",
953
"org.freedesktop.DBus.Properties"))])
954
def test_is_enabled_run_exits_successfully(self):
955
client = self.clients["foo"]
956
with self.assertRaises(SystemExit) as e:
957
IsEnabledCmd().run_on_one_client(client)
958
if e.exception.code is not None:
959
self.assertEqual(e.exception.code, 0)
961
self.assertIsNone(e.exception.code)
962
def test_is_enabled_run_exits_with_failure(self):
963
client = self.clients["foo"]
964
client["Enabled"] = dbus.Boolean(False)
965
with self.assertRaises(SystemExit) as e:
966
IsEnabledCmd().run_on_one_client(client)
967
if isinstance(e.exception.code, int):
968
self.assertNotEqual(e.exception.code, 0)
970
self.assertIsNotNone(e.exception.code)
974
def should_only_run_tests():
975
parser = argparse.ArgumentParser(add_help=False)
976
parser.add_argument("--check", action='store_true')
977
args, unknown_args = parser.parse_known_args()
978
run_tests = args.check
980
# Remove --check argument from sys.argv
981
sys.argv[1:] = unknown_args
984
# Add all tests from doctest strings
985
def load_tests(loader, tests, none):
987
tests.addTests(doctest.DocTestSuite())
496
990
if __name__ == "__main__":
991
if should_only_run_tests():
992
# Call using ./tdd-python-script --check [--verbose]