/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-17 10:59:24 UTC
  • Revision ID: teddy@recompile.se-20190317105924-3o9xb4pl8tld3rnp
mandos-ctl: Refactor

* mandos-ctl: Move all command classes into a class acting as a
              namespace.  Remove "Cmd" suffix from command class
              names.  All users of commands changed.  Also unify all
              tests of base commands into one test class.
  (command): New.
  (TestBaseCommands): New.
  (TestIsEnableCmd, TestApproveCmd, TestDenyCmd, TestRemoveCmd,
  TestDumpJSONCmd, TestPrintTableCmd): Remove; move tests into
                                       TestBaseCommands.

Show diffs side-by-side

added added

removed removed

Lines of Context:
mandos-ctl
61
61
 
62
62
if sys.version_info.major == 2:
63
63
    str = unicode
64
 
    import StringIO
65
 
    io.StringIO = StringIO.StringIO
66
64
 
67
65
locale.setlocale(locale.LC_ALL, "")
68
66
 
83
81
 
84
82
def main():
85
83
    parser = argparse.ArgumentParser()
 
84
 
86
85
    add_command_line_options(parser)
87
86
 
88
87
    options = parser.parse_args()
 
88
 
89
89
    check_option_syntax(parser, options)
90
90
 
91
91
    clientnames = options.client
460
460
 
461
461
    def __enter__(self):
462
462
        self.logger.addFilter(self.nullfilter)
 
463
        return self
463
464
 
464
465
    class NullFilter(logging.Filter):
465
466
        def filter(self, record):
612
613
                        "Checker", "ExtendedTimeout", "Expires",
613
614
                        "LastCheckerStatus")
614
615
 
 
616
        def run(self, clients, bus=None, mandos=None):
 
617
            print(self.output(clients.values()))
 
618
 
 
619
        def output(self, clients):
 
620
            raise NotImplementedError()
 
621
 
615
622
 
616
623
    class DumpJSON(Output):
617
 
        def run(self, clients, bus=None, mandos=None):
 
624
        def output(self, clients):
618
625
            data = {client["Name"]:
619
626
                    {key: self.dbus_boolean_to_bool(client[key])
620
627
                     for key in self.all_keywords}
621
 
                    for client in clients.values()}
622
 
            print(json.dumps(data, indent=4, separators=(',', ': ')))
 
628
                    for client in clients}
 
629
            return json.dumps(data, indent=4, separators=(',', ': '))
623
630
 
624
631
        @staticmethod
625
632
        def dbus_boolean_to_bool(value):
632
639
        def __init__(self, verbose=False):
633
640
            self.verbose = verbose
634
641
 
635
 
        def run(self, clients, bus=None, mandos=None):
 
642
        def output(self, clients):
636
643
            default_keywords = ("Name", "Enabled", "Timeout",
637
644
                                "LastCheckedOK")
638
645
            keywords = default_keywords
639
646
            if self.verbose:
640
647
                keywords = self.all_keywords
641
 
            print(self.TableOfClients(clients.values(), keywords))
 
648
            return str(self.TableOfClients(clients, keywords))
642
649
 
643
650
        class TableOfClients(object):
644
651
            tableheaders = {
725
732
                                seconds=td.seconds % 60))
726
733
 
727
734
 
728
 
    class PropertySetter(Base):
 
735
    class Property(Base):
729
736
        "Abstract class for Actions for setting one client property"
730
737
 
731
738
        def run_on_one_client(self, client, properties):
746
753
            raise NotImplementedError()
747
754
 
748
755
 
749
 
    class Enable(PropertySetter):
 
756
    class Enable(Property):
750
757
        propname = "Enabled"
751
758
        value_to_set = dbus.Boolean(True)
752
759
 
753
760
 
754
 
    class Disable(PropertySetter):
 
761
    class Disable(Property):
755
762
        propname = "Enabled"
756
763
        value_to_set = dbus.Boolean(False)
757
764
 
758
765
 
759
 
    class BumpTimeout(PropertySetter):
 
766
    class BumpTimeout(Property):
760
767
        propname = "LastCheckedOK"
761
768
        value_to_set = ""
762
769
 
763
770
 
764
 
    class StartChecker(PropertySetter):
765
 
        propname = "CheckerRunning"
766
 
        value_to_set = dbus.Boolean(True)
767
 
 
768
 
 
769
 
    class StopChecker(PropertySetter):
770
 
        propname = "CheckerRunning"
771
 
        value_to_set = dbus.Boolean(False)
772
 
 
773
 
 
774
 
    class ApproveByDefault(PropertySetter):
775
 
        propname = "ApprovedByDefault"
776
 
        value_to_set = dbus.Boolean(True)
777
 
 
778
 
 
779
 
    class DenyByDefault(PropertySetter):
780
 
        propname = "ApprovedByDefault"
781
 
        value_to_set = dbus.Boolean(False)
782
 
 
783
 
 
784
 
    class PropertySetterValue(PropertySetter):
785
 
        """Abstract class for PropertySetter recieving a value as
786
 
constructor argument instead of a class attribute."""
 
771
    class StartChecker(Property):
 
772
        propname = "CheckerRunning"
 
773
        value_to_set = dbus.Boolean(True)
 
774
 
 
775
 
 
776
    class StopChecker(Property):
 
777
        propname = "CheckerRunning"
 
778
        value_to_set = dbus.Boolean(False)
 
779
 
 
780
 
 
781
    class ApproveByDefault(Property):
 
782
        propname = "ApprovedByDefault"
 
783
        value_to_set = dbus.Boolean(True)
 
784
 
 
785
 
 
786
    class DenyByDefault(Property):
 
787
        propname = "ApprovedByDefault"
 
788
        value_to_set = dbus.Boolean(False)
 
789
 
 
790
 
 
791
    class PropertyValue(Property):
 
792
        "Abstract class for Property recieving a value as argument"
787
793
        def __init__(self, value):
788
794
            self.value_to_set = value
789
795
 
790
796
 
791
 
    class SetChecker(PropertySetterValue):
 
797
    class SetChecker(PropertyValue):
792
798
        propname = "Checker"
793
799
 
794
800
 
