/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-03 16:25:45 UTC
  • Revision ID: teddy@recompile.se-20190303162545-vjdm7icnhpprzuwb
mandos-ctl: Refactor; move TableOfClients into PrintTableCmd

* mandos-ctl (TableOfClients): Moved into PrintTableCmd; user changed.
  (PrintTableCmd.TableOfClients): New.

Show diffs side-by-side

added added

removed removed

Lines of Context:
278
278
        commands which want to operate on all clients at the same time
279
279
        can override this run() method instead."""
280
280
        self.mandos = mandos
281
 
        for client, properties in clients.items():
282
 
            self.run_on_one_client(client, properties)
 
281
        for client in clients:
 
282
            self.run_on_one_client(client)
283
283
 
284
284
class PrintCmd(Command):
285
285
    """Abstract class for commands printing client details"""
295
295
 
296
296
class PropertyCmd(Command):
297
297
    """Abstract class for Actions for setting one client property"""
298
 
    def run_on_one_client(self, client, properties):
 
298
    def run_on_one_client(self, client):
299
299
        """Set the Client's D-Bus property"""
300
300
        client.Set(client_interface, self.property, self.value_to_set,
301
301
                   dbus_interface=dbus.PROPERTIES_IFACE)
323
323
        self.verbose = verbose
324
324
 
325
325
    def output(self, clients):
326
 
        default_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
327
 
        keywords = default_keywords
328
326
        if self.verbose:
329
327
            keywords = self.all_keywords
 
328
        else:
 
329
            keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
330
330
        return str(self.TableOfClients(clients.values(), keywords))
331
331
 
332
332
    class TableOfClients(object):
419
419
        return value
420
420
 
421
421
class IsEnabledCmd(Command):
422
 
    def run_on_one_client(self, client, properties):
423
 
        if self.is_enabled(client, properties):
 
422
    def run_on_one_client(self, client):
 
423
        if self.is_enabled(client):
424
424
            sys.exit(0)
425
425
        sys.exit(1)
426
 
    def is_enabled(self, client, properties):
427
 
        return bool(properties["Enabled"])
 
426
    def is_enabled(self, client):
 
427
        return client.Get(client_interface, "Enabled",
 
428
                          dbus_interface=dbus.PROPERTIES_IFACE)
428
429
 
429
430
class RemoveCmd(Command):
430
 
    def run_on_one_client(self, client, properties):
 
431
    def run_on_one_client(self, client):
431
432
        self.mandos.RemoveClient(client.__dbus_object_path__)
432
433
 
433
434
class ApproveCmd(Command):
434
 
    def run_on_one_client(self, client, properties):
 
435
    def run_on_one_client(self, client):
435
436
        client.Approve(dbus.Boolean(True),
436
437
                       dbus_interface=client_interface)
437
438
 
438
439
class DenyCmd(Command):
439
 
    def run_on_one_client(self, client, properties):
 
440
    def run_on_one_client(self, client):
440
441
        client.Approve(dbus.Boolean(False),
441
442
                       dbus_interface=client_interface)
442
443
 
515
516
                options.approve,
516
517
                options.deny))
517
518
 
518
 
def add_command_line_options(parser):
 
519
 
 
520
def commands_and_clients_from_options(args=None):
 
521
    if args is None:
 
522
        args=sys.argv[1:]
 
523
    parser = argparse.ArgumentParser()
519
524
    parser.add_argument("--version", action="version",
520
525
                        version="%(prog)s {}".format(version),
521
526
                        help="show version number and exit")
578
583
    parser.add_argument("--check", action="store_true",
579
584
                        help="Run self-test")
580
585
    parser.add_argument("client", nargs="*", help="Client name")
581
 
 
582
 
 
583
 
def commands_from_options(options):
 
586
    options = parser.parse_args(args=args)
 
587
 
 
588
    if has_actions(options) and not (options.client or options.all):
 
589
        parser.error("Options require clients names or --all.")
 
590
    if options.verbose and has_actions(options):
 
591
        parser.error("--verbose can only be used alone.")
 
