/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-23 14:59:54 UTC
  • Revision ID: teddy@recompile.se-20190323145954-latqk2bg6xt989d5
mandos: Refactor; use classes, not instances, as namespaces

* mandos (Avahi): Rename to "avahi".
  (avahi.string_array_to_txt_array): Make static method.
  (GnuTLS): Rename to "gnutls".
  (gnutls.__init__): Remove; move version check up to class body.
  (gnutls.Error): Don't use class name anymore.

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