795
 
    class SetHost(PropertySetterValue):
 
801
    class SetHost(PropertyValue):
796
802
        propname = "Host"
797
803
 
798
804
 
799
 
    class SetSecret(PropertySetterValue):
 
805
    class SetSecret(PropertyValue):
800
806
        propname = "Secret"
801
807
 
802
808
        @property
810
816
            value.close()
811
817
 
812
818
 
813
 
    class PropertySetterValueMilliseconds(PropertySetterValue):
814
 
        """Abstract class for PropertySetterValue taking a value
815
 
argument as a datetime.timedelta() but should store it as
816
 
milliseconds."""
 
819
    class MillisecondsPropertyValueArgument(PropertyValue):
 
820
        """Abstract class for PropertyValue taking a value argument as
 
821
a datetime.timedelta() but should store it as milliseconds."""
817
822
 
818
823
        @property
819
824
        def value_to_set(self):
825
830
            self._vts = int(round(value.total_seconds() * 1000))
826
831
 
827
832
 
828
 
    class SetTimeout(PropertySetterValueMilliseconds):
 
833
    class SetTimeout(MillisecondsPropertyValueArgument):
829
834
        propname = "Timeout"
830
835
 
831
836
 
832
 
    class SetExtendedTimeout(PropertySetterValueMilliseconds):
 
837
    class SetExtendedTimeout(MillisecondsPropertyValueArgument):
833
838
        propname = "ExtendedTimeout"
834
839
 
835
840
 
836
 
    class SetInterval(PropertySetterValueMilliseconds):
 
841
    class SetInterval(MillisecondsPropertyValueArgument):
837
842
        propname = "Interval"
838
843
 
839
844
 
840
 
    class SetApprovalDelay(PropertySetterValueMilliseconds):
 
845
    class SetApprovalDelay(MillisecondsPropertyValueArgument):
841
846
        propname = "ApprovalDelay"
842
847
 
843
848
 
844
 
    class SetApprovalDuration(PropertySetterValueMilliseconds):
 
849
    class SetApprovalDuration(MillisecondsPropertyValueArgument):
845
850
        propname = "ApprovalDuration"
846
851
 
847
852
 
881
886
 
882
887
 
883
888
class Test_string_to_delta(TestCaseWithAssertLogs):
884
 
    # Just test basic RFC 3339 functionality here, the doc string for
885
 
    # rfc3339_duration_to_delta() already has more comprehensive
886
 
    # tests, which is run by doctest.
887
 
 
888
 
    def test_rfc3339_zero_seconds(self):
889
 
        self.assertEqual(datetime.timedelta(),
890
 
                         string_to_delta("PT0S"))
891
 
 
892
 
    def test_rfc3339_zero_days(self):
893
 
        self.assertEqual(datetime.timedelta(), string_to_delta("P0D"))
894
 
 
895
 
    def test_rfc3339_one_second(self):
896
 
        self.assertEqual(datetime.timedelta(0, 1),
897
 
                         string_to_delta("PT1S"))
898
 
 
899
 
    def test_rfc3339_two_hours(self):
900
 
        self.assertEqual(datetime.timedelta(0, 7200),
901
 
                         string_to_delta("PT2H"))
 
889
    def test_handles_basic_rfc3339(self):
 
890
        self.assertEqual(string_to_delta("PT0S"),
 
891
                         datetime.timedelta())
 
892
        self.assertEqual(string_to_delta("P0D"),
 
893
                         datetime.timedelta())
 
894
        self.assertEqual(string_to_delta("PT1S"),
 
895
                         datetime.timedelta(0, 1))
 
896
        self.assertEqual(string_to_delta("PT2H"),
 
897
                         datetime.timedelta(0, 7200))
902
898
 
903
899
    def test_falls_back_to_pre_1_6_1_with_warning(self):
904
900
        with self.assertLogs(log, logging.WARNING):
905
901
            value = string_to_delta("2h")
906
 
        self.assertEqual(datetime.timedelta(0, 7200), value)
 
902
        self.assertEqual(value, datetime.timedelta(0, 7200))
907
903
 
908
904
 
909
905
class Test_check_option_syntax(unittest.TestCase):
952
948
        # Exit code from argparse is guaranteed to be "2".  Reference:
953
949
        # https://docs.python.org/3/library
954
950
        # /argparse.html#exiting-methods
955
 
        self.assertEqual(2, e.exception.code)
 
951
        self.assertEqual(e.exception.code, 2)
956
952
 
957
953
    @staticmethod
958
954
    @contextlib.contextmanager
959
955
    def redirect_stderr_to_devnull():
960
 
        old_stderr = sys.stderr
961
 
        with contextlib.closing(open(os.devnull, "w")) as null:
962
 
            sys.stderr = null
963
 
            try:
964
 
                yield
965
 
            finally:
966
 
                sys.stderr = old_stderr
 
956
        null = os.open(os.path.devnull, os.O_RDWR)
 
957
        stderrcopy = os.dup(sys.stderr.fileno())
 
958
        os.dup2(null, sys.stderr.fileno())
 
959
        os.close(null)
 
960
        try:
 
961
            yield
 
962
        finally:
 
963
            # restore stderr
 
964
            os.dup2(stderrcopy, sys.stderr.fileno())
 
965
            os.close(stderrcopy)
967
966
 
968
967
    def check_option_syntax(self, options):
969
968
        check_option_syntax(self.parser, options)
982
981
            options = self.parser.parse_args()
983
982
            setattr(options, action, value)
984
983
            options.verbose = True
985
 
            options.client = ["client"]
 
984
            options.client = ["foo"]
986
985
            with self.assertParseError():
987
986
                self.check_option_syntax(options)
988
987
 
1018
1017
        for action, value in self.actions.items():
1019
1018
            options = self.parser.parse_args()
1020
1019
            setattr(options, action, value)
1021
 
            options.client = ["client"]
 
1020
            options.client = ["foo"]
1022
1021
            self.check_option_syntax(options)
1023
1022
 
1024
1023
    def test_one_client_with_all_actions_except_is_enabled(self):
1027
1026
            if action == "is_enabled":
1028
1027
                continue
1029
1028
            setattr(options, action, value)
