/mandos/trunk

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

  • Committer: Teddy Hogeborn
  • Date: 2019-03-06 22:21:51 UTC
  • Revision ID: teddy@recompile.se-20190306222151-cp303jcuqhast6nu
mandos-ctl: Refactor tests and add more tests

* mandos-ctl (TestOptions): Rename to "Test_command_from_options".
  (Test_command_from_options.test_default_is_show_table): Rename to
                         "Test_command_from_options.test_print_table".
  (Test_command_from_options.test_show_table_verbose): Rename to
                 "Test_command_from_options.test_print_table_verbose".
  (Test_command_from_options.test_bump_timeout): New.
  (Test_command_from_options.test_start_checker): - '' -
  (Test_command_from_options.test_stop_checker): - '' -
  (Test_command_from_options.test_remove): - '' -
  (Test_command_from_options.test_checker): - '' -
  (Test_command_from_options.test_timeout): - '' -
  (Test_command_from_options.test_extended_timeout): - '' -
  (Test_command_from_options.test_interval): - '' -
  (Test_command_from_options.test_approve_by_default): - '' -
  (Test_command_from_options.test_deny_by_default): - '' -
  (Test_command_from_options.test_approval_delay): - '' -
  (Test_command_from_options.test_approval_duration): - '' -
  (Test_command_from_options.test_host): - '' -
  (Test_command_from_options.test_secret_devnull): - '' -
  (Test_command_from_options.test_secret_tempfile): - '' -
  (Test_command_from_options.test_approve): - '' -
  (Test_command_from_options.test_deny): - '' -
  (Test_command_from_options.test_dump_json): - '' -
  (Test_command_from_options.test_is_enabled): - '' -

Show diffs side-by-side

added added

removed removed

Lines of Context:
mandos-ctl
44
44
import logging
45
45
import io
46
46
import tempfile
47
 
import contextlib
48
47
 
49
48
import dbus
50
49
 
294
293
                    "ApprovalDuration", "Checker", "ExtendedTimeout",
295
294
                    "Expires", "LastCheckerStatus")
296
295
    def run(self, mandos, clients):
297
 
        print(self.output(clients.values()))
 
296
        print(self.output(clients))
298
297
 
299
298
class PropertyCmd(Command):
300
299
    """Abstract class for Actions for setting one client property"""
301
300
    def run_on_one_client(self, client, properties):
302
301
        """Set the Client's D-Bus property"""
303
 
        log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
304
 
                  client.__dbus_object_path__,
305
 
                  dbus.PROPERTIES_IFACE, client_interface,
306
 
                  self.propname, self.value_to_set
307
 
                  if not isinstance(self.value_to_set, dbus.Boolean)
308
 
                  else bool(self.value_to_set))
