385
def check_option_syntax(parser, options):
386
"""Apply additional restrictions on options, not expressible in
389
def has_actions(options):
390
return any((options.enable,
392
options.bump_timeout,
393
options.start_checker,
394
options.stop_checker,
397
options.checker is not None,
398
options.timeout is not None,
399
options.extended_timeout is not None,
400
options.interval is not None,
401
options.approved_by_default is not None,
402
options.approval_delay is not None,
403
options.approval_duration is not None,
404
options.host is not None,
405
options.secret is not None,
271
class TableOfClients(object):
274
"Enabled": "Enabled",
275
"Timeout": "Timeout",
276
"LastCheckedOK": "Last Successful Check",
277
"LastApprovalRequest": "Last Approval Request",
278
"Created": "Created",
279
"Interval": "Interval",
281
"Fingerprint": "Fingerprint",
283
"CheckerRunning": "Check Is Running",
284
"LastEnabled": "Last Enabled",
285
"ApprovalPending": "Approval Is Pending",
286
"ApprovedByDefault": "Approved By Default",
287
"ApprovalDelay": "Approval Delay",
288
"ApprovalDuration": "Approval Duration",
289
"Checker": "Checker",
290
"ExtendedTimeout": "Extended Timeout",
291
"Expires": "Expires",
292
"LastCheckerStatus": "Last Checker Status",
295
def __init__(self, clients, keywords, tablewords=None):
296
self.clients = clients
297
self.keywords = keywords
298
if tablewords is not None:
299
self.tablewords = tablewords
302
return "\n".join(self.rows())
304
if sys.version_info.major == 2:
305
__unicode__ = __str__
307
return str(self).encode(locale.getpreferredencoding())
310
format_string = self.row_formatting_string()
311
rows = [self.header_line(format_string)]
312
rows.extend(self.client_line(client, format_string)
313
for client in self.clients)
316
def row_formatting_string(self):
317
"Format string used to format table rows"
318
return " ".join("{{{key}:{width}}}".format(
319
width=max(len(self.tablewords[key]),
320
*(len(self.string_from_client(client, key))
321
for client in self.clients)),
323
for key in self.keywords)
325
def string_from_client(self, client, key):
326
return self.valuetostring(client[key], key)
329
def valuetostring(value, keyword):
330
if isinstance(value, dbus.Boolean):
331
return "Yes" if value else "No"
332
if keyword in ("Timeout", "Interval", "ApprovalDelay",
333
"ApprovalDuration", "ExtendedTimeout"):
334
return milliseconds_to_string(value)
337
def header_line(self, format_string):
338
return format_string.format(**self.tablewords)
340
def client_line(self, client, format_string):
341
return format_string.format(
342
**{key: self.string_from_client(client, key)
343
for key in self.keywords})
346
## Classes for commands.
348
# Abstract classes first
349
class Command(object):
350
"""Abstract class for commands"""
351
def run(self, clients):
352
"""Normal commands should implement run_on_one_client(), but
353
commands which want to operate on all clients at the same time
354
can override this run() method instead."""
355
for client in clients:
356
self.run_on_one_client(client)
358
class PrintCmd(Command):
359
"""Abstract class for commands printing client details"""
360
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
361
"Created", "Interval", "Host", "KeyID",
362
"Fingerprint", "CheckerRunning", "LastEnabled",
363
"ApprovalPending", "ApprovedByDefault",
364
"LastApprovalRequest", "ApprovalDelay",
365
"ApprovalDuration", "Checker", "ExtendedTimeout",
366
"Expires", "LastCheckerStatus")
367
def run(self, clients):
368
print(self.output(clients))
370
class PropertyCmd(Command):
371
"""Abstract class for Actions for setting one client property"""
372
def run_on_one_client(self, client):
373
"""Set the Client's D-Bus property"""
374
client.Set(client_interface, self.property, self.value_to_set,
375
dbus_interface=dbus.PROPERTIES_IFACE)
377
class ValueArgumentMixIn(object):
378
"""Mixin class for commands taking a value as argument"""
379
def __init__(self, value):
380
self.value_to_set = value
382
class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):
383
"""Mixin class for commands taking a value argument as
386
def value_to_set(self):
389
def value_to_set(self, value):
390
"""When setting, convert value to a datetime.timedelta"""
391
self._vts = string_to_delta(value).total_seconds() * 1000
393
# Actual (non-abstract) command classes
395
class PrintTableCmd(PrintCmd):
396
def __init__(self, verbose=False):
397
self.verbose = verbose
398
def output(self, clients):
400
keywords = self.all_keywords
402
keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
403
return str(TableOfClients(clients.values(), keywords))
405
class DumpJSONCmd(PrintCmd):
406
def output(self, clients):
407
data = {client["Name"]:
408
{key: self.dbus_boolean_to_bool(client[key])
409
for key in self.all_keywords}
410
for client in clients.values()}
411
return json.dumps(data, indent=4, separators=(',', ': '))
413
def dbus_boolean_to_bool(value):
414
if isinstance(value, dbus.Boolean):
418
class IsEnabledCmd(Command):
419
def run_on_one_client(self, client):
420
if self.is_enabled(client):
423
def is_enabled(self, client):
424
return client.Get(client_interface, "Enabled",
425
dbus_interface=dbus.PROPERTIES_IFACE)
427
class RemoveCmd(Command):
428
def __init__(self, mandos):
430
def run_on_one_client(self, client):
431
self.mandos.RemoveClient(client.__dbus_object_path__)
433
class ApproveCmd(Command):
434
def run_on_one_client(self, client):
435
client.Approve(dbus.Boolean(True),
436
dbus_interface=client_interface)
438
class DenyCmd(Command):
439
def run_on_one_client(self, client):
440
client.Approve(dbus.Boolean(False),
441
dbus_interface=client_interface)
443
class EnableCmd(PropertyCmd):
445
value_to_set = dbus.Boolean(True)
447
class DisableCmd(PropertyCmd):
449
value_to_set = dbus.Boolean(False)
451
class BumpTimeoutCmd(PropertyCmd):
452
property = "LastCheckedOK"
455
class StartCheckerCmd(PropertyCmd):
456
property = "CheckerRunning"
457
value_to_set = dbus.Boolean(True)
459
class StopCheckerCmd(PropertyCmd):
460
property = "CheckerRunning"
461
value_to_set = dbus.Boolean(False)
463
class ApproveByDefaultCmd(PropertyCmd):
464
property = "ApprovedByDefault"
465
value_to_set = dbus.Boolean(True)
467
class DenyByDefaultCmd(PropertyCmd):
468
property = "ApprovedByDefault"
469
value_to_set = dbus.Boolean(False)
471
class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):
474
class SetHostCmd(PropertyCmd, ValueArgumentMixIn):
477
class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):
480
class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
483
class SetExtendedTimeoutCmd(PropertyCmd,
484
MillisecondsValueArgumentMixIn):
485
property = "ExtendedTimeout"
487
class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
488
property = "Interval"
490
class SetApprovalDelayCmd(PropertyCmd,
491
MillisecondsValueArgumentMixIn):
492
property = "ApprovalDelay"
494
class SetApprovalDurationCmd(PropertyCmd,
495
MillisecondsValueArgumentMixIn):
496
property = "ApprovalDuration"
498
def has_actions(options):
499
return any((options.enable,
501
options.bump_timeout,
502
options.start_checker,
503
options.stop_checker,
506
options.checker is not None,
507
options.timeout is not None,
508
options.extended_timeout is not None,
509
options.interval is not None,
510
options.approved_by_default is not None,
511
options.approval_delay is not None,
512
options.approval_duration is not None,
513
options.host is not None,
514
options.secret is not None,
520
parser = argparse.ArgumentParser()
521
parser.add_argument("--version", action="version",
522
version="%(prog)s {}".format(version),
523
help="show version number and exit")
524
parser.add_argument("-a", "--all", action="store_true",
525
help="Select all clients")
526
parser.add_argument("-v", "--verbose", action="store_true",
527
help="Print all fields")
528
parser.add_argument("-j", "--dump-json", action="store_true",
529
help="Dump client data in JSON format")
530
enable_disable = parser.add_mutually_exclusive_group()
531
enable_disable.add_argument("-e", "--enable", action="store_true",
532
help="Enable client")
533
enable_disable.add_argument("-d", "--disable",
535
help="disable client")
536
parser.add_argument("-b", "--bump-timeout", action="store_true",
537
help="Bump timeout for client")
538
start_stop_checker = parser.add_mutually_exclusive_group()
539
start_stop_checker.add_argument("--start-checker",
541
help="Start checker for client")
542
start_stop_checker.add_argument("--stop-checker",
544
help="Stop checker for client")
545
parser.add_argument("-V", "--is-enabled", action="store_true",
546
help="Check if client is enabled")
547
parser.add_argument("-r", "--remove", action="store_true",
548
help="Remove client")
549
parser.add_argument("-c", "--checker",
550
help="Set checker command for client")
551
parser.add_argument("-t", "--timeout",
552
help="Set timeout for client")
553
parser.add_argument("--extended-timeout",
554
help="Set extended timeout for client")
555
parser.add_argument("-i", "--interval",
556
help="Set checker interval for client")
557
approve_deny_default = parser.add_mutually_exclusive_group()
558
approve_deny_default.add_argument(
559
"--approve-by-default", action="store_true",
560
default=None, dest="approved_by_default",
561
help="Set client to be approved by default")
562
approve_deny_default.add_argument(
563
"--deny-by-default", action="store_false",
564
dest="approved_by_default",
565
help="Set client to be denied by default")
566
parser.add_argument("--approval-delay",
567
help="Set delay before client approve/deny")
568
parser.add_argument("--approval-duration",
569
help="Set duration of one client approval")
570
parser.add_argument("-H", "--host", help="Set host for client")
571
parser.add_argument("-s", "--secret",
572
type=argparse.FileType(mode="rb"),
573
help="Set password blob (file) for client")
574
approve_deny = parser.add_mutually_exclusive_group()
575
approve_deny.add_argument(
576
"-A", "--approve", action="store_true",
577
help="Approve any current client request")
578
approve_deny.add_argument("-D", "--deny", action="store_true",
579
help="Deny any current client request")
580
parser.add_argument("--check", action="store_true",
581
help="Run self-test")
582
parser.add_argument("client", nargs="*", help="Client name")
583
options = parser.parse_args()
409
585
if has_actions(options) and not (options.client or options.all):
410
586
parser.error("Options require clients names or --all.")
529
643
SetExtendedTimeoutCmd(options.extended_timeout))
531
645
if options.interval is not None:
532
commands.append(SetIntervalCmd(options.interval))
646
command.append(SetIntervalCmd(options.interval))
648
if options.approved_by_default is not None:
649
if options.approved_by_default:
650
command.append(ApproveByDefaultCmd())
652
command.append(DenyByDefaultCmd())
534
654
if options.approval_delay is not None:
535
commands.append(SetApprovalDelayCmd(options.approval_delay))
655
command.append(SetApprovalDelayCmd(options.approval_delay))
537
657
if options.approval_duration is not None:
539
659
SetApprovalDurationCmd(options.approval_duration))
661
if options.host is not None:
662
command.append(SetHostCmd(options.host))
664
if options.secret is not None:
665
command.append(SetSecretCmd(options.secret))
668
commands.append(ApproveCmd())
671
commands.append(DenyCmd())
541
673
# If no command option has been given, show table of clients,
542
674
# optionally verbosely
544
676
commands.append(PrintTableCmd(verbose=options.verbose))
549
class Command(object):
550
"""Abstract class for commands"""
551
def run(self, clients, bus=None, mandos=None):
552
"""Normal commands should implement run_on_one_client(), but
553
commands which want to operate on all clients at the same time
554
can override this run() method instead."""
556
for clientpath, properties in clients.items():
557
log.debug("D-Bus: Connect to: (busname=%r, path=%r)",
558
dbus_busname, str(clientpath))
559
client = bus.get_object(dbus_busname, clientpath)
560
self.run_on_one_client(client, properties)
563
class IsEnabledCmd(Command):
564
def run(self, clients, bus=None, mandos=None):
565
client, properties = next(iter(clients.items()))
566
if self.is_enabled(client, properties):
678
# block stderr since dbus library prints to stderr
679
null = os.open(os.path.devnull, os.O_RDWR)
680
stderrcopy = os.dup(sys.stderr.fileno())
681
os.dup2(null, sys.stderr.fileno())
685
mandos_clients = {path: ifs_and_props[client_interface]
686
for path, ifs_and_props in
687
mandos_serv_object_manager
688
.GetManagedObjects().items()
689
if client_interface in ifs_and_props}
692
os.dup2(stderrcopy, sys.stderr.fileno())
694
except dbus.exceptions.DBusException as e:
695
log.critical("Failed to access Mandos server through D-Bus:"
569
def is_enabled(self, client, properties):
570
return properties["Enabled"]
573
class ApproveCmd(Command):
574
def run_on_one_client(self, client, properties):
575
log.debug("D-Bus: %s:%s:%s.Approve(True)", dbus_busname,
576
client.__dbus_object_path__, client_dbus_interface)
577
client.Approve(dbus.Boolean(True),
578
dbus_interface=client_dbus_interface)
581
class DenyCmd(Command):
582
def run_on_one_client(self, client, properties):
583
log.debug("D-Bus: %s:%s:%s.Approve(False)", dbus_busname,
584
client.__dbus_object_path__, client_dbus_interface)
585
client.Approve(dbus.Boolean(False),
586
dbus_interface=client_dbus_interface)
589
class RemoveCmd(Command):
590
def run_on_one_client(self, client, properties):
591
log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)", dbus_busname,
592
server_dbus_path, server_dbus_interface,
593
str(client.__dbus_object_path__))
594
self.mandos.RemoveClient(client.__dbus_object_path__)
597
class OutputCmd(Command):
598
"""Abstract class for commands outputting client details"""
599
all_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
600
"Created", "Interval", "Host", "KeyID",
601
"Fingerprint", "CheckerRunning", "LastEnabled",
602
"ApprovalPending", "ApprovedByDefault",
603
"LastApprovalRequest", "ApprovalDelay",
604
"ApprovalDuration", "Checker", "ExtendedTimeout",
605
"Expires", "LastCheckerStatus")
607
def run(self, clients, bus=None, mandos=None):
608
print(self.output(clients.values()))
610
def output(self, clients):
611
raise NotImplementedError()
614
class DumpJSONCmd(OutputCmd):
615
def output(self, clients):
616
data = {client["Name"]:
617
{key: self.dbus_boolean_to_bool(client[key])
618
for key in self.all_keywords}
619
for client in clients}
620
return json.dumps(data, indent=4, separators=(',', ': '))
623
def dbus_boolean_to_bool(value):
624
if isinstance(value, dbus.Boolean):
629
class PrintTableCmd(OutputCmd):
630
def __init__(self, verbose=False):
631
self.verbose = verbose
633
def output(self, clients):
634
default_keywords = ("Name", "Enabled", "Timeout",
636
keywords = default_keywords
638
keywords = self.all_keywords
639
return str(self.TableOfClients(clients, keywords))
641
class TableOfClients(object):
644
"Enabled": "Enabled",
645
"Timeout": "Timeout",
646
"LastCheckedOK": "Last Successful Check",
647
"LastApprovalRequest": "Last Approval Request",
648
"Created": "Created",
649
"Interval": "Interval",
651
"Fingerprint": "Fingerprint",
653
"CheckerRunning": "Check Is Running",
654
"LastEnabled": "Last Enabled",
655
"ApprovalPending": "Approval Is Pending",
656
"ApprovedByDefault": "Approved By Default",
657
"ApprovalDelay": "Approval Delay",
658
"ApprovalDuration": "Approval Duration",
659
"Checker": "Checker",
660
"ExtendedTimeout": "Extended Timeout",
661
"Expires": "Expires",
662
"LastCheckerStatus": "Last Checker Status",
665
def __init__(self, clients, keywords):
666
self.clients = clients
667
self.keywords = keywords
670
return "\n".join(self.rows())
672
if sys.version_info.major == 2:
673
__unicode__ = __str__
675
return str(self).encode(locale.getpreferredencoding())
678
format_string = self.row_formatting_string()
679
rows = [self.header_line(format_string)]
680
rows.extend(self.client_line(client, format_string)
681
for client in self.clients)
684
def row_formatting_string(self):
685
"Format string used to format table rows"
686
return " ".join("{{{key}:{width}}}".format(
687
width=max(len(self.tableheaders[key]),
688
*(len(self.string_from_client(client, key))
689
for client in self.clients)),
691
for key in self.keywords)
693
def string_from_client(self, client, key):
694
return self.valuetostring(client[key], key)
697
def valuetostring(cls, value, keyword):
698
if isinstance(value, dbus.Boolean):
699
return "Yes" if value else "No"
700
if keyword in ("Timeout", "Interval", "ApprovalDelay",
701
"ApprovalDuration", "ExtendedTimeout"):
702
return cls.milliseconds_to_string(value)
705
def header_line(self, format_string):
706
return format_string.format(**self.tableheaders)
708
def client_line(self, client, format_string):
709
return format_string.format(
710
**{key: self.string_from_client(client, key)
711
for key in self.keywords})
714
def milliseconds_to_string(ms):
715
td = datetime.timedelta(0, 0, 0, ms)
716
return ("{days}{hours:02}:{minutes:02}:{seconds:02}"
717
.format(days="{}T".format(td.days)
719
hours=td.seconds // 3600,
720
minutes=(td.seconds % 3600) // 60,
721
seconds=td.seconds % 60))
724
class PropertyCmd(Command):
725
"""Abstract class for Actions for setting one client property"""
727
def run_on_one_client(self, client, properties):
728
"""Set the Client's D-Bus property"""
729
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", dbus_busname,
730
client.__dbus_object_path__,
731
dbus.PROPERTIES_IFACE, client_dbus_interface,
732
self.propname, self.value_to_set
733
if not isinstance(self.value_to_set, dbus.Boolean)
734
else bool(self.value_to_set))
735
client.Set(client_dbus_interface, self.propname,
737
dbus_interface=dbus.PROPERTIES_IFACE)
741
raise NotImplementedError()
744
class EnableCmd(PropertyCmd):
746
value_to_set = dbus.Boolean(True)
749
class DisableCmd(PropertyCmd):
751
value_to_set = dbus.Boolean(False)
754
class BumpTimeoutCmd(PropertyCmd):
755
propname = "LastCheckedOK"
759
class StartCheckerCmd(PropertyCmd):
760
propname = "CheckerRunning"
761
value_to_set = dbus.Boolean(True)
764
class StopCheckerCmd(PropertyCmd):
765
propname = "CheckerRunning"
766
value_to_set = dbus.Boolean(False)
769
class ApproveByDefaultCmd(PropertyCmd):
770
propname = "ApprovedByDefault"
771
value_to_set = dbus.Boolean(True)
774
class DenyByDefaultCmd(PropertyCmd):
775
propname = "ApprovedByDefault"
776
value_to_set = dbus.Boolean(False)
779
class PropertyValueCmd(PropertyCmd):
780
"""Abstract class for PropertyCmd recieving a value as argument"""
781
def __init__(self, value):
782
self.value_to_set = value
785
class SetCheckerCmd(PropertyValueCmd):
789
class SetHostCmd(PropertyValueCmd):
793
class SetSecretCmd(PropertyValueCmd):
797
def value_to_set(self):
801
def value_to_set(self, value):
802
"""When setting, read data from supplied file object"""
803
self._vts = value.read()
807
class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):
808
"""Abstract class for PropertyValueCmd taking a value argument as
809
a datetime.timedelta() but should store it as milliseconds."""
812
def value_to_set(self):
816
def value_to_set(self, value):
817
"""When setting, convert value from a datetime.timedelta"""
818
self._vts = int(round(value.total_seconds() * 1000))
821
class SetTimeoutCmd(MillisecondsPropertyValueArgumentCmd):
825
class SetExtendedTimeoutCmd(MillisecondsPropertyValueArgumentCmd):
826
propname = "ExtendedTimeout"
829
class SetIntervalCmd(MillisecondsPropertyValueArgumentCmd):
830
propname = "Interval"
833
class SetApprovalDelayCmd(MillisecondsPropertyValueArgumentCmd):
834
propname = "ApprovalDelay"
837
class SetApprovalDurationCmd(MillisecondsPropertyValueArgumentCmd):
838
propname = "ApprovalDuration"
699
# Compile dict of (clients: properties) to process
702
if options.all or not options.client:
703
clients = {bus.get_object(busname, path): properties
704
for path, properties in mandos_clients.items()}
706
for name in options.client:
707
for path, client in mandos_clients.items():
708
if client["Name"] == name:
709
client_objc = bus.get_object(busname, path)
710
clients[client_objc] = client
713
log.critical("Client not found on server: %r", name)
716
# Run all commands on clients
717
for command in commands:
842
class TestCaseWithAssertLogs(unittest.TestCase):
843
"""unittest.TestCase.assertLogs only exists in Python 3.4"""
845
if not hasattr(unittest.TestCase, "assertLogs"):
846
@contextlib.contextmanager
847
def assertLogs(self, logger, level=logging.INFO):
848
capturing_handler = self.CapturingLevelHandler(level)
849
old_level = logger.level
850
old_propagate = logger.propagate
851
logger.addHandler(capturing_handler)
852
logger.setLevel(level)
853
logger.propagate = False
855
yield capturing_handler.watcher
857
logger.propagate = old_propagate
858
logger.removeHandler(capturing_handler)
859
logger.setLevel(old_level)
860
self.assertGreater(len(capturing_handler.watcher.records),
863
class CapturingLevelHandler(logging.Handler):
864
def __init__(self, level, *args, **kwargs):
865
logging.Handler.__init__(self, *args, **kwargs)
866
self.watcher = self.LoggingWatcher([], [])
867
def emit(self, record):
868
self.watcher.records.append(record)
869
self.watcher.output.append(self.format(record))
871
LoggingWatcher = collections.namedtuple("LoggingWatcher",
875
class Test_string_to_delta(TestCaseWithAssertLogs):
721
class Test_milliseconds_to_string(unittest.TestCase):
723
self.assertEqual(milliseconds_to_string(93785000),
725
def test_no_days(self):
726
self.assertEqual(milliseconds_to_string(7385000), "02:03:05")
727
def test_all_zero(self):
728
self.assertEqual(milliseconds_to_string(0), "00:00:00")
729
def test_no_fractional_seconds(self):
730
self.assertEqual(milliseconds_to_string(400), "00:00:00")
731
self.assertEqual(milliseconds_to_string(900), "00:00:00")
732
self.assertEqual(milliseconds_to_string(1900), "00:00:01")
734
class Test_string_to_delta(unittest.TestCase):
876
735
def test_handles_basic_rfc3339(self):
877
self.assertEqual(string_to_delta("PT0S"),
878
datetime.timedelta())
879
self.assertEqual(string_to_delta("P0D"),
880
datetime.timedelta())
881
self.assertEqual(string_to_delta("PT1S"),
882
datetime.timedelta(0, 1))
883
736
self.assertEqual(string_to_delta("PT2H"),
884
737
datetime.timedelta(0, 7200))
886
738
def test_falls_back_to_pre_1_6_1_with_warning(self):
887
with self.assertLogs(log, logging.WARNING):
739
# assertLogs only exists in Python 3.4
740
if hasattr(self, "assertLogs"):
741
with self.assertLogs(log, logging.WARNING):
742
value = string_to_delta("2h")
888
744
value = string_to_delta("2h")
889
745
self.assertEqual(value, datetime.timedelta(0, 7200))
892
class Test_check_option_syntax(unittest.TestCase):
894
self.parser = argparse.ArgumentParser()
895
add_command_line_options(self.parser)
897
def test_actions_requires_client_or_all(self):
898
for action, value in self.actions.items():
899
options = self.parser.parse_args()
900
setattr(options, action, value)
901
with self.assertParseError():
902
self.check_option_syntax(options)
904
# This mostly corresponds to the definition from has_actions() in
905
# check_option_syntax()
907
# The actual values set here are not that important, but we do
908
# at least stick to the correct types, even though they are
912
"bump_timeout": True,
913
"start_checker": True,
914
"stop_checker": True,
918
"timeout": datetime.timedelta(),
919
"extended_timeout": datetime.timedelta(),
920
"interval": datetime.timedelta(),
921
"approved_by_default": True,
922
"approval_delay": datetime.timedelta(),
923
"approval_duration": datetime.timedelta(),
925
"secret": io.BytesIO(b"x"),
930
@contextlib.contextmanager
931
def assertParseError(self):
932
with self.assertRaises(SystemExit) as e:
933
with self.temporarily_suppress_stderr():
935
# Exit code from argparse is guaranteed to be "2". Reference:
936
# https://docs.python.org/3/library
937
# /argparse.html#exiting-methods
938
self.assertEqual(e.exception.code, 2)
941
@contextlib.contextmanager
942
def temporarily_suppress_stderr():
943
null = os.open(os.path.devnull, os.O_RDWR)
944
stderrcopy = os.dup(sys.stderr.fileno())
945
os.dup2(null, sys.stderr.fileno())
951
os.dup2(stderrcopy, sys.stderr.fileno())
954
def check_option_syntax(self, options):
955
check_option_syntax(self.parser, options)
957
def test_actions_conflicts_with_verbose(self):
958
for action, value in self.actions.items():
959
options = self.parser.parse_args()
960
setattr(options, action, value)
961
options.verbose = True
962
with self.assertParseError():
963
self.check_option_syntax(options)
965
def test_dump_json_conflicts_with_verbose(self):
966
options = self.parser.parse_args()
967
options.dump_json = True
968
options.verbose = True
969
with self.assertParseError():
970
self.check_option_syntax(options)
972
def test_dump_json_conflicts_with_action(self):
973
for action, value in self.actions.items():
974
options = self.parser.parse_args()
975
setattr(options, action, value)
976
options.dump_json = True
977
with self.assertParseError():
978
self.check_option_syntax(options)
980
def test_all_can_not_be_alone(self):
981
options = self.parser.parse_args()
983
with self.assertParseError():
984
self.check_option_syntax(options)
986
def test_all_is_ok_with_any_action(self):
987
for action, value in self.actions.items():
988
options = self.parser.parse_args()
989
setattr(options, action, value)
991
self.check_option_syntax(options)
993
def test_is_enabled_fails_without_client(self):
994
options = self.parser.parse_args()
995
options.is_enabled = True
996
with self.assertParseError():
997
self.check_option_syntax(options)
999
def test_is_enabled_works_with_one_client(self):
1000
options = self.parser.parse_args()
1001
options.is_enabled = True
1002
options.client = ["foo"]
1003
self.check_option_syntax(options)
1005
def test_is_enabled_fails_with_two_clients(self):
1006
options = self.parser.parse_args()
1007
options.is_enabled = True
1008
options.client = ["foo", "barbar"]
1009
with self.assertParseError():
1010
self.check_option_syntax(options)
1012
def test_remove_can_only_be_combined_with_action_deny(self):
1013
for action, value in self.actions.items():
1014
if action in {"remove", "deny"}:
1016
options = self.parser.parse_args()
1017
setattr(options, action, value)
1019
options.remove = True
1020
with self.assertParseError():
1021
self.check_option_syntax(options)
1024
class Test_get_mandos_dbus_object(TestCaseWithAssertLogs):
1025
def test_calls_and_returns_get_object_on_bus(self):
1026
class MockBus(object):
1028
def get_object(mockbus_self, busname, dbus_path):
1029
# Note that "self" is still the testcase instance,
1030
# this MockBus instance is in "mockbus_self".
1031
self.assertEqual(busname, dbus_busname)
1032
self.assertEqual(dbus_path, server_dbus_path)
1033
mockbus_self.called = True
1036
mockbus = get_mandos_dbus_object(bus=MockBus())
1037
self.assertIsInstance(mockbus, MockBus)
1038
self.assertTrue(mockbus.called)
1040
def test_logs_and_exits_on_dbus_error(self):
1041
class MockBusFailing(object):
1042
def get_object(self, busname, dbus_path):
1043
raise dbus.exceptions.DBusException("Test")
1045
with self.assertLogs(log, logging.CRITICAL):
1046
with self.assertRaises(SystemExit) as e:
1047
bus = get_mandos_dbus_object(bus=MockBusFailing())
1049
if isinstance(e.exception.code, int):
1050
self.assertNotEqual(e.exception.code, 0)
1052
self.assertIsNotNone(e.exception.code)
1055
class Test_get_managed_objects(TestCaseWithAssertLogs):
1056
def test_calls_and_returns_GetManagedObjects(self):
1057
managed_objects = {"/clients/foo": { "Name": "foo"}}
1058
class MockObjectManager(object):
1059
def GetManagedObjects(self):
1060
return managed_objects
1061
retval = get_managed_objects(MockObjectManager())
1062
self.assertDictEqual(managed_objects, retval)
1064
def test_logs_and_exits_on_dbus_error(self):
1065
dbus_logger = logging.getLogger("dbus.proxies")
1067
class MockObjectManagerFailing(object):
1068
def GetManagedObjects(self):
1069
dbus_logger.error("Test")
1070
raise dbus.exceptions.DBusException("Test")
1072
class CountingHandler(logging.Handler):
1074
def emit(self, record):
1077
counting_handler = CountingHandler()
1079
dbus_logger.addHandler(counting_handler)
1082
with self.assertLogs(log, logging.CRITICAL) as watcher:
1083
with self.assertRaises(SystemExit) as e:
1084
get_managed_objects(MockObjectManagerFailing())
1086
dbus_logger.removeFilter(counting_handler)
1088
# Make sure the dbus logger was suppressed
1089
self.assertEqual(counting_handler.count, 0)
1091
# Test that the dbus_logger still works
1092
with self.assertLogs(dbus_logger, logging.ERROR):
1093
dbus_logger.error("Test")
1095
if isinstance(e.exception.code, int):
1096
self.assertNotEqual(e.exception.code, 0)
1098
self.assertIsNotNone(e.exception.code)
1101
class Test_commands_from_options(unittest.TestCase):
1103
self.parser = argparse.ArgumentParser()
1104
add_command_line_options(self.parser)
1106
def test_is_enabled(self):
1107
self.assert_command_from_args(["--is-enabled", "foo"],
1110
def assert_command_from_args(self, args, command_cls,
1112
"""Assert that parsing ARGS should result in an instance of
1113
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1114
options = self.parser.parse_args(args)
1115
check_option_syntax(self.parser, options)
1116
commands = commands_from_options(options)
1117
self.assertEqual(len(commands), 1)
1118
command = commands[0]
1119
self.assertIsInstance(command, command_cls)
1120
for key, value in cmd_attrs.items():
1121
self.assertEqual(getattr(command, key), value)
1123
def test_is_enabled_short(self):
1124
self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)
1126
def test_approve(self):
1127
self.assert_command_from_args(["--approve", "foo"],
1130
def test_approve_short(self):
1131
self.assert_command_from_args(["-A", "foo"], ApproveCmd)
1133
def test_deny(self):
1134
self.assert_command_from_args(["--deny", "foo"], DenyCmd)
1136
def test_deny_short(self):
1137
self.assert_command_from_args(["-D", "foo"], DenyCmd)
1139
def test_remove(self):
1140
self.assert_command_from_args(["--remove", "foo"],
1143
def test_deny_before_remove(self):
1144
options = self.parser.parse_args(["--deny", "--remove",
1146
check_option_syntax(self.parser, options)
1147
commands = commands_from_options(options)
1148
self.assertEqual(len(commands), 2)
1149
self.assertIsInstance(commands[0], DenyCmd)
1150
self.assertIsInstance(commands[1], RemoveCmd)
1152
def test_deny_before_remove_reversed(self):
1153
options = self.parser.parse_args(["--remove", "--deny",
1155
check_option_syntax(self.parser, options)
1156
commands = commands_from_options(options)
1157
self.assertEqual(len(commands), 2)
1158
self.assertIsInstance(commands[0], DenyCmd)
1159
self.assertIsInstance(commands[1], RemoveCmd)
1161
def test_remove_short(self):
1162
self.assert_command_from_args(["-r", "foo"], RemoveCmd)
1164
def test_dump_json(self):
1165
self.assert_command_from_args(["--dump-json"], DumpJSONCmd)
1167
def test_enable(self):
1168
self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1170
def test_enable_short(self):
1171
self.assert_command_from_args(["-e", "foo"], EnableCmd)
1173
def test_disable(self):
1174
self.assert_command_from_args(["--disable", "foo"],
1177
def test_disable_short(self):
1178
self.assert_command_from_args(["-d", "foo"], DisableCmd)
1180
def test_bump_timeout(self):
1181
self.assert_command_from_args(["--bump-timeout", "foo"],
1184
def test_bump_timeout_short(self):
1185
self.assert_command_from_args(["-b", "foo"], BumpTimeoutCmd)
1187
def test_start_checker(self):
1188
self.assert_command_from_args(["--start-checker", "foo"],
1191
def test_stop_checker(self):
1192
self.assert_command_from_args(["--stop-checker", "foo"],
1195
def test_approve_by_default(self):
1196
self.assert_command_from_args(["--approve-by-default", "foo"],
1197
ApproveByDefaultCmd)
1199
def test_deny_by_default(self):
1200
self.assert_command_from_args(["--deny-by-default", "foo"],
1203
def test_checker(self):
1204
self.assert_command_from_args(["--checker", ":", "foo"],
1205
SetCheckerCmd, value_to_set=":")
1207
def test_checker_empty(self):
1208
self.assert_command_from_args(["--checker", "", "foo"],
1209
SetCheckerCmd, value_to_set="")
1211
def test_checker_short(self):
1212
self.assert_command_from_args(["-c", ":", "foo"],
1213
SetCheckerCmd, value_to_set=":")
1215
def test_host(self):
1216
self.assert_command_from_args(["--host", "foo.example.org",
1218
value_to_set="foo.example.org")
1220
def test_host_short(self):
1221
self.assert_command_from_args(["-H", "foo.example.org",
1223
value_to_set="foo.example.org")
1225
def test_secret_devnull(self):
1226
self.assert_command_from_args(["--secret", os.path.devnull,
1227
"foo"], SetSecretCmd,
1230
def test_secret_tempfile(self):
1231
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1232
value = b"secret\0xyzzy\nbar"
1235
self.assert_command_from_args(["--secret", f.name,
1236
"foo"], SetSecretCmd,
1239
def test_secret_devnull_short(self):
1240
self.assert_command_from_args(["-s", os.path.devnull, "foo"],
1241
SetSecretCmd, value_to_set=b"")
1243
def test_secret_tempfile_short(self):
1244
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1245
value = b"secret\0xyzzy\nbar"
1248
self.assert_command_from_args(["-s", f.name, "foo"],
1252
def test_timeout(self):
1253
self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1255
value_to_set=300000)
1257
def test_timeout_short(self):
1258
self.assert_command_from_args(["-t", "PT5M", "foo"],
1260
value_to_set=300000)
1262
def test_extended_timeout(self):
1263
self.assert_command_from_args(["--extended-timeout", "PT15M",
1265
SetExtendedTimeoutCmd,
1266
value_to_set=900000)
1268
def test_interval(self):
1269
self.assert_command_from_args(["--interval", "PT2M", "foo"],
1271
value_to_set=120000)
1273
def test_interval_short(self):
1274
self.assert_command_from_args(["-i", "PT2M", "foo"],
1276
value_to_set=120000)
1278
def test_approval_delay(self):
1279
self.assert_command_from_args(["--approval-delay", "PT30S",
1280
"foo"], SetApprovalDelayCmd,
1283
def test_approval_duration(self):
1284
self.assert_command_from_args(["--approval-duration", "PT1S",
1285
"foo"], SetApprovalDurationCmd,
1288
def test_print_table(self):
1289
self.assert_command_from_args([], PrintTableCmd,
1292
def test_print_table_verbose(self):
1293
self.assert_command_from_args(["--verbose"], PrintTableCmd,
1296
def test_print_table_verbose_short(self):
1297
self.assert_command_from_args(["-v"], PrintTableCmd,
1301
class TestCmd(unittest.TestCase):
1302
"""Abstract class for tests of command classes"""
1306
class MockClient(object):
1307
def __init__(self, name, **attributes):
1308
self.__dbus_object_path__ = "/clients/{}".format(name)
1309
self.attributes = attributes
1310
self.attributes["Name"] = name
1312
def Set(self, interface, propname, value, dbus_interface):
1313
testcase.assertEqual(interface, client_dbus_interface)
1314
testcase.assertEqual(dbus_interface,
1315
dbus.PROPERTIES_IFACE)
1316
self.attributes[propname] = value
1317
def Get(self, interface, propname, dbus_interface):
1318
testcase.assertEqual(interface, client_dbus_interface)
1319
testcase.assertEqual(dbus_interface,
1320
dbus.PROPERTIES_IFACE)
1321
return self.attributes[propname]
1322
def Approve(self, approve, dbus_interface):
1323
testcase.assertEqual(dbus_interface,
1324
client_dbus_interface)
1325
self.calls.append(("Approve", (approve,
1327
self.client = MockClient(
1329
KeyID=("92ed150794387c03ce684574b1139a65"
1330
"94a34f895daaaf09fd8ea90a27cddb12"),
1332
Host="foo.example.org",
1333
Enabled=dbus.Boolean(True),
1335
LastCheckedOK="2019-02-03T00:00:00",
1336
Created="2019-01-02T00:00:00",
1338
Fingerprint=("778827225BA7DE539C5A"
1339
"7CFA59CFF7CDBD9A5920"),
1340
CheckerRunning=dbus.Boolean(False),
1341
LastEnabled="2019-01-03T00:00:00",
1342
ApprovalPending=dbus.Boolean(False),
1343
ApprovedByDefault=dbus.Boolean(True),
1344
LastApprovalRequest="",
1346
ApprovalDuration=1000,
1347
Checker="fping -q -- %(host)s",
1348
ExtendedTimeout=900000,
1349
Expires="2019-02-04T00:00:00",
1350
LastCheckerStatus=0)
1351
self.other_client = MockClient(
1353
KeyID=("0558568eedd67d622f5c83b35a115f79"
1354
"6ab612cff5ad227247e46c2b020f441c"),
1355
Secret=b"secretbar",
1357
Enabled=dbus.Boolean(True),
1359
LastCheckedOK="2019-02-04T00:00:00",
1360
Created="2019-01-03T00:00:00",
1362
Fingerprint=("3E393AEAEFB84C7E89E2"
1363
"F547B3A107558FCA3A27"),
1364
CheckerRunning=dbus.Boolean(True),
1365
LastEnabled="2019-01-04T00:00:00",
1366
ApprovalPending=dbus.Boolean(False),
1367
ApprovedByDefault=dbus.Boolean(False),
1368
LastApprovalRequest="2019-01-03T00:00:00",
1369
ApprovalDelay=30000,
1370
ApprovalDuration=93785000,
1372
ExtendedTimeout=900000,
1373
Expires="2019-02-05T00:00:00",
1374
LastCheckerStatus=-2)
1375
self.clients = collections.OrderedDict(
1377
("/clients/foo", self.client.attributes),
1378
("/clients/barbar", self.other_client.attributes),
1380
self.one_client = {"/clients/foo": self.client.attributes}
1386
def get_object(client_bus_name, path):
1387
self.assertEqual(client_bus_name, dbus_busname)
1389
# Note: "self" here is the TestCmd instance, not
1390
# the Bus instance, since this is a static method!
1391
"/clients/foo": self.client,
1392
"/clients/barbar": self.other_client,
1397
class TestIsEnabledCmd(TestCmd):
1398
def test_is_enabled(self):
1399
self.assertTrue(all(IsEnabledCmd().is_enabled(client,
1401
for client, properties
1402
in self.clients.items()))
1404
def test_is_enabled_run_exits_successfully(self):
1405
with self.assertRaises(SystemExit) as e:
1406
IsEnabledCmd().run(self.one_client)
1407
if e.exception.code is not None:
1408
self.assertEqual(e.exception.code, 0)
1410
self.assertIsNone(e.exception.code)
1412
def test_is_enabled_run_exits_with_failure(self):
1413
self.client.attributes["Enabled"] = dbus.Boolean(False)
1414
with self.assertRaises(SystemExit) as e:
1415
IsEnabledCmd().run(self.one_client)
1416
if isinstance(e.exception.code, int):
1417
self.assertNotEqual(e.exception.code, 0)
1419
self.assertIsNotNone(e.exception.code)
1422
class TestApproveCmd(TestCmd):
1423
def test_approve(self):
1424
ApproveCmd().run(self.clients, self.bus)
1425
for clientpath in self.clients:
1426
client = self.bus.get_object(dbus_busname, clientpath)
1427
self.assertIn(("Approve", (True, client_dbus_interface)),
1431
class TestDenyCmd(TestCmd):
1432
def test_deny(self):
1433
DenyCmd().run(self.clients, self.bus)
1434
for clientpath in self.clients:
1435
client = self.bus.get_object(dbus_busname, clientpath)
1436
self.assertIn(("Approve", (False, client_dbus_interface)),
1440
class TestRemoveCmd(TestCmd):
1441
def test_remove(self):
1442
class MockMandos(object):
1445
def RemoveClient(self, dbus_path):
1446
self.calls.append(("RemoveClient", (dbus_path,)))
1447
mandos = MockMandos()
1448
super(TestRemoveCmd, self).setUp()
1449
RemoveCmd().run(self.clients, self.bus, mandos)
1450
self.assertEqual(len(mandos.calls), 2)
1451
for clientpath in self.clients:
1452
self.assertIn(("RemoveClient", (clientpath,)),
1456
class TestDumpJSONCmd(TestCmd):
1458
self.expected_json = {
1461
"KeyID": ("92ed150794387c03ce684574b1139a65"
1462
"94a34f895daaaf09fd8ea90a27cddb12"),
1463
"Host": "foo.example.org",
1466
"LastCheckedOK": "2019-02-03T00:00:00",
1467
"Created": "2019-01-02T00:00:00",
1469
"Fingerprint": ("778827225BA7DE539C5A"
1470
"7CFA59CFF7CDBD9A5920"),
1471
"CheckerRunning": False,
1472
"LastEnabled": "2019-01-03T00:00:00",
1473
"ApprovalPending": False,
1474
"ApprovedByDefault": True,
1475
"LastApprovalRequest": "",
1477
"ApprovalDuration": 1000,
1478
"Checker": "fping -q -- %(host)s",
1479
"ExtendedTimeout": 900000,
1480
"Expires": "2019-02-04T00:00:00",
1481
"LastCheckerStatus": 0,
1485
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
1486
"6ab612cff5ad227247e46c2b020f441c"),
1487
"Host": "192.0.2.3",
1490
"LastCheckedOK": "2019-02-04T00:00:00",
1491
"Created": "2019-01-03T00:00:00",
1493
"Fingerprint": ("3E393AEAEFB84C7E89E2"
1494
"F547B3A107558FCA3A27"),
1495
"CheckerRunning": True,
1496
"LastEnabled": "2019-01-04T00:00:00",
1497
"ApprovalPending": False,
1498
"ApprovedByDefault": False,
1499
"LastApprovalRequest": "2019-01-03T00:00:00",
1500
"ApprovalDelay": 30000,
1501
"ApprovalDuration": 93785000,
1503
"ExtendedTimeout": 900000,
1504
"Expires": "2019-02-05T00:00:00",
1505
"LastCheckerStatus": -2,
747
class Test_TableOfClients(unittest.TestCase):
753
"Bool": "A D-BUS Boolean",
754
"NonDbusBoolean": "A Non-D-BUS Boolean",
755
"Integer": "An Integer",
756
"Timeout": "Timedelta 1",
757
"Interval": "Timedelta 2",
758
"ApprovalDelay": "Timedelta 3",
759
"ApprovalDuration": "Timedelta 4",
760
"ExtendedTimeout": "Timedelta 5",
761
"String": "A String",
1508
return super(TestDumpJSONCmd, self).setUp()
1510
def test_normal(self):
1511
output = DumpJSONCmd().output(self.clients.values())
1512
json_data = json.loads(output)
1513
self.assertDictEqual(json_data, self.expected_json)
1515
def test_one_client(self):
1516
output = DumpJSONCmd().output(self.one_client.values())
1517
json_data = json.loads(output)
1518
expected_json = {"foo": self.expected_json["foo"]}
1519
self.assertDictEqual(json_data, expected_json)
1522
class TestPrintTableCmd(TestCmd):
1523
def test_normal(self):
1524
output = PrintTableCmd().output(self.clients.values())
1525
expected_output = "\n".join((
1526
"Name Enabled Timeout Last Successful Check",
1527
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1528
"barbar Yes 00:05:00 2019-02-04T00:00:00 ",
1530
self.assertEqual(output, expected_output)
1532
def test_verbose(self):
1533
output = PrintTableCmd(verbose=True).output(
1534
self.clients.values())
1549
"Last Successful Check ",
1550
"2019-02-03T00:00:00 ",
1551
"2019-02-04T00:00:00 ",
1554
"2019-01-02T00:00:00 ",
1555
"2019-01-03T00:00:00 ",
1567
("92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8"
1569
("0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e"
1573
"778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 ",
1574
"3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 ",
1576
"Check Is Running ",
1581
"2019-01-03T00:00:00 ",
1582
"2019-01-04T00:00:00 ",
1584
"Approval Is Pending ",
1588
"Approved By Default ",
1592
"Last Approval Request ",
1594
"2019-01-03T00:00:00 ",
1600
"Approval Duration ",
1605
"fping -q -- %(host)s ",
1608
"Extended Timeout ",
1613
"2019-02-04T00:00:00 ",
1614
"2019-02-05T00:00:00 ",
1616
"Last Checker Status",
1621
num_lines = max(len(rows) for rows in columns)
1622
expected_output = "\n".join("".join(rows[line]
1623
for rows in columns)
1624
for line in range(num_lines))
1625
self.assertEqual(output, expected_output)
1627
def test_one_client(self):
1628
output = PrintTableCmd().output(self.one_client.values())
1629
expected_output = "\n".join((
1630
"Name Enabled Timeout Last Successful Check",
1631
"foo Yes 00:05:00 2019-02-03T00:00:00 ",
1633
self.assertEqual(output, expected_output)
1636
class TestPropertyCmd(TestCmd):
1637
"""Abstract class for tests of PropertyCmd classes"""
1639
if not hasattr(self, "command"):
1641
values_to_get = getattr(self, "values_to_get",
1643
for value_to_set, value_to_get in zip(self.values_to_set,
1645
for clientpath in self.clients:
1646
client = self.bus.get_object(dbus_busname, clientpath)
1647
old_value = client.attributes[self.propname]
1648
self.assertNotIsInstance(old_value, self.Unique)
1649
client.attributes[self.propname] = self.Unique()
1650
self.run_command(value_to_set, self.clients)
1651
for clientpath in self.clients:
1652
client = self.bus.get_object(dbus_busname, clientpath)
1653
value = client.attributes[self.propname]
1654
self.assertNotIsInstance(value, self.Unique)
1655
self.assertEqual(value, value_to_get)
1657
class Unique(object):
1658
"""Class for objects which exist only to be unique objects,
1659
since unittest.mock.sentinel only exists in Python 3.3"""
1661
def run_command(self, value, clients):
1662
self.command().run(clients, self.bus)
1665
class TestEnableCmd(TestPropertyCmd):
1667
propname = "Enabled"
1668
values_to_set = [dbus.Boolean(True)]
1671
class TestDisableCmd(TestPropertyCmd):
1672
command = DisableCmd
1673
propname = "Enabled"
1674
values_to_set = [dbus.Boolean(False)]
1677
class TestBumpTimeoutCmd(TestPropertyCmd):
1678
command = BumpTimeoutCmd
1679
propname = "LastCheckedOK"
1680
values_to_set = [""]
1683
class TestStartCheckerCmd(TestPropertyCmd):
1684
command = StartCheckerCmd
1685
propname = "CheckerRunning"
1686
values_to_set = [dbus.Boolean(True)]
1689
class TestStopCheckerCmd(TestPropertyCmd):
1690
command = StopCheckerCmd
1691
propname = "CheckerRunning"
1692
values_to_set = [dbus.Boolean(False)]
1695
class TestApproveByDefaultCmd(TestPropertyCmd):
1696
command = ApproveByDefaultCmd
1697
propname = "ApprovedByDefault"
1698
values_to_set = [dbus.Boolean(True)]
1701
class TestDenyByDefaultCmd(TestPropertyCmd):
1702
command = DenyByDefaultCmd
1703
propname = "ApprovedByDefault"
1704
values_to_set = [dbus.Boolean(False)]
1707
class TestPropertyValueCmd(TestPropertyCmd):
1708
"""Abstract class for tests of PropertyValueCmd classes"""
1711
if type(self) is TestPropertyValueCmd:
1713
return super(TestPropertyValueCmd, self).runTest()
1715
def run_command(self, value, clients):
1716
self.command(value).run(clients, self.bus)
1719
class TestSetCheckerCmd(TestPropertyValueCmd):
1720
command = SetCheckerCmd
1721
propname = "Checker"
1722
values_to_set = ["", ":", "fping -q -- %s"]
1725
class TestSetHostCmd(TestPropertyValueCmd):
1726
command = SetHostCmd
1728
values_to_set = ["192.0.2.3", "foo.example.org"]
1731
class TestSetSecretCmd(TestPropertyValueCmd):
1732
command = SetSecretCmd
1734
values_to_set = [io.BytesIO(b""),
1735
io.BytesIO(b"secret\0xyzzy\nbar")]
1736
values_to_get = [b"", b"secret\0xyzzy\nbar"]
1739
class TestSetTimeoutCmd(TestPropertyValueCmd):
1740
command = SetTimeoutCmd
1741
propname = "Timeout"
1742
values_to_set = [datetime.timedelta(),
1743
datetime.timedelta(minutes=5),
1744
datetime.timedelta(seconds=1),
1745
datetime.timedelta(weeks=1),
1746
datetime.timedelta(weeks=52)]
1747
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1750
class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):
1751
command = SetExtendedTimeoutCmd
1752
propname = "ExtendedTimeout"
1753
values_to_set = [datetime.timedelta(),
1754
datetime.timedelta(minutes=5),
1755
datetime.timedelta(seconds=1),
1756
datetime.timedelta(weeks=1),
1757
datetime.timedelta(weeks=52)]
1758
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1761
class TestSetIntervalCmd(TestPropertyValueCmd):
1762
command = SetIntervalCmd
1763
propname = "Interval"
1764
values_to_set = [datetime.timedelta(),
1765
datetime.timedelta(minutes=5),
1766
datetime.timedelta(seconds=1),
1767
datetime.timedelta(weeks=1),
1768
datetime.timedelta(weeks=52)]
1769
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1772
class TestSetApprovalDelayCmd(TestPropertyValueCmd):
1773
command = SetApprovalDelayCmd
1774
propname = "ApprovalDelay"
1775
values_to_set = [datetime.timedelta(),
1776
datetime.timedelta(minutes=5),
1777
datetime.timedelta(seconds=1),
1778
datetime.timedelta(weeks=1),
1779
datetime.timedelta(weeks=52)]
1780
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1783
class TestSetApprovalDurationCmd(TestPropertyValueCmd):
1784
command = SetApprovalDurationCmd
1785
propname = "ApprovalDuration"
1786
values_to_set = [datetime.timedelta(),
1787
datetime.timedelta(minutes=5),
1788
datetime.timedelta(seconds=1),
1789
datetime.timedelta(weeks=1),
1790
datetime.timedelta(weeks=52)]
1791
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
763
self.keywords = ["Attr1", "AttrTwo"]
769
"Bool": dbus.Boolean(False),
770
"NonDbusBoolean": False,
774
"ApprovalDelay": 2000,
775
"ApprovalDuration": 3000,
776
"ExtendedTimeout": 4000,
783
"Bool": dbus.Boolean(True),
784
"NonDbusBoolean": True,
787
"Interval": 93786000,
788
"ApprovalDelay": 93787000,
789
"ApprovalDuration": 93788000,
790
"ExtendedTimeout": 93789000,
791
"String": "A huge string which will not fit," * 10,
794
def test_short_header(self):
795
text = str(TableOfClients(self.clients, self.keywords,
802
self.assertEqual(text, expected_text)
803
def test_booleans(self):
804
keywords = ["Bool", "NonDbusBoolean"]
805
text = str(TableOfClients(self.clients, keywords,
808
A D-BUS Boolean A Non-D-BUS Boolean
812
self.assertEqual(text, expected_text)
813
def test_milliseconds_detection(self):
814
keywords = ["Integer", "Timeout", "Interval", "ApprovalDelay",
815
"ApprovalDuration", "ExtendedTimeout"]
816
text = str(TableOfClients(self.clients, keywords,
819
An Integer Timedelta 1 Timedelta 2 Timedelta 3 Timedelta 4 Timedelta 5
820
0 00:00:00 00:00:01 00:00:02 00:00:03 00:00:04
821
1 1T02:03:05 1T02:03:06 1T02:03:07 1T02:03:08 1T02:03:09
823
self.assertEqual(text, expected_text)
824
def test_empty_and_long_string_values(self):
825
keywords = ["String"]
826
text = str(TableOfClients(self.clients, keywords,
831
A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,
833
self.assertEqual(text, expected_text)