592
    if options.dump_json and (options.verbose
 
593
                              or has_actions(options)):
 
594
        parser.error("--dump-json can only be used alone.")
 
595
    if options.all and not has_actions(options):
 
596
        parser.error("--all requires an action.")
 
597
    if options.is_enabled and len(options.client) > 1:
 
598
            parser.error("--is-enabled requires exactly one client")
584
599
 
585
600
    commands = []
586
601
 
594
609
        commands.append(DisableCmd())
595
610
 
596
611
    if options.bump_timeout:
597
 
        commands.append(BumpTimeoutCmd())
 
612
        commands.append(BumpTimeoutCmd(options.bump_timeout))
598
613
 
599
614
    if options.start_checker:
600
615
        commands.append(StartCheckerCmd())
651
666
    if not commands:
652
667
        commands.append(PrintTableCmd(verbose=options.verbose))
653
668
 
654
 
    return commands
 
669
    return commands, options.client
655
670
 
656
671
 
657
672
def main():
658
 
    parser = argparse.ArgumentParser()
659
 
 
660
 
    add_command_line_options(parser)
661
 
 
662
 
    options = parser.parse_args()
663
 
 
664
 
    if has_actions(options) and not (options.client or options.all):
665
 
        parser.error("Options require clients names or --all.")
666
 
    if options.verbose and has_actions(options):
667
 
        parser.error("--verbose can only be used alone.")
668
 
    if options.dump_json and (options.verbose
669
 
                              or has_actions(options)):
670
 
        parser.error("--dump-json can only be used alone.")
671
 
    if options.all and not has_actions(options):
672
 
        parser.error("--all requires an action.")
673
 
    if options.is_enabled and len(options.client) > 1:
674
 
        parser.error("--is-enabled requires exactly one client")
675
 
 
676
 
    clientnames = options.client
 
673
    commands, clientnames = commands_and_clients_from_options()
677
674
 
678
675
    try:
679
676
        bus = dbus.SystemBus()
693
690
        def filter(self, record):
694
691
            return False
695
692
    dbus_filter = NullFilter()
 
693
    dbus_logger.addFilter(dbus_filter)
696
694
    try:
697
 
        dbus_logger.addFilter(dbus_filter)
698
 
        mandos_clients = {path: ifs_and_props[client_interface]
699
 
                          for path, ifs_and_props in
700
 
                          mandos_serv_object_manager
701
 
                          .GetManagedObjects().items()
702
 
                          if client_interface in ifs_and_props}
 
695
        try:
 
696
            mandos_clients = {path: ifs_and_props[client_interface]
 
697
                              for path, ifs_and_props in
 
698
                              mandos_serv_object_manager
 
699
                              .GetManagedObjects().items()
 
700
                              if client_interface in ifs_and_props}
 
701
        finally:
 
702
            # restore dbus logger
 
703
            dbus_logger.removeFilter(dbus_filter)
703
704
    except dbus.exceptions.DBusException as e:
704
705
        log.critical("Failed to access Mandos server through D-Bus:"
705
706
                     "\n%s", e)
706
707
        sys.exit(1)
707
 
    finally:
708
 
        # restore dbus logger
709
 
        dbus_logger.removeFilter(dbus_filter)
710
708
 
711
709
    # Compile dict of (clients: properties) to process
712
710
    clients = {}
726
724
                sys.exit(1)
727
725
 
728
726
    # Run all commands on clients
729
 
    commands = commands_from_options(options)
730
727
    for command in commands:
731
728
        command.run(mandos_serv, clients)
732
729
 
746
743
 
747
744
class Test_string_to_delta(unittest.TestCase):
748
745
    def test_handles_basic_rfc3339(self):
749
 
        self.assertEqual(string_to_delta("PT0S"),
750
 
                         datetime.timedelta())
751
 
        self.assertEqual(string_to_delta("P0D"),
752
 
                         datetime.timedelta())
753
 
        self.assertEqual(string_to_delta("PT1S"),
754
 
                         datetime.timedelta(0, 1))