1030
 
        options.client = ["client"]
 
1029
        options.client = ["foo"]
1031
1030
        self.check_option_syntax(options)
1032
1031
 
1033
1032
    def test_two_clients_with_all_actions_except_is_enabled(self):
1036
1035
            if action == "is_enabled":
1037
1036
                continue
1038
1037
            setattr(options, action, value)
1039
 
        options.client = ["client1", "client2"]
 
1038
        options.client = ["foo", "barbar"]
1040
1039
        self.check_option_syntax(options)
1041
1040
 
1042
1041
    def test_two_clients_are_ok_with_actions_except_is_enabled(self):
1045
1044
                continue
1046
1045
            options = self.parser.parse_args()
1047
1046
            setattr(options, action, value)
1048
 
            options.client = ["client1", "client2"]
 
1047
            options.client = ["foo", "barbar"]
1049
1048
            self.check_option_syntax(options)
1050
1049
 
1051
1050
    def test_is_enabled_fails_without_client(self):
1057
1056
    def test_is_enabled_fails_with_two_clients(self):
1058
1057
        options = self.parser.parse_args()
1059
1058
        options.is_enabled = True
1060
 
        options.client = ["client1", "client2"]
 
1059
        options.client = ["foo", "barbar"]
1061
1060
        with self.assertParseError():
1062
1061
            self.check_option_syntax(options)
1063
1062
 
1080
1079
            def get_object(mockbus_self, busname, dbus_path):
1081
1080
                # Note that "self" is still the testcase instance,
1082
1081
                # this MockBus instance is in "mockbus_self".
1083
 
                self.assertEqual(dbus_busname, busname)
1084
 
                self.assertEqual(server_dbus_path, dbus_path)
 
1082
                self.assertEqual(busname, dbus_busname)
 
1083
                self.assertEqual(dbus_path, server_dbus_path)
1085
1084
                mockbus_self.called = True
1086
1085
                return mockbus_self
1087
1086
 
1090
1089
        self.assertTrue(mockbus.called)
1091
1090
 
1092
1091
    def test_logs_and_exits_on_dbus_error(self):
1093
 
        class FailingBusStub(object):
 
1092
        class MockBusFailing(object):
1094
1093
            def get_object(self, busname, dbus_path):
1095
1094
                raise dbus.exceptions.DBusException("Test")
1096
1095
 
1097
1096
        with self.assertLogs(log, logging.CRITICAL):
1098
1097
            with self.assertRaises(SystemExit) as e:
1099
 
                bus = get_mandos_dbus_object(bus=FailingBusStub())
 
1098
                bus = get_mandos_dbus_object(bus=MockBusFailing())
1100
1099
 
1101
1100
        if isinstance(e.exception.code, int):
1102
 
            self.assertNotEqual(0, e.exception.code)
 
1101
            self.assertNotEqual(e.exception.code, 0)
1103
1102
        else:
1104
1103
            self.assertIsNotNone(e.exception.code)
1105
1104
 
1106
1105
 
1107
1106
class Test_get_managed_objects(TestCaseWithAssertLogs):
1108
1107
    def test_calls_and_returns_GetManagedObjects(self):
1109
 
        managed_objects = {"/clients/client": { "Name": "client"}}
1110
 
        class ObjectManagerStub(object):
 
1108
        managed_objects = {"/clients/foo": { "Name": "foo"}}
 
1109
        class MockObjectManager(object):
1111
1110
            def GetManagedObjects(self):
1112
1111
                return managed_objects
1113
 
        retval = get_managed_objects(ObjectManagerStub())
 
1112
        retval = get_managed_objects(MockObjectManager())
1114
1113
        self.assertDictEqual(managed_objects, retval)
1115
1114
 
1116
1115
    def test_logs_and_exits_on_dbus_error(self):
1117
1116
        dbus_logger = logging.getLogger("dbus.proxies")
1118
1117
 
1119
 
        class ObjectManagerFailingStub(object):
 
1118
        class MockObjectManagerFailing(object):
1120
1119
            def GetManagedObjects(self):
1121
1120
                dbus_logger.error("Test")
1122
1121
                raise dbus.exceptions.DBusException("Test")
1133
1132
        try:
1134
1133
            with self.assertLogs(log, logging.CRITICAL) as watcher:
1135
1134
                with self.assertRaises(SystemExit) as e:
1136
 
                    get_managed_objects(ObjectManagerFailingStub())
 
1135
                    get_managed_objects(MockObjectManagerFailing())
1137
1136
        finally:
1138
1137
            dbus_logger.removeFilter(counting_handler)
1139
1138
 
1140
1139
        # Make sure the dbus logger was suppressed
1141
 
        self.assertEqual(0, counting_handler.count)
 
1140
        self.assertEqual(counting_handler.count, 0)
1142
1141
 
1143
1142
        # Test that the dbus_logger still works
1144
1143
        with self.assertLogs(dbus_logger, logging.ERROR):
1145
1144
            dbus_logger.error("Test")
1146
1145
 
1147
1146
        if isinstance(e.exception.code, int):
1148
 
            self.assertNotEqual(0, e.exception.code)
 
1147
            self.assertNotEqual(e.exception.code, 0)
1149
1148
        else:
1150
1149
            self.assertIsNotNone(e.exception.code)
1151
1150
 
1156
1155
        add_command_line_options(self.parser)
1157
1156
 
1158
1157
    def test_is_enabled(self):