309
 
        client.Set(client_interface, self.propname, self.value_to_set,
 
302
        client.Set(client_interface, self.property, self.value_to_set,
310
303
                   dbus_interface=dbus.PROPERTIES_IFACE)
311
304
 
312
305
class ValueArgumentMixIn(object):
323
316
    @value_to_set.setter
324
317
    def value_to_set(self, value):
325
318
        """When setting, convert value to a datetime.timedelta"""
326
 
        self._vts = int(round(value.total_seconds() * 1000))
 
319
        self._vts = string_to_delta(value).total_seconds() * 1000
327
320
 
328
321
# Actual (non-abstract) command classes
329
322
 
336
329
        keywords = default_keywords
337
330
        if self.verbose:
338
331
            keywords = self.all_keywords
339
 
        return str(self.TableOfClients(clients, keywords))
 
332
        return str(self.TableOfClients(clients.values(), keywords))
340
333
 
341
334
    class TableOfClients(object):
342
335
        tableheaders = {
433
426
            sys.exit(0)
434
427
        sys.exit(1)
435
428
    def is_enabled(self, client, properties):
436
 
        log.debug("D-Bus: %s:%s:%s.Get(%r, %r)", busname,
437
 
                  client.__dbus_object_path__,
438
 
                  dbus.PROPERTIES_IFACE, client_interface,
439
 
                  "Enabled")
440
 
        return bool(client.Get(client_interface, "Enabled",
441
 
                               dbus_interface=dbus.PROPERTIES_IFACE))
 
429
        return bool(properties["Enabled"])
442
430
 
443
431
class RemoveCmd(Command):
444
432
    def run_on_one_client(self, client, properties):
445
 
        log.debug("D-Bus: %s:%s:%s.RemoveClient(%r)", busname,
446
 
                  server_path, server_interface,
447
 
                  str(client.__dbus_object_path__))
448
433
        self.mandos.RemoveClient(client.__dbus_object_path__)
449
434
 
450
435
class ApproveCmd(Command):
451
436
    def run_on_one_client(self, client, properties):
452
 
        log.debug("D-Bus: %s:%s.Approve(True)",
453
 
                  client.__dbus_object_path__, client_interface)
454
437
        client.Approve(dbus.Boolean(True),
455
438
                       dbus_interface=client_interface)
456
439
 
457
440
class DenyCmd(Command):
458
441
    def run_on_one_client(self, client, properties):
459
 
        log.debug("D-Bus: %s:%s.Approve(False)",
460
 
                  client.__dbus_object_path__, client_interface)
461
442
        client.Approve(dbus.Boolean(False),
462
443
                       dbus_interface=client_interface)
463
444
 
464
445
class EnableCmd(PropertyCmd):
465
 
    propname = "Enabled"
 
446
    property = "Enabled"
466
447
    value_to_set = dbus.Boolean(True)
467
448
 
468
449
class DisableCmd(PropertyCmd):
469
 
    propname = "Enabled"
 
450
    property = "Enabled"
470
451
    value_to_set = dbus.Boolean(False)
471
452
 
472
453
class BumpTimeoutCmd(PropertyCmd):
473
 
    propname = "LastCheckedOK"
 
454
    property = "LastCheckedOK"
474
455
    value_to_set = ""
475
456
 
476
457
class StartCheckerCmd(PropertyCmd):
477
 
    propname = "CheckerRunning"
 
458
    property = "CheckerRunning"
478
459
    value_to_set = dbus.Boolean(True)
479
460
 
480
461
class StopCheckerCmd(PropertyCmd):
481
 
    propname = "CheckerRunning"
 
462
    property = "CheckerRunning"
482
463
    value_to_set = dbus.Boolean(False)
483
464
 
484
465
class ApproveByDefaultCmd(PropertyCmd):
485
 
    propname = "ApprovedByDefault"
 
466
    property = "ApprovedByDefault"
486
467
    value_to_set = dbus.Boolean(True)
487
468
 
488
469
class DenyByDefaultCmd(PropertyCmd):
489
 
    propname = "ApprovedByDefault"
 
470
    property = "ApprovedByDefault"
490
471
    value_to_set = dbus.Boolean(False)
491
472
 
492
473
class SetCheckerCmd(PropertyCmd, ValueArgumentMixIn):
493
 
    propname = "Checker"
 
474
    property = "Checker"
494
475
 
495
476
class SetHostCmd(PropertyCmd, ValueArgumentMixIn):
496
 
    propname = "Host"
 
477
    property = "Host"
497
478
 
498
479
class SetSecretCmd(PropertyCmd, ValueArgumentMixIn):
499
 
    propname = "Secret"
500
480
    @property
501
481
    def value_to_set(self):
502
482
        return self._vts
505
485
        """When setting, read data from supplied file object"""
506
486
        self._vts = value.read()
507
487
        value.close()
 
488
    property = "Secret"
508
489
 
509
490
class SetTimeoutCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
510
 
    propname = "Timeout"
 
491
    property = "Timeout"
511
492
 
512
493
class SetExtendedTimeoutCmd(PropertyCmd,
513
494
                            MillisecondsValueArgumentMixIn):
514
 
    propname = "ExtendedTimeout"
 
495
    property = "ExtendedTimeout"
515
496
 
516
497
class SetIntervalCmd(PropertyCmd, MillisecondsValueArgumentMixIn):
517
 
    propname = "Interval"
 
498
    property = "Interval"
518
499
 
519
500
class SetApprovalDelayCmd(PropertyCmd,
520
501
                          MillisecondsValueArgumentMixIn):
521
 
    propname = "ApprovalDelay"
 
502
    property = "ApprovalDelay"
522
503
 
523
504
class SetApprovalDurationCmd(PropertyCmd,
524
505
                             MillisecondsValueArgumentMixIn):
525
 
    propname = "ApprovalDuration"
 
506
    property = "ApprovalDuration"
 
507
 
 
508
def has_actions(options):
 
509
    return any((options.enable,
 
510
                options.disable,
 
511
                options.bump_timeout,
 
512
                options.start_checker,
 
513
                options.stop_checker,
 
514
                options.is_enabled,
 
515
                options.remove,
 
516
                options.checker is not None,
 
517
                options.timeout is not None,
 
518
                options.extended_timeout is not None,
 
519
                options.interval is not None,
 
520
                options.approved_by_default is not None,
 
521
                options.approval_delay is not None,
 
522
                options.approval_duration is not None,
 
523
                options.host is not None,
 
524
                options.secret is not None,
 
525
                options.approve,
 
526
                options.deny))
526
527
 
527
528
def add_command_line_options(parser):
528
529
    parser.add_argument("--version", action="version",
555
556
                        help="Remove client")
556
557
    parser.add_argument("-c", "--checker",
557
558
                        help="Set checker command for client")
558
 
    parser.add_argument("-t", "--timeout", type=string_to_delta,
 
559
    parser.add_argument("-t", "--timeout",
559
560
                        help="Set timeout for client")
560
 
    parser.add_argument("--extended-timeout", type=string_to_delta,
 
561
    parser.add_argument("--extended-timeout",
561
562
                        help="Set extended timeout for client")
562
 
    parser.add_argument("-i", "--interval", type=string_to_delta,
 
563
    parser.add_argument("-i", "--interval",
563
564
                        help="Set checker interval for client")
564
565
    approve_deny_default = parser.add_mutually_exclusive_group()
565
566
    approve_deny_default.add_argument(
570
571
        "--deny-by-default", action="store_false",
571
572
        dest="approved_by_default",
572
573
        help="Set client to be denied by default")
573
 
    parser.add_argument("--approval-delay", type=string_to_delta,
 
574
    parser.add_argument("--approval-delay",
574
575
                        help="Set delay before client approve/deny")
575
 
    parser.add_argument("--approval-duration", type=string_to_delta,
 
576
    parser.add_argument("--approval-duration",
576
577
                        help="Set duration of one client approval")
577
578
    parser.add_argument("-H", "--host", help="Set host for client")
578
579
    parser.add_argument("-s", "--secret",
584
585
        help="Approve any current client request")
585
586
    approve_deny.add_argument("-D", "--deny", action="store_true",
586
587
                              help="Deny any current client request")
587
 
    parser.add_argument("--debug", action="store_true",
588
 
                        help="Debug mode (show D-Bus commands)")
589
588
    parser.add_argument("--check", action="store_true",
590
589
                        help="Run self-test")
591
590
    parser.add_argument("client", nargs="*", help="Client name")
616
615
    if options.is_enabled:
617
616
        commands.append(IsEnabledCmd())
618
617
 
 
618
    if options.remove:
 
619
        commands.append(RemoveCmd())
 
620
 
619
621
    if options.checker is not None:
620
622
        commands.append(SetCheckerCmd(options.checker))
621
623
 
654
656
    if options.deny:
655
657
        commands.append(DenyCmd())
656
658
 
657
 
    if options.remove:
658
 
        commands.append(RemoveCmd())
659
 
 
660
659
    # If no command option has been given, show table of clients,
661
660
    # optionally verbosely
662
661
    if not commands:
665
664
    return commands
666
665
 
667
666
 
668
 
def check_option_syntax(parser, options):
669
 
    """Apply additional restrictions on options, not expressible in
670
 
argparse"""
671
 
 
672
 
    def has_actions(options):
673
 
        return any((options.enable,
674
 
                    options.disable,
675
 
                    options.bump_timeout,
676
 
                    options.start_checker,
677
 
                    options.stop_checker,
678
 
                    options.is_enabled,
679
 
                    options.remove,
680
 
                    options.checker is not None,
681
 
                    options.timeout is not None,
682
 
                    options.extended_timeout is not None,
683
 
                    options.interval is not None,
684
 
                    options.approved_by_default is not None,
685
 
                    options.approval_delay is not None,
686
 
                    options.approval_duration is not None,
687
 
                    options.host is not None,
688
 
                    options.secret is not None,
689
 
                    options.approve,
690
 
                    options.deny))
 
667
def main():
 
668
    parser = argparse.ArgumentParser()
 
669
 
 
670
    add_command_line_options(parser)
 
671
 
 
672
    options = parser.parse_args()
691
673
 
692
674
    if has_actions(options) and not (options.client or options.all):
693
675
        parser.error("Options require clients names or --all.")
700
682
        parser.error("--all requires an action.")
701
683
    if options.is_enabled and len(options.client) > 1:
702
684
        parser.error("--is-enabled requires exactly one client")
703
 
    if options.remove:
704
 
        options.remove = False
705
 
        if has_actions(options) and not options.deny:
706
 
            parser.error("--remove can only be combined with --deny")
707
 
        options.remove = True
708
 
 
709
 
 
710
 
def main():
711
 
    parser = argparse.ArgumentParser()
712
 
 
713
 
    add_command_line_options(parser)
714
 
 
715
 
    options = parser.parse_args()
716
 
 
717
 
    check_option_syntax(parser, options)
718
685
 
719
686
    clientnames = options.client
720
687
 
721
 
    if options.debug:
722
 
        log.setLevel(logging.DEBUG)
723
 
 
724
688
    try:
725
689
        bus = dbus.SystemBus()
726
 
        log.debug("D-Bus: Connect to: (name=%r, path=%r)", busname,
727
 
                  server_path)
728
690
        mandos_dbus_objc = bus.get_object(busname, server_path)
729
691
    except dbus.exceptions.DBusException:
730
692
        log.critical("Could not connect to Mandos server")
743
705
    dbus_filter = NullFilter()
744
706
    try:
745
707
        dbus_logger.addFilter(dbus_filter)
746
 
        log.debug("D-Bus: %s:%s:%s.GetManagedObjects()", busname,
747
 
                  server_path, dbus.OBJECT_MANAGER_IFACE)
748
708
        mandos_clients = {path: ifs_and_props[client_interface]
749
709
                          for path, ifs_and_props in
750
710
                          mandos_serv_object_manager
762
722
    clients = {}
763
723
 
764
724
    if not clientnames:
765
 
        clients = {(log.debug("D-Bus: Connect to: (name=%r, path=%r)",
766
 
                              busname, str(path)) and False) or
767
 
                   bus.get_object(busname, path): properties
 
725
        clients = {bus.get_object(busname, path): properties
768
726
                   for path, properties in mandos_clients.items()}
769
727
    else:
770
728
        for name in clientnames:
771
729
            for path, client in mandos_clients.items():
772
730
                if client["Name"] == name:
773
 
                    log.debug("D-Bus: Connect to: (name=%r, path=%r)",
774
 
                              busname, str(path))
775
731
                    client_objc = bus.get_object(busname, path)
776
732
                    clients[client_objc] = client
777
733
                    break
841
797
                self.attributes = attributes
842
798
                self.attributes["Name"] = name
843
799
                self.calls = []
844
 
            def Set(self, interface, propname, value, dbus_interface):
845
 
                testcase.assertEqual(interface, client_interface)
846
 
                testcase.assertEqual(dbus_interface,
847
 
                                     dbus.PROPERTIES_IFACE)
848
 
                self.attributes[propname] = value
849
 
            def Get(self, interface, propname, dbus_interface):
850
 
                testcase.assertEqual(interface, client_interface)
851
 
                testcase.assertEqual(dbus_interface,
852
 
                                     dbus.PROPERTIES_IFACE)
853
 
                return self.attributes[propname]
 
800
            def Set(self, interface, property, value, dbus_interface):
 
801
                testcase.assertEqual(interface, client_interface)
 
802
                testcase.assertEqual(dbus_interface,
 
803
                                     dbus.PROPERTIES_IFACE)
 
804
                self.attributes[property] = value
 
805
            def Get(self, interface, property, dbus_interface):
 
806
                testcase.assertEqual(interface, client_interface)
 
807
                testcase.assertEqual(dbus_interface,
 
808
                                     dbus.PROPERTIES_IFACE)
 
809
                return self.attributes[property]
854
810
            def Approve(self, approve, dbus_interface):
855
811
                testcase.assertEqual(dbus_interface, client_interface)
856
812
                self.calls.append(("Approve", (approve,
912
868
 
913
869
class TestPrintTableCmd(TestCmd):
914
870
    def test_normal(self):
915
 
        output = PrintTableCmd().output(self.clients.values())
 
871
        output = PrintTableCmd().output(self.clients)
916
872
        expected_output = """
917
873
Name   Enabled Timeout  Last Successful Check
918
874
foo    Yes     00:05:00 2019-02-03T00:00:00  
920
876
"""[1:-1]
921
877
        self.assertEqual(output, expected_output)
922
878
    def test_verbose(self):
923
 
        output = PrintTableCmd(verbose=True).output(
924
 
            self.clients.values())
 
879
        output = PrintTableCmd(verbose=True).output(self.clients)
925
880
        expected_output = """
926
881
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
927
882
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                  
929
884
"""[1:-1]
930
885
        self.assertEqual(output, expected_output)
931
886
    def test_one_client(self):
932
 
        output = PrintTableCmd().output(self.one_client.values())
 
887
        output = PrintTableCmd().output(self.one_client)
933
888
        expected_output = """
934
889
Name Enabled Timeout  Last Successful Check
935
890
foo  Yes     00:05:00 2019-02-03T00:00:00  
1079
1034
        for value_to_set, value_to_get in zip(self.values_to_set,
1080
1035
                                              values_to_get):
1081
1036
            for client in self.clients:
1082
 
                old_value = client.attributes[self.propname]
 
1037
                old_value = client.attributes[self.property]
1083
1038
                self.assertNotIsInstance(old_value, Unique)
1084
 
                client.attributes[self.propname] = Unique()
 
1039
                client.attributes[self.property] = Unique()
1085
1040
            self.run_command(value_to_set, self.clients)
1086
1041
            for client in self.clients:
1087
 
                value = client.attributes[self.propname]
 
1042
                value = client.attributes[self.property]
1088
1043
                self.assertNotIsInstance(value, Unique)
1089
1044
                self.assertEqual(value, value_to_get)
1090
1045
    def run_command(self, value, clients):
1092
1047
 
1093
1048
class TestBumpTimeoutCmd(TestPropertyCmd):
1094
1049
    command = BumpTimeoutCmd
1095
 
    propname = "LastCheckedOK"
 
1050
    property = "LastCheckedOK"
1096
1051
    values_to_set = [""]
1097
1052
 
1098
1053
class TestStartCheckerCmd(TestPropertyCmd):
1099
1054
    command = StartCheckerCmd
1100
 
    propname = "CheckerRunning"
 
1055
    property = "CheckerRunning"
1101
1056
    values_to_set = [dbus.Boolean(True)]
1102
1057
 
1103
1058
class TestStopCheckerCmd(TestPropertyCmd):
1104
1059
    command = StopCheckerCmd
1105
 
    propname = "CheckerRunning"
 
1060
    property = "CheckerRunning"
1106
1061
    values_to_set = [dbus.Boolean(False)]
1107
1062
 
1108
1063
class TestApproveByDefaultCmd(TestPropertyCmd):
1109
1064
    command = ApproveByDefaultCmd
1110
 
    propname = "ApprovedByDefault"
 
1065
    property = "ApprovedByDefault"
1111
1066
    values_to_set = [dbus.Boolean(True)]
1112
1067
 
1113
1068
class TestDenyByDefaultCmd(TestPropertyCmd):
1114
1069
    command = DenyByDefaultCmd
1115
 
    propname = "ApprovedByDefault"
 
1070
    property = "ApprovedByDefault"
1116
1071
    values_to_set = [dbus.Boolean(False)]
1117
1072
 
1118
1073
class TestValueArgumentPropertyCmd(TestPropertyCmd):
1127
1082
 
1128
1083
class TestSetCheckerCmd(TestValueArgumentPropertyCmd):
1129
1084
    command = SetCheckerCmd
1130
 
    propname = "Checker"
 
1085
    property = "Checker"
1131
1086
    values_to_set = ["", ":", "fping -q -- %s"]
1132
1087
 
1133
1088
class TestSetHostCmd(TestValueArgumentPropertyCmd):
1134
1089
    command = SetHostCmd
1135
 
    propname = "Host"
 
1090
    property = "Host"
1136
1091
    values_to_set = ["192.0.2.3", "foo.example.org"]
1137
1092
 
1138
1093
class TestSetSecretCmd(TestValueArgumentPropertyCmd):
1139
1094
    command = SetSecretCmd
1140
 
    propname = "Secret"
1141
 
    values_to_set = [io.BytesIO(b""),
 
1095
    property = "Secret"
 
1096
    values_to_set = [open("/dev/null", "rb"),
1142
1097
                     io.BytesIO(b"secret\0xyzzy\nbar")]
1143
1098
    values_to_get = [b"", b"secret\0xyzzy\nbar"]
1144
1099
 
1145
1100
class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):
1146
1101
    command = SetTimeoutCmd
1147
 
    propname = "Timeout"
1148
 
    values_to_set = [datetime.timedelta(),
1149
 
                     datetime.timedelta(minutes=5),
1150
 
                     datetime.timedelta(seconds=1),
1151
 
                     datetime.timedelta(weeks=1),
1152
 
                     datetime.timedelta(weeks=52)]
1153
 
    values_to_get = [0, 300000, 1000, 604800000, 31449600000]
 
1102
    property = "Timeout"
 
1103
    values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
 
1104
    values_to_get = [0, 300000, 1000, 120000, 31449600000]
1154
1105
 
1155
1106
class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):
1156
1107
    command = SetExtendedTimeoutCmd
1157
 
    propname = "ExtendedTimeout"
1158
 
    values_to_set = [datetime.timedelta(),
1159
 
                     datetime.timedelta(minutes=5),
1160
 
                     datetime.timedelta(seconds=1),
1161
 
                     datetime.timedelta(weeks=1),
1162
 
                     datetime.timedelta(weeks=52)]
1163
 
    values_to_get = [0, 300000, 1000, 604800000, 31449600000]
 
1108
    property = "ExtendedTimeout"
 
1109
    values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
 
1110
    values_to_get = [0, 300000, 1000, 120000, 31449600000]
1164
1111
 
1165
1112
class TestSetIntervalCmd(TestValueArgumentPropertyCmd):
1166
1113
    command = SetIntervalCmd
1167
 
    propname = "Interval"
1168
 
    values_to_set = [datetime.timedelta(),
1169
 
                     datetime.timedelta(minutes=5),
1170
 
                     datetime.timedelta(seconds=1),
1171
 
                     datetime.timedelta(weeks=1),
1172
 
                     datetime.timedelta(weeks=52)]
1173
 
    values_to_get = [0, 300000, 1000, 604800000, 31449600000]
 
1114
    property = "Interval"
 
1115
    values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
 
1116
    values_to_get = [0, 300000, 1000, 120000, 31449600000]
1174
1117
 
1175
1118
class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):
1176
1119
    command = SetApprovalDelayCmd
1177
 
    propname = "ApprovalDelay"
1178
 
    values_to_set = [datetime.timedelta(),
1179
 
                     datetime.timedelta(minutes=5),
1180
 
                     datetime.timedelta(seconds=1),
1181
 
                     datetime.timedelta(weeks=1),
1182
 
                     datetime.timedelta(weeks=52)]
1183
 
    values_to_get = [0, 300000, 1000, 604800000, 31449600000]
 
1120
    property = "ApprovalDelay"
 
1121
    values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
 
1122
    values_to_get = [0, 300000, 1000, 120000, 31449600000]
1184
1123
 
1185
1124
class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):
1186
1125
    command = SetApprovalDurationCmd
1187
 
    propname = "ApprovalDuration"
1188
 
    values_to_set = [datetime.timedelta(),
1189
 
                     datetime.timedelta(minutes=5),
1190
 
                     datetime.timedelta(seconds=1),
1191
 
                     datetime.timedelta(weeks=1),
1192
 
                     datetime.timedelta(weeks=52)]
1193
 
    values_to_get = [0, 300000, 1000, 604800000, 31449600000]
 
1126
    property = "ApprovalDuration"
 
1127
    values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
 
1128
    values_to_get = [0, 300000, 1000, 120000, 31449600000]
1194
1129
 
1195
1130
class Test_command_from_options(unittest.TestCase):
1196
1131
    def setUp(self):
1200
1135
        """Assert that parsing ARGS should result in an instance of
1201
1136
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1202
1137
        options = self.parser.parse_args(args)
1203
 
        check_option_syntax(self.parser, options)
1204
1138
        commands = commands_from_options(options)
1205
1139
        self.assertEqual(len(commands), 1)
1206
1140
        command = commands[0]
1215
1149
        self.assert_command_from_args(["--verbose"], PrintTableCmd,
1216
1150
                                      verbose=True)
1217
1151
 
1218
 
    def test_print_table_verbose_short(self):
1219
 
        self.assert_command_from_args(["-v"], PrintTableCmd,
1220
 
                                      verbose=True)
1221
 
 
1222
1152
    def test_enable(self):
1223
1153
        self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1224
1154
 
1225
 
    def test_enable_short(self):
1226
 
        self.assert_command_from_args(["-e", "foo"], EnableCmd)
1227
 
 
1228
1155
    def test_disable(self):
1229
1156
        self.assert_command_from_args(["--disable", "foo"],
1230
1157
                                      DisableCmd)
1231
1158
 
1232
 
    def test_disable_short(self):
1233
 
        self.assert_command_from_args(["-d", "foo"], DisableCmd)
1234
 
 
1235
1159
    def test_bump_timeout(self):
1236
1160
        self.assert_command_from_args(["--bump-timeout", "foo"],
1237
1161
                                      BumpTimeoutCmd)
1238
1162
 
1239
 
    def test_bump_timeout_short(self):
1240
 
        self.assert_command_from_args(["-b", "foo"], BumpTimeoutCmd)
1241
 
 
1242
1163
    def test_start_checker(self):
1243
1164
        self.assert_command_from_args(["--start-checker", "foo"],
1244
1165
                                      StartCheckerCmd)
1251
1172
        self.assert_command_from_args(["--remove", "foo"],
1252
1173
                                      RemoveCmd)
1253
1174
 
1254
 
    def test_remove_short(self):
1255
 
        self.assert_command_from_args(["-r", "foo"], RemoveCmd)
1256
 
 
1257
1175
    def test_checker(self):
1258
1176
        self.assert_command_from_args(["--checker", ":", "foo"],
1259
1177
                                      SetCheckerCmd, value_to_set=":")
1260
1178
 
1261
 
    def test_checker_empty(self):
1262
 
        self.assert_command_from_args(["--checker", "", "foo"],
1263
 
                                      SetCheckerCmd, value_to_set="")
1264
 
 
1265
 
    def test_checker_short(self):
1266
 
        self.assert_command_from_args(["-c", ":", "foo"],
1267
 
                                      SetCheckerCmd, value_to_set=":")
1268
 
 
1269
1179
    def test_timeout(self):
1270
1180
        self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1271
1181
                                      SetTimeoutCmd,
1272
1182
                                      value_to_set=300000)
1273
1183
 
1274
 
    def test_timeout_short(self):
1275
 
        self.assert_command_from_args(["-t", "PT5M", "foo"],
1276
 
                                      SetTimeoutCmd,
1277
 
                                      value_to_set=300000)
1278
 
 
1279
1184
    def test_extended_timeout(self):
1280
1185
        self.assert_command_from_args(["--extended-timeout", "PT15M",
1281
1186
                                       "foo"],
1287
1192
                                      SetIntervalCmd,
1288
1193
                                      value_to_set=120000)
1289
1194
 
1290
 
    def test_interval_short(self):
1291
 
        self.assert_command_from_args(["-i", "PT2M", "foo"],
1292
 
                                      SetIntervalCmd,
1293
 
                                      value_to_set=120000)
1294
 
 
1295
1195
    def test_approve_by_default(self):
1296
1196
        self.assert_command_from_args(["--approve-by-default", "foo"],
1297
1197
                                      ApproveByDefaultCmd)
1315
1215
                                       "foo"], SetHostCmd,
1316
1216
                                      value_to_set="foo.example.org")
1317
1217
 
1318
 
    def test_host_short(self):
1319
 
        self.assert_command_from_args(["-H", "foo.example.org",
1320
 
                                       "foo"], SetHostCmd,
1321
 
                                      value_to_set="foo.example.org")
1322
 
 
1323
1218
    def test_secret_devnull(self):
1324
1219
        self.assert_command_from_args(["--secret", os.path.devnull,
1325
1220
                                       "foo"], SetSecretCmd,
1334
1229
                                           "foo"], SetSecretCmd,
1335
1230
                                          value_to_set=value)
1336
1231
 
1337
 
    def test_secret_devnull_short(self):
1338
 
        self.assert_command_from_args(["-s", os.path.devnull, "foo"],
1339
 
                                      SetSecretCmd, value_to_set=b"")
1340
 
 
1341
 
    def test_secret_tempfile_short(self):
1342
 
        with tempfile.NamedTemporaryFile(mode="r+b") as f:
1343
 
            value = b"secret\0xyzzy\nbar"
1344
 
            f.write(value)
1345
 
            f.seek(0)
1346
 
            self.assert_command_from_args(["-s", f.name, "foo"],
1347
 
                                          SetSecretCmd,
1348
 
                                          value_to_set=value)
1349
 
 
1350
1232
    def test_approve(self):
1351
1233
        self.assert_command_from_args(["--approve", "foo"],
1352
1234
                                      ApproveCmd)
1353
1235
 
1354
 
    def test_approve_short(self):
1355
 
        self.assert_command_from_args(["-A", "foo"], ApproveCmd)
1356
 
 
1357
1236
    def test_deny(self):
1358
1237
        self.assert_command_from_args(["--deny", "foo"], DenyCmd)
1359
1238
 
1360
 
    def test_deny_short(self):
1361
 
        self.assert_command_from_args(["-D", "foo"], DenyCmd)
1362
 
 
1363
1239
    def test_dump_json(self):
1364
1240
        self.assert_command_from_args(["--dump-json"], DumpJSONCmd)
1365
1241
 
1367
1243
        self.assert_command_from_args(["--is-enabled", "foo"],
1368
1244
                                      IsEnabledCmd)
1369
1245
 
1370
 
    def test_is_enabled_short(self):
1371
 
        self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)
1372
 
 
1373
 
    def test_deny_before_remove(self):
1374
 
        options = self.parser.parse_args(["--deny", "--remove", "foo"])
1375
 
        check_option_syntax(self.parser, options)
1376
 
        commands = commands_from_options(options)
1377
 
        self.assertEqual(len(commands), 2)
1378
 
        self.assertIsInstance(commands[0], DenyCmd)
1379
 
        self.assertIsInstance(commands[1], RemoveCmd)
1380
 
 
1381
 
    def test_deny_before_remove_reversed(self):
1382
 
        options = self.parser.parse_args(["--remove", "--deny", "--all"])
1383
 
        check_option_syntax(self.parser, options)
1384
 
        commands = commands_from_options(options)
1385
 
        self.assertEqual(len(commands), 2)
1386
 
        self.assertIsInstance(commands[0], DenyCmd)
1387
 
        self.assertIsInstance(commands[1], RemoveCmd)
1388
 
 
1389
 
 
1390
 
class Test_check_option_syntax(unittest.TestCase):
1391
 
    # This mostly corresponds to the definition from has_actions() in
1392
 
    # check_option_syntax()
1393
 
    actions = {
1394
 
        # The actual values set here are not that important, but we do
1395
 
        # at least stick to the correct types, even though they are
1396
 
        # never used
1397
 
        "enable": True,
1398
 
        "disable": True,
1399
 
        "bump_timeout": True,
1400
 
        "start_checker": True,
1401
 
        "stop_checker": True,
1402
 
        "is_enabled": True,
1403
 
        "remove": True,
1404
 
        "checker": "x",
1405
 
        "timeout": datetime.timedelta(),
1406
 
        "extended_timeout": datetime.timedelta(),
1407
 
        "interval": datetime.timedelta(),
1408
 
        "approved_by_default": True,
1409
 
        "approval_delay": datetime.timedelta(),
1410
 
        "approval_duration": datetime.timedelta(),
1411
 
        "host": "x",
1412
 
        "secret": io.BytesIO(b"x"),
1413
 
        "approve": True,
1414
 
        "deny": True,
1415
 
    }
1416
 
 
1417
 
    def setUp(self):
1418
 
        self.parser = argparse.ArgumentParser()
1419
 
        add_command_line_options(self.parser)
1420
 
 
1421
 
    @contextlib.contextmanager
1422
 
    def assertParseError(self):
1423
 
        with self.assertRaises(SystemExit) as e:
1424
 
            with self.temporarily_suppress_stderr():
1425
 
                yield
1426
 
        # Exit code from argparse is guaranteed to be "2".  Reference:
1427
 
        # https://docs.python.org/3/library/argparse.html#exiting-methods
1428
 
        self.assertEqual(e.exception.code, 2)
1429
 
 
1430
 
    @staticmethod
1431
 
    @contextlib.contextmanager
1432
 
    def temporarily_suppress_stderr():
1433
 
        null = os.open(os.path.devnull, os.O_RDWR)
1434
 
        stderrcopy = os.dup(sys.stderr.fileno())
1435
 
        os.dup2(null, sys.stderr.fileno())
1436
 
        os.close(null)
1437
 
        try:
1438
 
            yield
1439
 
        finally:
1440
 
            # restore stderr
1441
 
            os.dup2(stderrcopy, sys.stderr.fileno())
1442
 
            os.close(stderrcopy)
1443
 
 
1444
 
    def check_option_syntax(self, options):
1445
 
        check_option_syntax(self.parser, options)
1446
 
 
1447
 
    def test_actions_requires_client_or_all(self):
1448
 
        for action, value in self.actions.items():
1449
 
            options = self.parser.parse_args()
1450
 
            setattr(options, action, value)
1451
 
            with self.assertParseError():
1452
 
                self.check_option_syntax(options)
1453
 
 
1454
 
    def test_actions_conflicts_with_verbose(self):
1455
 
        for action, value in self.actions.items():
1456
 
            options = self.parser.parse_args()
1457
 
            setattr(options, action, value)
1458
 
            options.verbose = True
1459
 
            with self.assertParseError():
1460
 
                self.check_option_syntax(options)
1461
 
 
1462
 
    def test_dump_json_conflicts_with_verbose(self):
1463
 
        options = self.parser.parse_args()
1464
 
        options.dump_json = True
1465
 
        options.verbose = True
1466
 
        with self.assertParseError():
1467
 
            self.check_option_syntax(options)
1468
 
 
1469
 
    def test_dump_json_conflicts_with_action(self):
1470
 
        for action, value in self.actions.items():
1471
 
            options = self.parser.parse_args()
1472
 
            setattr(options, action, value)
1473
 
            options.dump_json = True
1474
 
            with self.assertParseError():
1475
 
                self.check_option_syntax(options)
1476
 
 
1477
 
    def test_all_can_not_be_alone(self):
1478
 
        options = self.parser.parse_args()
1479
 
        options.all = True
1480
 
        with self.assertParseError():
1481
 
            self.check_option_syntax(options)
1482
 
 
1483
 
    def test_all_is_ok_with_any_action(self):
1484
 
        for action, value in self.actions.items():
1485
 
            options = self.parser.parse_args()
1486
 
            setattr(options, action, value)
1487
 
            options.all = True
1488
 
            self.check_option_syntax(options)
1489
 
 
1490
 
    def test_is_enabled_fails_without_client(self):
1491
 
        options = self.parser.parse_args()
1492
 
        options.is_enabled = True
1493
 
        with self.assertParseError():
1494
 
            self.check_option_syntax(options)
1495
 
 
1496
 
    def test_is_enabled_works_with_one_client(self):
1497
 
        options = self.parser.parse_args()
1498
 
        options.is_enabled = True
1499
 
        options.client = ["foo"]
1500
 
        self.check_option_syntax(options)
1501
 
 
1502
 
    def test_is_enabled_fails_with_two_clients(self):
1503
 
        options = self.parser.parse_args()
1504
 
        options.is_enabled = True
1505
 
        options.client = ["foo", "barbar"]
1506
 
        with self.assertParseError():
1507
 
            self.check_option_syntax(options)
1508
 
 
1509
 
    def test_remove_can_only_be_combined_with_action_deny(self):
1510
 
        for action, value in self.actions.items():
1511
 
            if action in {"remove", "deny"}:
1512
 
                continue
1513
 
            options = self.parser.parse_args()
1514
 
            setattr(options, action, value)
1515
 
            options.all = True
1516
 
            options.remove = True
1517
 
            with self.assertParseError():
1518
 
                self.check_option_syntax(options)
1519
 
 
1520
1246
 
1521
1247
 
1522
1248
def should_only_run_tests():