755
746
        self.assertEqual(string_to_delta("PT2H"),
756
747
                         datetime.timedelta(0, 7200))
757
748
    def test_falls_back_to_pre_1_6_1_with_warning(self):
786
777
                self.__dbus_object_path__ = "objpath_{}".format(name)
787
778
                self.attributes = attributes
788
779
                self.attributes["Name"] = name
789
 
                self.calls = []
790
 
            def Set(self, interface, property, value, dbus_interface):
 
780
            def Set(interface, property, value,
 
781
                    properties_interface):
791
782
                testcase.assertEqual(interface, client_interface)
792
 
                testcase.assertEqual(dbus_interface,
 
783
                testcase.assertEqual(properties_interface,
793
784
                                     dbus.PROPERTIES_IFACE)
794
785
                self.attributes[property] = value
795
 
            def Get(self, interface, property, dbus_interface):
 
786
            def Get(interface, property, properties_interface):
796
787
                testcase.assertEqual(interface, client_interface)
797
 
                testcase.assertEqual(dbus_interface,
 
788
                testcase.assertEqual(properties_interface,
798
789
                                     dbus.PROPERTIES_IFACE)
799
790
                return self.attributes[property]
800
 
            def Approve(self, approve, dbus_interface):
801
 
                testcase.assertEqual(dbus_interface, client_interface)
802
 
                self.calls.append(("Approve", (approve,
803
 
                                               dbus_interface)))
804
 
        self.client = MockClient(
805
 
            "foo",
806
 
            KeyID=("92ed150794387c03ce684574b1139a65"
807
 
                   "94a34f895daaaf09fd8ea90a27cddb12"),
808
 
            Secret=b"secret",
809
 
            Host="foo.example.org",
810
 
            Enabled=dbus.Boolean(True),
811
 
            Timeout=300000,
812
 
            LastCheckedOK="2019-02-03T00:00:00",
813
 
            Created="2019-01-02T00:00:00",
814
 
            Interval=120000,
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="",
822
 
            ApprovalDelay=0,
823
 
            ApprovalDuration=1000,
824
 
            Checker="fping -q -- %(host)s",
825
 
            ExtendedTimeout=900000,
826
 
            Expires="2019-02-04T00:00:00",
827
 
            LastCheckerStatus=0)
828
 
        self.other_client = MockClient(
829
 
            "barbar",
830
 
            KeyID=("0558568eedd67d622f5c83b35a115f79"
831
 
                   "6ab612cff5ad227247e46c2b020f441c"),
832
 
            Secret=b"secretbar",
833
 
            Host="192.0.2.3",
834
 
            Enabled=dbus.Boolean(True),
835
 
            Timeout=300000,
836
 
            LastCheckedOK="2019-02-04T00:00:00",
837
 
            Created="2019-01-03T00:00:00",
838
 
            Interval=120000,
839
 
            Fingerprint=("3E393AEAEFB84C7E89E2"
840
 
                         "F547B3A107558FCA3A27"),
841
 
            CheckerRunning=dbus.Boolean(True),
842
 
            LastEnabled="2019-01-04T00:00:00",
843
 
            ApprovalPending=dbus.Boolean(False),
844
 
            ApprovedByDefault=dbus.Boolean(False),
845
 
            LastApprovalRequest="2019-01-03T00:00:00",
846
 
            ApprovalDelay=30000,
847
 
            ApprovalDuration=1000,
848
 
            Checker=":",
849
 
            ExtendedTimeout=900000,
850
 
            Expires="2019-02-05T00:00:00",
851
 
            LastCheckerStatus=-2)
852
 
        self.clients =  collections.OrderedDict(
853
 
            [
854
 
                (self.client, self.client.attributes),
855
 
                (self.other_client, self.other_client.attributes),
 
791
            def __getitem__(self, key):
 
792
                return self.attributes[key]
 
793
        self.clients = collections.OrderedDict([
 
794
            ("foo",
 
795
             MockClient(
 
796
                 "foo",
 
797
                 KeyID=("92ed150794387c03ce684574b1139a65"
 
798
                        "94a34f895daaaf09fd8ea90a27cddb12"),
 
799
                 Secret=b"secret",
 
800
                 Host="foo.example.org",
 
801
                 Enabled=dbus.Boolean(True),
 
802
                 Timeout=300000,
 
803
                 LastCheckedOK="2019-02-03T00:00:00",
 
804
                 Created="2019-01-02T00:00:00",
 
805
                 Interval=120000,
 
806
                 Fingerprint=("778827225BA7DE539C5A"
 
807
                              "7CFA59CFF7CDBD9A5920"),
 
808
                 CheckerRunning=dbus.Boolean(False),
 
809
                 LastEnabled="2019-01-03T00:00:00",
 
810
                 ApprovalPending=dbus.Boolean(False),
 
811
                 ApprovedByDefault=dbus.Boolean(True),
 
812
                 LastApprovalRequest="",
 
813
                 ApprovalDelay=0,
 
814
                 ApprovalDuration=1000,
 
815
                 Checker="fping -q -- %(host)s",
 
816
                 ExtendedTimeout=900000,
 
817
                 Expires="2019-02-04T00:00:00",
 
818
                 LastCheckerStatus=0)),
 
819
            ("barbar",
 
820
             MockClient(
 
821
                 "barbar",
 
822
                 KeyID=("0558568eedd67d622f5c83b35a115f79"
 
823
                        "6ab612cff5ad227247e46c2b020f441c"),
 
824
                 Secret=b"secretbar",
 
825
                 Host="192.0.2.3",
 
826
                 Enabled=dbus.Boolean(True),
 
827
                 Timeout=300000,
 
828
                 LastCheckedOK="2019-02-04T00:00:00",
 
829
                 Created="2019-01-03T00:00:00",
 
830
                 Interval=120000,
 
831
                 Fingerprint=("3E393AEAEFB84C7E89E2"
 
832
                              "F547B3A107558FCA3A27"),
 
833
                 CheckerRunning=dbus.Boolean(True),
 
834
                 LastEnabled="2019-01-04T00:00:00",
 
835
                 ApprovalPending=dbus.Boolean(False),
 
836
                 ApprovedByDefault=dbus.Boolean(False),
 
837
                 LastApprovalRequest="2019-01-03T00:00:00",
 
838
                 ApprovalDelay=30000,
 
839
                 ApprovalDuration=1000,
 
840
                 Checker=":",
 
841
                 ExtendedTimeout=900000,
 
842
                 Expires="2019-02-05T00:00:00",
 
843
                 LastCheckerStatus=-2)),
856
844
            ])
857
 
        self.one_client = {self.client: self.client.attributes}
858
845
 
859
846
class TestPrintTableCmd(TestCmd):
860
847
    def test_normal(self):
874
861
"""[1:-1]
875
862
        self.assertEqual(output, expected_output)
876
863
    def test_one_client(self):
877
 
        output = PrintTableCmd().output(self.one_client)
 
864
        output = PrintTableCmd().output({"foo": self.clients["foo"]})
878
865
        expected_output = """
879
866
Name Enabled Timeout  Last Successful Check
880
867
foo  Yes     00:05:00 2019-02-03T00:00:00  
881
868
"""[1:-1]
882
869
        self.assertEqual(output, expected_output)
883
870
 
884
 
class TestDumpJSONCmd(TestCmd):
885
 
    def setUp(self):
886
 
        self.expected_json = {
887
 
            "foo": {
888
 
                "Name": "foo",
889
 
                "KeyID": ("92ed150794387c03ce684574b1139a65"
890
 
                          "94a34f895daaaf09fd8ea90a27cddb12"),
891
 
                "Host": "foo.example.org",
892
 
                "Enabled": True,
893
 
                "Timeout": 300000,
894
 
                "LastCheckedOK": "2019-02-03T00:00:00",
895
 
                "Created": "2019-01-02T00:00:00",
896
 
                "Interval": 120000,
897
 
                "Fingerprint": ("778827225BA7DE539C5A"
898
 
                                "7CFA59CFF7CDBD9A5920"),
899
 
                "CheckerRunning": False,
900
 
                "LastEnabled": "2019-01-03T00:00:00",
901
 
                "ApprovalPending": False,
902
 
                "ApprovedByDefault": True,
903
 
                "LastApprovalRequest": "",
904
 
                "ApprovalDelay": 0,
905
 
                "ApprovalDuration": 1000,
906
 
                "Checker": "fping -q -- %(host)s",
907
 
                "ExtendedTimeout": 900000,
908
 
                "Expires": "2019-02-04T00:00:00",
909
 
                "LastCheckerStatus": 0,
910
 
            },
911
 
            "barbar": {
912
 
                "Name": "barbar",
913
 
                "KeyID": ("0558568eedd67d622f5c83b35a115f79"
914
 
                          "6ab612cff5ad227247e46c2b020f441c"),
915
 
                "Host": "192.0.2.3",
916
 
                "Enabled": True,
917
 
                "Timeout": 300000,
918
 
                "LastCheckedOK": "2019-02-04T00:00:00",
919
 
                "Created": "2019-01-03T00:00:00",
920
 
                "Interval": 120000,
921
 
                "Fingerprint": ("3E393AEAEFB84C7E89E2"
922
 
                                "F547B3A107558FCA3A27"),
923
 
                "CheckerRunning": True,
924
 
                "LastEnabled": "2019-01-04T00:00:00",
925
 
                "ApprovalPending": False,
926
 
                "ApprovedByDefault": False,
927
 
                "LastApprovalRequest": "2019-01-03T00:00:00",
928
 
                "ApprovalDelay": 30000,
929
 
                "ApprovalDuration": 1000,
930
 
                "Checker": ":",
931
 
                "ExtendedTimeout": 900000,
932
 
                "Expires": "2019-02-05T00:00:00",
933
 
                "LastCheckerStatus": -2,
934
 
            },
935
 
        }
936
 
        return super(TestDumpJSONCmd, self).setUp()
937
 
    def test_normal(self):
938
 
        json_data = json.loads(DumpJSONCmd().output(self.clients))
939
 
        self.assertDictEqual(json_data, self.expected_json)
940
 
    def test_one_client(self):
941
 
        clients = self.one_client
942
 
        json_data = json.loads(DumpJSONCmd().output(clients))
943
 
        expected_json = {"foo": self.expected_json["foo"]}
944
 
        self.assertDictEqual(json_data, expected_json)
945
 
 
946
 
class TestIsEnabledCmd(TestCmd):
947
 
    def test_is_enabled(self):
948
 
        self.assertTrue(all(IsEnabledCmd().is_enabled(client, properties)
949
 
                            for client, properties in self.clients.items()))
950
 
    def test_is_enabled_run_exits_successfully(self):
951
 
        with self.assertRaises(SystemExit) as e:
952
 
            IsEnabledCmd().run(None, self.one_client)
953
 
        if e.exception.code is not None:
954
 
            self.assertEqual(e.exception.code, 0)
955
 
        else:
956
 
            self.assertIsNone(e.exception.code)
957
 
    def test_is_enabled_run_exits_with_failure(self):
958
 
        self.client.attributes["Enabled"] = dbus.Boolean(False)
959
 
        with self.assertRaises(SystemExit) as e:
960
 
            IsEnabledCmd().run(None, self.one_client)
961
 
        if isinstance(e.exception.code, int):
962
 
            self.assertNotEqual(e.exception.code, 0)
963
 
        else:
964
 
            self.assertIsNotNone(e.exception.code)
965
 
 
966
 
class TestRemoveCmd(TestCmd):
967
 
    def test_remove(self):
968
 
        class MockMandos(object):
969
 
            def __init__(self):
970
 
                self.calls = []
971
 
            def RemoveClient(self, dbus_path):
972
 
                self.calls.append(("RemoveClient", (dbus_path,)))
973
 
        mandos = MockMandos()
974
 
        super(TestRemoveCmd, self).setUp()
975
 
        RemoveCmd().run(mandos, self.clients)
976
 
        self.assertEqual(len(mandos.calls), 2)
977
 
        for client in self.clients:
978
 
            self.assertIn(("RemoveClient",
979
 
                           (client.__dbus_object_path__,)),
980
 
                          mandos.calls)
981
 
 
982
 
class TestApproveCmd(TestCmd):
983
 
    def test_approve(self):
984
 
        ApproveCmd().run(None, self.clients)
985
 
        for client in self.clients:
986
 
            self.assertIn(("Approve", (True, client_interface)),
987
 
                          client.calls)
988
 
 
989
 
class TestDenyCmd(TestCmd):
990
 
    def test_deny(self):
991
 
        DenyCmd().run(None, self.clients)
992
 
        for client in self.clients:
993
 
            self.assertIn(("Approve", (False, client_interface)),
994
 
                          client.calls)
995
 
 
996
 
class TestEnableCmd(TestCmd):
997
 
    def test_enable(self):
998
 
        for client in self.clients:
999
 
            client.attributes["Enabled"] = False
1000
 
 
1001
 
        EnableCmd().run(None, self.clients)
1002
 
 
1003
 
        for client in self.clients:
1004
 
            self.assertTrue(client.attributes["Enabled"])
1005
 
 
1006
 
class TestDisableCmd(TestCmd):
1007
 
    def test_disable(self):
1008
 
        DisableCmd().run(None, self.clients)
1009
 
 
1010
 
        for client in self.clients:
1011
 
            self.assertFalse(client.attributes["Enabled"])
1012
 
 
1013
 
class Unique(object):
1014
 
    """Class for objects which exist only to be unique objects, since
1015
 
unittest.mock.sentinel only exists in Python 3.3"""
1016
 
 
1017
 
class TestPropertyCmd(TestCmd):
1018
 
    """Abstract class for tests of PropertyCmd classes"""
1019
 
    def runTest(self):
1020
 
        if not hasattr(self, "command"):
1021
 
            return
1022
 
        values_to_get = getattr(self, "values_to_get",
1023
 
                                self.values_to_set)
1024
 
        for value_to_set, value_to_get in zip(self.values_to_set,
1025
 
                                              values_to_get):
1026
 
            for client in self.clients:
1027
 
                old_value = client.attributes[self.property]
1028
 
                self.assertNotIsInstance(old_value, Unique)
1029
 
                client.attributes[self.property] = Unique()
1030
 
            self.run_command(value_to_set, self.clients)
1031
 
            for client in self.clients:
1032
 
                value = client.attributes[self.property]
1033
 
                self.assertNotIsInstance(value, Unique)
1034
 
                self.assertEqual(value, value_to_get)
1035
 
    def run_command(self, value, clients):
1036
 
        self.command().run(None, clients)
1037
 
 
1038
 
class TestBumpTimeoutCmd(TestPropertyCmd):
1039
 
    command = BumpTimeoutCmd
1040
 
    property = "LastCheckedOK"
1041
 
    values_to_set = [""]
1042
 
 
1043
 
class TestStartCheckerCmd(TestPropertyCmd):
1044
 
    command = StartCheckerCmd
1045
 
    property = "CheckerRunning"
1046
 
    values_to_set = [dbus.Boolean(True)]
1047
 
 
1048
 
class TestStopCheckerCmd(TestPropertyCmd):
1049
 
    command = StopCheckerCmd
1050
 
    property = "CheckerRunning"
1051
 
    values_to_set = [dbus.Boolean(False)]
1052
 
 
1053
 
class TestApproveByDefaultCmd(TestPropertyCmd):
1054
 
    command = ApproveByDefaultCmd
1055
 
    property = "ApprovedByDefault"
1056
 
    values_to_set = [dbus.Boolean(True)]
1057
 
 
1058
 
class TestDenyByDefaultCmd(TestPropertyCmd):
1059
 
    command = DenyByDefaultCmd
1060
 
    property = "ApprovedByDefault"
1061
 
    values_to_set = [dbus.Boolean(False)]
1062
 
 
1063
 
class TestValueArgumentPropertyCmd(TestPropertyCmd):
1064
 
    """Abstract class for tests of PropertyCmd classes using the
1065
 
ValueArgumentMixIn"""
1066
 
    def runTest(self):
1067
 
        if type(self) is TestValueArgumentPropertyCmd:
1068
 
            return
1069
 
        return super(TestValueArgumentPropertyCmd, self).runTest()
1070
 
    def run_command(self, value, clients):
1071
 
        self.command(value).run(None, clients)
1072
 
 
1073
 
class TestSetCheckerCmd(TestValueArgumentPropertyCmd):
1074
 
    command = SetCheckerCmd
1075
 
    property = "Checker"
1076
 
    values_to_set = ["", ":", "fping -q -- %s"]
1077
 
 
1078
 
class TestSetHostCmd(TestValueArgumentPropertyCmd):
1079
 
    command = SetHostCmd
1080
 
    property = "Host"
1081
 
    values_to_set = ["192.0.2.3", "foo.example.org"]
1082
 
 
1083
 
class TestSetSecretCmd(TestValueArgumentPropertyCmd):
1084
 
    command = SetSecretCmd
1085
 
    property = "Secret"
1086
 
    values_to_set = [b"", b"secret"]
1087
 
 
1088
 
class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):
1089
 
    command = SetTimeoutCmd
1090
 
    property = "Timeout"
1091
 
    values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1092
 
    values_to_get = [0, 300000, 1000, 120000, 31449600000]
1093
 
 
1094
 
class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):
1095
 
    command = SetExtendedTimeoutCmd
1096
 
    property = "ExtendedTimeout"
1097
 
    values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1098
 
    values_to_get = [0, 300000, 1000, 120000, 31449600000]
1099
 
 
1100
 
class TestSetIntervalCmd(TestValueArgumentPropertyCmd):
1101
 
    command = SetIntervalCmd
1102
 
    property = "Interval"
1103
 
    values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1104
 
    values_to_get = [0, 300000, 1000, 120000, 31449600000]
1105
 
 
1106
 
class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):
1107
 
    command = SetApprovalDelayCmd
1108
 
    property = "ApprovalDelay"
1109
 
    values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1110
 
    values_to_get = [0, 300000, 1000, 120000, 31449600000]
1111
 
 
1112
 
class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):
1113
 
    command = SetApprovalDurationCmd
1114
 
    property = "ApprovalDuration"
1115
 
    values_to_set = ["P0D", "PT5M", "PT1S", "PT120S", "P1Y"]
1116
 
    values_to_get = [0, 300000, 1000, 120000, 31449600000]
1117
 
 
1118
 
class TestOptions(unittest.TestCase):
1119
 
    def setUp(self):
1120
 
        self.parser = argparse.ArgumentParser()
1121
 
        add_command_line_options(self.parser)
1122
 
    def assert_command_from_args(self, args, command_cls, **cmd_attrs):
1123
 
        """Assert that parsing ARGS should result in an instance of
1124
 
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1125
 
        options = self.parser.parse_args(args)
1126
 
        commands = commands_from_options(options)
1127
 
        self.assertEqual(len(commands), 1)
1128
 
        command = commands[0]
1129
 
        self.assertIsInstance(command, command_cls)
1130
 
        for key, value in cmd_attrs.items():
1131
 
            self.assertEqual(getattr(command, key), value)
1132
 
    def test_default_is_show_table(self):
1133
 
        self.assert_command_from_args([], PrintTableCmd,
1134
 
                                      verbose=False)
1135
 
    def test_show_table_verbose(self):
1136
 
        self.assert_command_from_args(["--verbose"], PrintTableCmd,
1137
 
                                      verbose=True)
1138
 
    def test_enable(self):
1139
 
        self.assert_command_from_args(["--enable"], EnableCmd)
1140
 
    def test_disable(self):
1141
 
        self.assert_command_from_args(["--disable"], DisableCmd)
1142
 
 
1143
871
 
1144
872
 
1145
873
def should_only_run_tests():