1159
 
        self.assert_command_from_args(["--is-enabled", "client"],
 
1158
        self.assert_command_from_args(["--is-enabled", "foo"],
1160
1159
                                      command.IsEnabled)
1161
1160
 
1162
1161
    def assert_command_from_args(self, args, command_cls,
1166
1165
        options = self.parser.parse_args(args)
1167
1166
        check_option_syntax(self.parser, options)
1168
1167
        commands = commands_from_options(options)
1169
 
        self.assertEqual(1, len(commands))
 
1168
        self.assertEqual(len(commands), 1)
1170
1169
        command = commands[0]
1171
1170
        self.assertIsInstance(command, command_cls)
1172
1171
        for key, value in cmd_attrs.items():
1173
 
            self.assertEqual(value, getattr(command, key))
 
1172
            self.assertEqual(getattr(command, key), value)
1174
1173
 
1175
1174
    def test_is_enabled_short(self):
1176
 
        self.assert_command_from_args(["-V", "client"],
 
1175
        self.assert_command_from_args(["-V", "foo"],
1177
1176
                                      command.IsEnabled)
1178
1177
 
1179
1178
    def test_approve(self):
1180
 
        self.assert_command_from_args(["--approve", "client"],
 
1179
        self.assert_command_from_args(["--approve", "foo"],
1181
1180
                                      command.Approve)
1182
1181
 
1183
1182
    def test_approve_short(self):
1184
 
        self.assert_command_from_args(["-A", "client"],
1185
 
                                      command.Approve)
 
1183
        self.assert_command_from_args(["-A", "foo"], command.Approve)
1186
1184
 
1187
1185
    def test_deny(self):
1188
 
        self.assert_command_from_args(["--deny", "client"],
1189
 
                                      command.Deny)
 
1186
        self.assert_command_from_args(["--deny", "foo"], command.Deny)
1190
1187
 
1191
1188
    def test_deny_short(self):
1192
 
        self.assert_command_from_args(["-D", "client"], command.Deny)
 
1189
        self.assert_command_from_args(["-D", "foo"], command.Deny)
1193
1190
 
1194
1191
    def test_remove(self):
1195
 
        self.assert_command_from_args(["--remove", "client"],
 
1192
        self.assert_command_from_args(["--remove", "foo"],
1196
1193
                                      command.Remove)
1197
1194
 
1198
1195
    def test_deny_before_remove(self):
1199
1196
        options = self.parser.parse_args(["--deny", "--remove",
1200
 
                                          "client"])
 
1197
                                          "foo"])
1201
1198
        check_option_syntax(self.parser, options)
1202
1199
        commands = commands_from_options(options)
1203
 
        self.assertEqual(2, len(commands))
 
1200
        self.assertEqual(len(commands), 2)
1204
1201
        self.assertIsInstance(commands[0], command.Deny)
1205
1202
        self.assertIsInstance(commands[1], command.Remove)
1206
1203
 
1209
1206
                                          "--all"])
1210
1207
        check_option_syntax(self.parser, options)
1211
1208
        commands = commands_from_options(options)
1212
 
        self.assertEqual(2, len(commands))
 
1209
        self.assertEqual(len(commands), 2)
1213
1210
        self.assertIsInstance(commands[0], command.Deny)
1214
1211
        self.assertIsInstance(commands[1], command.Remove)
1215
1212
 
1216
1213
    def test_remove_short(self):
1217
 
        self.assert_command_from_args(["-r", "client"],
1218
 
                                      command.Remove)
 
1214
        self.assert_command_from_args(["-r", "foo"], command.Remove)
1219
1215
 
1220
1216
    def test_dump_json(self):
1221
1217
        self.assert_command_from_args(["--dump-json"],
1222
1218
                                      command.DumpJSON)
1223
1219
 
1224
1220
    def test_enable(self):
1225
 
        self.assert_command_from_args(["--enable", "client"],
 
1221
        self.assert_command_from_args(["--enable", "foo"],
1226
1222
                                      command.Enable)
1227
1223
 
1228
1224
    def test_enable_short(self):
1229
 
        self.assert_command_from_args(["-e", "client"],
1230
 
                                      command.Enable)
 
1225
        self.assert_command_from_args(["-e", "foo"], command.Enable)
1231
1226
 
1232
1227
    def test_disable(self):
1233
 
        self.assert_command_from_args(["--disable", "client"],
 
1228
        self.assert_command_from_args(["--disable", "foo"],
1234
1229
                                      command.Disable)
1235
1230
 
1236
1231
    def test_disable_short(self):
1237
 
        self.assert_command_from_args(["-d", "client"],
1238
 
                                      command.Disable)
 
1232
        self.assert_command_from_args(["-d", "foo"], command.Disable)
1239
1233
 
1240
1234
    def test_bump_timeout(self):
1241
 
        self.assert_command_from_args(["--bump-timeout", "client"],
 
1235
        self.assert_command_from_args(["--bump-timeout", "foo"],
1242
1236
                                      command.BumpTimeout)
1243
1237
 
1244
1238
    def test_bump_timeout_short(self):
1245
 
        self.assert_command_from_args(["-b", "client"],
 
1239
        self.assert_command_from_args(["-b", "foo"],
1246
1240
                                      command.BumpTimeout)
1247
1241
 
1248
1242
    def test_start_checker(self):
1249
 
        self.assert_command_from_args(["--start-checker", "client"],
 
1243
        self.assert_command_from_args(["--start-checker", "foo"],
1250
1244
                                      command.StartChecker)
1251
1245
 
1252
1246
    def test_stop_checker(self):
1253
 
        self.assert_command_from_args(["--stop-checker", "client"],
 
1247
        self.assert_command_from_args(["--stop-checker", "foo"],
1254
1248
                                      command.StopChecker)
1255
1249
 
1256
1250
    def test_approve_by_default(self):
1257
 
        self.assert_command_from_args(["--approve-by-default",
1258
 
                                       "client"],
 
1251
        self.assert_command_from_args(["--approve-by-default", "foo"],
1259
1252
                                      command.ApproveByDefault)
1260
1253
 
1261
1254
    def test_deny_by_default(self):
1262
 
        self.assert_command_from_args(["--deny-by-default", "client"],
 
1255
        self.assert_command_from_args(["--deny-by-default", "foo"],
1263
1256
                                      command.DenyByDefault)
1264
1257
 
1265
1258
    def test_checker(self):
1266
 
        self.assert_command_from_args(["--checker", ":", "client"],
 
1259
        self.assert_command_from_args(["--checker", ":", "foo"],
1267
1260
                                      command.SetChecker,
1268
1261
                                      value_to_set=":")
1269
1262
 
1270
1263
    def test_checker_empty(self):
1271
 
        self.assert_command_from_args(["--checker", "", "client"],
 
1264
        self.assert_command_from_args(["--checker", "", "foo"],
1272
1265
                                      command.SetChecker,
1273
1266
                                      value_to_set="")
1274
1267
 
1275
1268
    def test_checker_short(self):
1276
 
        self.assert_command_from_args(["-c", ":", "client"],
 
1269
        self.assert_command_from_args(["-c", ":", "foo"],
1277
1270
                                      command.SetChecker,
1278
1271
                                      value_to_set=":")
1279
1272
 
1280
1273
    def test_host(self):
1281
 
        self.assert_command_from_args(
1282
 
            ["--host", "client.example.org", "client"],
1283
 
            command.SetHost, value_to_set="client.example.org")
 
1274
        self.assert_command_from_args(["--host", "foo.example.org",
 
1275
                                       "foo"], command.SetHost,
 
1276
                                      value_to_set="foo.example.org")
1284
1277
 
1285
1278
    def test_host_short(self):
1286
 
        self.assert_command_from_args(
1287
 
            ["-H", "client.example.org", "client"], command.SetHost,
1288
 
            value_to_set="client.example.org")
 
1279
        self.assert_command_from_args(["-H", "foo.example.org",
 
1280
                                       "foo"], command.SetHost,
 
1281
                                      value_to_set="foo.example.org")
1289
1282
 
1290
1283
    def test_secret_devnull(self):
1291
1284
        self.assert_command_from_args(["--secret", os.path.devnull,
1292
 
                                       "client"], command.SetSecret,
 
1285
                                       "foo"], command.SetSecret,
1293
1286
                                      value_to_set=b"")
1294
1287
 
1295
1288
    def test_secret_tempfile(self):
1298
1291
            f.write(value)
1299
1292
            f.seek(0)
1300
1293
            self.assert_command_from_args(["--secret", f.name,
1301
 
                                           "client"],
1302
 
                                          command.SetSecret,
 
1294
                                           "foo"], command.SetSecret,
1303
1295
                                          value_to_set=value)
1304
1296
 
1305
1297
    def test_secret_devnull_short(self):
1306
 
        self.assert_command_from_args(["-s", os.path.devnull,
1307
 
                                       "client"], command.SetSecret,
 
1298
        self.assert_command_from_args(["-s", os.path.devnull, "foo"],
 
1299
                                      command.SetSecret,
1308
1300
                                      value_to_set=b"")
1309
1301
 
1310
1302
    def test_secret_tempfile_short(self):
1312
1304
            value = b"secret\0xyzzy\nbar"
1313
1305
            f.write(value)
1314
1306
            f.seek(0)
1315
 
            self.assert_command_from_args(["-s", f.name, "client"],
 
1307
            self.assert_command_from_args(["-s", f.name, "foo"],
1316
1308
                                          command.SetSecret,
1317
1309
                                          value_to_set=value)
1318
1310
 
1319
1311
    def test_timeout(self):
1320
 
        self.assert_command_from_args(["--timeout", "PT5M", "client"],
 
1312
        self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1321
1313
                                      command.SetTimeout,
1322
1314
                                      value_to_set=300000)
1323
1315
 
1324
1316
    def test_timeout_short(self):
1325
 
        self.assert_command_from_args(["-t", "PT5M", "client"],
 
1317
        self.assert_command_from_args(["-t", "PT5M", "foo"],
1326
1318
                                      command.SetTimeout,
1327
1319
                                      value_to_set=300000)
1328
1320
 
1329
1321
    def test_extended_timeout(self):
1330
1322
        self.assert_command_from_args(["--extended-timeout", "PT15M",
1331
 
                                       "client"],
 
1323
                                       "foo"],
1332
1324
                                      command.SetExtendedTimeout,
1333
1325
                                      value_to_set=900000)
1334
1326
 
1335
1327
    def test_interval(self):
1336
 
        self.assert_command_from_args(["--interval", "PT2M",
1337
 
                                       "client"], command.SetInterval,
 
1328
        self.assert_command_from_args(["--interval", "PT2M", "foo"],
 
1329
                                      command.SetInterval,
1338
1330
                                      value_to_set=120000)
1339
1331
 
1340
1332
    def test_interval_short(self):
1341
 
        self.assert_command_from_args(["-i", "PT2M", "client"],
 
1333
        self.assert_command_from_args(["-i", "PT2M", "foo"],
1342
1334
                                      command.SetInterval,
1343
1335
                                      value_to_set=120000)
1344
1336
 
1345
1337
    def test_approval_delay(self):
1346
1338
        self.assert_command_from_args(["--approval-delay", "PT30S",
1347
 
                                       "client"],
 
1339
                                       "foo"],
1348
1340
                                      command.SetApprovalDelay,
1349
1341
                                      value_to_set=30000)
1350
1342
 
1351
1343
    def test_approval_duration(self):
1352
1344
        self.assert_command_from_args(["--approval-duration", "PT1S",
1353
 
                                       "client"],
 
1345
                                       "foo"],
1354
1346
                                      command.SetApprovalDuration,
1355
1347
                                      value_to_set=1000)
1356
1348
 
1380
1372
                self.attributes["Name"] = name
1381
1373
                self.calls = []
1382
1374
            def Set(self, interface, propname, value, dbus_interface):
1383
 
                testcase.assertEqual(client_dbus_interface, interface)
1384
 
                testcase.assertEqual(dbus.PROPERTIES_IFACE,
1385
 
                                     dbus_interface)
 
1375
                testcase.assertEqual(interface, client_dbus_interface)
 
1376
                testcase.assertEqual(dbus_interface,
 
1377
                                     dbus.PROPERTIES_IFACE)
1386
1378
                self.attributes[propname] = value
 
1379
            def Get(self, interface, propname, dbus_interface):
 
1380
                testcase.assertEqual(interface, client_dbus_interface)
 
1381
                testcase.assertEqual(dbus_interface,
 
1382
                                     dbus.PROPERTIES_IFACE)
 
1383
                return self.attributes[propname]
1387
1384
            def Approve(self, approve, dbus_interface):
1388
 
                testcase.assertEqual(client_dbus_interface,
1389
 
                                     dbus_interface)
 
1385
                testcase.assertEqual(dbus_interface,
 
1386
                                     client_dbus_interface)
1390
1387
                self.calls.append(("Approve", (approve,
1391
1388
                                               dbus_interface)))
1392
1389
        self.client = MockClient(
1439
1436
            LastCheckerStatus=-2)
1440
1437
        self.clients =  collections.OrderedDict(
1441
1438
            [
1442
 
                (self.client.__dbus_object_path__,
1443
 
                 self.client.attributes),
1444
 
                (self.other_client.__dbus_object_path__,
1445
 
                 self.other_client.attributes),
 
1439
                ("/clients/foo", self.client.attributes),
 
1440
                ("/clients/barbar", self.other_client.attributes),
1446
1441
            ])
1447
 
        self.one_client = {self.client.__dbus_object_path__:
1448
 
                           self.client.attributes}
 
1442
        self.one_client = {"/clients/foo": self.client.attributes}
1449
1443
 
1450
1444
    @property
1451
1445
    def bus(self):
1452
 
        class MockBus(object):
 
1446
        class Bus(object):
1453
1447
            @staticmethod
1454
1448
            def get_object(client_bus_name, path):
1455
 
                self.assertEqual(dbus_busname, client_bus_name)
1456
 
                # Note: "self" here is the TestCmd instance, not the
1457
 
                # MockBus instance, since this is a static method!
1458
 
                if path == self.client.__dbus_object_path__:
1459
 
                    return self.client
1460
 
                elif path == self.other_client.__dbus_object_path__:
1461
 
                    return self.other_client
1462
 
        return MockBus()
 
1449
                self.assertEqual(client_bus_name, dbus_busname)
 
1450
                return {
 
1451
                    # Note: "self" here is the TestCmd instance, not
 
1452
                    # the Bus instance, since this is a static method!
 
1453
                    "/clients/foo": self.client,
 
1454
                    "/clients/barbar": self.other_client,
 
1455
                }[path]
 
1456
        return Bus()
1463
1457
 
1464
1458
 
1465
1459
class TestBaseCommands(TestCommand):
1466
1460
 
1467
 
    def test_IsEnabled_exits_successfully(self):
 
1461
    def test_is_enabled(self):
 
1462
        self.assertTrue(all(command.IsEnabled().is_enabled(client,
 
1463
                                                      properties)
 
1464
                            for client, properties
 
1465
                            in self.clients.items()))
 
1466
 
 
1467
    def test_is_enabled_run_exits_successfully(self):
1468
1468
        with self.assertRaises(SystemExit) as e:
1469
1469
            command.IsEnabled().run(self.one_client)
1470
1470
        if e.exception.code is not None:
1471
 
            self.assertEqual(0, e.exception.code)
 
1471
            self.assertEqual(e.exception.code, 0)
1472
1472
        else:
1473
1473
            self.assertIsNone(e.exception.code)
1474
1474
 
1475
 
    def test_IsEnabled_exits_with_failure(self):
 
1475
    def test_is_enabled_run_exits_with_failure(self):
1476
1476
        self.client.attributes["Enabled"] = dbus.Boolean(False)
1477
1477
        with self.assertRaises(SystemExit) as e:
1478
1478
            command.IsEnabled().run(self.one_client)
1479
1479
        if isinstance(e.exception.code, int):
1480
 
            self.assertNotEqual(0, e.exception.code)
 
1480
            self.assertNotEqual(e.exception.code, 0)
1481
1481
        else:
1482
1482
            self.assertIsNotNone(e.exception.code)
1483
1483
 
1484
 
    def test_Approve(self):
 
1484
    def test_approve(self):
1485
1485
        command.Approve().run(self.clients, self.bus)
1486
1486
        for clientpath in self.clients:
1487
1487
            client = self.bus.get_object(dbus_busname, clientpath)
1488
1488
            self.assertIn(("Approve", (True, client_dbus_interface)),
1489
1489
                          client.calls)
1490
1490
 
1491
 
    def test_Deny(self):
 
1491
    def test_deny(self):
1492
1492
        command.Deny().run(self.clients, self.bus)
1493
1493
        for clientpath in self.clients:
1494
1494
            client = self.bus.get_object(dbus_busname, clientpath)
1495
1495
            self.assertIn(("Approve", (False, client_dbus_interface)),
1496
1496
                          client.calls)
1497
1497
 
1498
 
    def test_Remove(self):
1499
 
        class MandosSpy(object):
 
1498
    def test_remove(self):
 
1499
        class MockMandos(object):
1500
1500
            def __init__(self):
1501
1501
                self.calls = []
1502
1502
            def RemoveClient(self, dbus_path):
1503
1503
                self.calls.append(("RemoveClient", (dbus_path,)))
1504
 
        mandos = MandosSpy()
 
1504
        mandos = MockMandos()
 
1505
        super(TestBaseCommands, self).setUp()
1505
1506
        command.Remove().run(self.clients, self.bus, mandos)
 
1507
        self.assertEqual(len(mandos.calls), 2)
1506
1508
        for clientpath in self.clients:
1507
1509
            self.assertIn(("RemoveClient", (clientpath,)),
1508
1510
                          mandos.calls)
1559
1561
    }
1560
1562
 
1561
1563
    def test_DumpJSON_normal(self):
1562
 
        with self.capture_stdout_to_buffer() as buffer:
1563
 
            command.DumpJSON().run(self.clients)
1564
 
        json_data = json.loads(buffer.getvalue())
1565
 
        self.assertDictEqual(self.expected_json, json_data)
1566
 
 
1567
 
    @staticmethod
1568
 
    @contextlib.contextmanager
1569
 
    def capture_stdout_to_buffer():
1570
 
        capture_buffer = io.StringIO()
1571
 
        old_stdout = sys.stdout
1572
 
        sys.stdout = capture_buffer
1573
 
        try:
1574
 
            yield capture_buffer
1575
 
        finally:
1576
 
            sys.stdout = old_stdout
 
1564
        output = command.DumpJSON().output(self.clients.values())
 
1565
        json_data = json.loads(output)
 
1566
        self.assertDictEqual(json_data, self.expected_json)
1577
1567
 
1578
1568
    def test_DumpJSON_one_client(self):
1579
 
        with self.capture_stdout_to_buffer() as buffer:
1580
 
            command.DumpJSON().run(self.one_client)
1581
 
        json_data = json.loads(buffer.getvalue())
 
1569
        output = command.DumpJSON().output(self.one_client.values())
 
1570
        json_data = json.loads(output)
1582
1571
        expected_json = {"foo": self.expected_json["foo"]}
1583
 
        self.assertDictEqual(expected_json, json_data)
 
1572
        self.assertDictEqual(json_data, expected_json)
1584
1573
 
1585
1574
    def test_PrintTable_normal(self):
1586
 
        with self.capture_stdout_to_buffer() as buffer:
1587
 
            command.PrintTable().run(self.clients)
 
1575
        output = command.PrintTable().output(self.clients.values())
1588
1576
        expected_output = "\n".join((
1589
1577
            "Name   Enabled Timeout  Last Successful Check",
1590
1578
            "foo    Yes     00:05:00 2019-02-03T00:00:00  ",
1591
1579
            "barbar Yes     00:05:00 2019-02-04T00:00:00  ",
1592
 
        )) + "\n"
1593
 
        self.assertEqual(expected_output, buffer.getvalue())
 
1580
        ))
 
1581
        self.assertEqual(output, expected_output)
1594
1582
 
1595
1583
    def test_PrintTable_verbose(self):
1596
 
        with self.capture_stdout_to_buffer() as buffer:
1597
 
            command.PrintTable(verbose=True).run(self.clients)
 
1584
        output = command.PrintTable(verbose=True).output(
 
1585
            self.clients.values())
1598
1586
        columns = (
1599
1587
            (
1600
1588
                "Name   ",
1682
1670
            )
1683
1671
        )
1684
1672
        num_lines = max(len(rows) for rows in columns)
1685
 
        expected_output = ("\n".join("".join(rows[line]
1686
 
                                             for rows in columns)
1687
 
                                     for line in range(num_lines))
1688
 
                           + "\n")
1689
 
        self.assertEqual(expected_output, buffer.getvalue())
 
1673
        expected_output = "\n".join("".join(rows[line]
 
1674
                                            for rows in columns)
 
1675
                                    for line in range(num_lines))
 
1676
        self.assertEqual(output, expected_output)
1690
1677
 
1691
1678
    def test_PrintTable_one_client(self):
1692
 
        with self.capture_stdout_to_buffer() as buffer:
1693
 
            command.PrintTable().run(self.one_client)
 
1679
        output = command.PrintTable().output(self.one_client.values())
1694
1680
        expected_output = "\n".join((
1695
1681
            "Name Enabled Timeout  Last Successful Check",
1696
1682
            "foo  Yes     00:05:00 2019-02-03T00:00:00  ",
1697
 
        )) + "\n"
1698
 
        self.assertEqual(expected_output, buffer.getvalue())
1699
 
 
1700
 
 
1701
 
class TestPropertySetterCmd(TestCommand):
1702
 
    """Abstract class for tests of command.PropertySetter classes"""
 
1683
        ))
 
1684
        self.assertEqual(output, expected_output)
 
1685
 
 
1686
 
 
1687
class TestPropertyCmd(TestCommand):
 
1688
    """Abstract class for tests of command.Property classes"""
1703
1689
    def runTest(self):
1704
1690
        if not hasattr(self, "command"):
1705
1691
            return
1710
1696
            for clientpath in self.clients:
1711
1697
                client = self.bus.get_object(dbus_busname, clientpath)
1712
1698
                old_value = client.attributes[self.propname]
 
1699
                self.assertNotIsInstance(old_value, self.Unique)
1713
1700
                client.attributes[self.propname] = self.Unique()
1714
1701
            self.run_command(value_to_set, self.clients)
1715
1702
            for clientpath in self.clients:
1716
1703
                client = self.bus.get_object(dbus_busname, clientpath)
1717
1704
                value = client.attributes[self.propname]
1718
1705
                self.assertNotIsInstance(value, self.Unique)
1719
 
                self.assertEqual(value_to_get, value)
 
1706
                self.assertEqual(value, value_to_get)
1720
1707
 
1721
1708
    class Unique(object):
1722
1709
        """Class for objects which exist only to be unique objects,
1726
1713
        self.command().run(clients, self.bus)
1727
1714
 
1728
1715
 
1729
 
class TestEnableCmd(TestPropertySetterCmd):
 
1716
class TestEnableCmd(TestPropertyCmd):
1730
1717
    command = command.Enable
1731
1718
    propname = "Enabled"
1732
1719
    values_to_set = [dbus.Boolean(True)]
1733
1720
 
1734
1721
 
1735
 
class TestDisableCmd(TestPropertySetterCmd):
 
1722
class TestDisableCmd(TestPropertyCmd):
1736
1723
    command = command.Disable
1737
1724
    propname = "Enabled"
1738
1725
    values_to_set = [dbus.Boolean(False)]
1739
1726
 
1740
1727
 
1741
 
class TestBumpTimeoutCmd(TestPropertySetterCmd):
 
1728
class TestBumpTimeoutCmd(TestPropertyCmd):
1742
1729
    command = command.BumpTimeout
1743
1730
    propname = "LastCheckedOK"
1744
1731
    values_to_set = [""]
1745
1732
 
1746
1733
 
1747
 
class TestStartCheckerCmd(TestPropertySetterCmd):
 
1734
class TestStartCheckerCmd(TestPropertyCmd):
1748
1735
    command = command.StartChecker
1749
1736
    propname = "CheckerRunning"
1750
1737
    values_to_set = [dbus.Boolean(True)]
1751
1738
 
1752
1739
 
1753
 
class TestStopCheckerCmd(TestPropertySetterCmd):
 
1740
class TestStopCheckerCmd(TestPropertyCmd):
1754
1741
    command = command.StopChecker
1755
1742
    propname = "CheckerRunning"
1756
1743
    values_to_set = [dbus.Boolean(False)]
1757
1744
 
1758
1745
 
1759
 
class TestApproveByDefaultCmd(TestPropertySetterCmd):
 
1746
class TestApproveByDefaultCmd(TestPropertyCmd):
1760
1747
    command = command.ApproveByDefault
1761
1748
    propname = "ApprovedByDefault"
1762
1749
    values_to_set = [dbus.Boolean(True)]
1763
1750
 
1764
1751
 
1765
 
class TestDenyByDefaultCmd(TestPropertySetterCmd):
 
1752
class TestDenyByDefaultCmd(TestPropertyCmd):
1766
1753
    command = command.DenyByDefault
1767
1754
    propname = "ApprovedByDefault"
1768
1755
    values_to_set = [dbus.Boolean(False)]
1769
1756
 
1770
1757
 
1771
 
class TestPropertySetterValueCmd(TestPropertySetterCmd):
1772
 
    """Abstract class for tests of PropertySetterValueCmd classes"""
 
1758
class TestPropertyValueCmd(TestPropertyCmd):
 
1759
    """Abstract class for tests of PropertyValueCmd classes"""
1773
1760
 
1774
1761
    def runTest(self):
1775
 
        if type(self) is TestPropertySetterValueCmd:
 
1762
        if type(self) is TestPropertyValueCmd:
1776
1763
            return
1777
 
        return super(TestPropertySetterValueCmd, self).runTest()
 
1764
        return super(TestPropertyValueCmd, self).runTest()
1778
1765
 
1779
1766
    def run_command(self, value, clients):
1780
1767
        self.command(value).run(clients, self.bus)
1781
1768
 
1782
1769
 
1783
 
class TestSetCheckerCmd(TestPropertySetterValueCmd):
 
1770
class TestSetCheckerCmd(TestPropertyValueCmd):
1784
1771
    command = command.SetChecker
1785
1772
    propname = "Checker"
1786
1773
    values_to_set = ["", ":", "fping -q -- %s"]
1787
1774
 
1788
1775
 
1789
 
class TestSetHostCmd(TestPropertySetterValueCmd):
 
1776
class TestSetHostCmd(TestPropertyValueCmd):
1790
1777
    command = command.SetHost
1791
1778
    propname = "Host"
1792
 
    values_to_set = ["192.0.2.3", "client.example.org"]
1793
 
 
1794
 
 
1795
 
class TestSetSecretCmd(TestPropertySetterValueCmd):
 
1779
    values_to_set = ["192.0.2.3", "foo.example.org"]
 
1780
 
 
1781
 
 
1782
class TestSetSecretCmd(TestPropertyValueCmd):
1796
1783
    command = command.SetSecret
1797
1784
    propname = "Secret"
1798
1785
    values_to_set = [io.BytesIO(b""),
1799
1786
                     io.BytesIO(b"secret\0xyzzy\nbar")]
1800
 
    values_to_get = [f.getvalue() for f in values_to_set]
1801
 
 
1802
 
 
1803
 
class TestSetTimeoutCmd(TestPropertySetterValueCmd):
 
1787
    values_to_get = [b"", b"secret\0xyzzy\nbar"]
 
1788
 
 
1789
 
 
1790
class TestSetTimeoutCmd(TestPropertyValueCmd):
1804
1791
    command = command.SetTimeout
1805
1792
    propname = "Timeout"
1806
1793
    values_to_set = [datetime.timedelta(),
1808
1795
                     datetime.timedelta(seconds=1),
1809
1796
                     datetime.timedelta(weeks=1),
1810
1797
                     datetime.timedelta(weeks=52)]
1811
 
    values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
1812
 
 
1813
 
 
1814
 
class TestSetExtendedTimeoutCmd(TestPropertySetterValueCmd):
 
1798
    values_to_get = [0, 300000, 1000, 604800000, 31449600000]
 
1799
 
 
1800
 
 
1801
class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):
1815
1802
    command = command.SetExtendedTimeout
1816
1803
    propname = "ExtendedTimeout"
1817
1804
    values_to_set = [datetime.timedelta(),
1819
1806
                     datetime.timedelta(seconds=1),
1820
1807
                     datetime.timedelta(weeks=1),
1821
1808
                     datetime.timedelta(weeks=52)]
1822
 
    values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
1823
 
 
1824
 
 
1825
 
class TestSetIntervalCmd(TestPropertySetterValueCmd):
 
1809
    values_to_get = [0, 300000, 1000, 604800000, 31449600000]
 
1810
 
 
1811
 
 
1812
class TestSetIntervalCmd(TestPropertyValueCmd):
1826
1813
    command = command.SetInterval
1827
1814
    propname = "Interval"
1828
1815
    values_to_set = [datetime.timedelta(),
1830
1817
                     datetime.timedelta(seconds=1),
1831
1818
                     datetime.timedelta(weeks=1),
1832
1819
                     datetime.timedelta(weeks=52)]
1833
 
    values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
1834
 
 
1835
 
 
1836
 
class TestSetApprovalDelayCmd(TestPropertySetterValueCmd):
 
1820
    values_to_get = [0, 300000, 1000, 604800000, 31449600000]
 
1821
 
 
1822
 
 
1823
class TestSetApprovalDelayCmd(TestPropertyValueCmd):
1837
1824
    command = command.SetApprovalDelay
1838
1825
    propname = "ApprovalDelay"
1839
1826
    values_to_set = [datetime.timedelta(),
1841
1828
                     datetime.timedelta(seconds=1),
1842
1829
                     datetime.timedelta(weeks=1),
1843
1830
                     datetime.timedelta(weeks=52)]
1844
 
    values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
1845
 
 
1846
 
 
1847
 
class TestSetApprovalDurationCmd(TestPropertySetterValueCmd):
 
1831
    values_to_get = [0, 300000, 1000, 604800000, 31449600000]
 
1832
 
 
1833
 
 
1834
class TestSetApprovalDurationCmd(TestPropertyValueCmd):
1848
1835
    command = command.SetApprovalDuration
1849
1836
    propname = "ApprovalDuration"
1850
1837
    values_to_set = [datetime.timedelta(),
1852
1839
                     datetime.timedelta(seconds=1),
1853
1840
                     datetime.timedelta(weeks=1),
1854
1841
                     datetime.timedelta(weeks=52)]
1855
 
    values_to_get = [dt.total_seconds()*1000 for dt in values_to_set]
 
1842
    values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1856
1843
 
1857
1844
 
1858
1845