389
316
@value_to_set.setter
390
317
def value_to_set(self, value):
391
318
"""When setting, convert value to a datetime.timedelta"""
392
self._vts = string_to_delta(value).total_seconds() * 1000
319
self._vts = int(round(value.total_seconds() * 1000))
394
321
# Actual (non-abstract) command classes
396
323
class PrintTableCmd(PrintCmd):
397
324
def __init__(self, verbose=False):
398
325
self.verbose = verbose
399
327
def output(self, clients):
328
default_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
329
keywords = default_keywords
401
331
keywords = self.all_keywords
403
keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
404
return str(TableOfClients(clients.values(), keywords))
332
return str(self.TableOfClients(clients.values(), keywords))
334
class TableOfClients(object):
337
"Enabled": "Enabled",
338
"Timeout": "Timeout",
339
"LastCheckedOK": "Last Successful Check",
340
"LastApprovalRequest": "Last Approval Request",
341
"Created": "Created",
342
"Interval": "Interval",
344
"Fingerprint": "Fingerprint",
346
"CheckerRunning": "Check Is Running",
347
"LastEnabled": "Last Enabled",
348
"ApprovalPending": "Approval Is Pending",
349
"ApprovedByDefault": "Approved By Default",
350
"ApprovalDelay": "Approval Delay",
351
"ApprovalDuration": "Approval Duration",
352
"Checker": "Checker",
353
"ExtendedTimeout": "Extended Timeout",
354
"Expires": "Expires",
355
"LastCheckerStatus": "Last Checker Status",
358
def __init__(self, clients, keywords, tableheaders=None):
359
self.clients = clients
360
self.keywords = keywords
361
if tableheaders is not None:
362
self.tableheaders = tableheaders
365
return "\n".join(self.rows())
367
if sys.version_info.major == 2:
368
__unicode__ = __str__
370
return str(self).encode(locale.getpreferredencoding())
373
format_string = self.row_formatting_string()
374
rows = [self.header_line(format_string)]
375
rows.extend(self.client_line(client, format_string)
376
for client in self.clients)
379
def row_formatting_string(self):
380
"Format string used to format table rows"
381
return " ".join("{{{key}:{width}}}".format(
382
width=max(len(self.tableheaders[key]),
383
*(len(self.string_from_client(client, key))
384
for client in self.clients)),
386
for key in self.keywords)
388
def string_from_client(self, client, key):
389
return self.valuetostring(client[key], key)
392
def valuetostring(value, keyword):
393
if isinstance(value, dbus.Boolean):
394
return "Yes" if value else "No"
395
if keyword in ("Timeout", "Interval", "ApprovalDelay",
396
"ApprovalDuration", "ExtendedTimeout"):
397
return milliseconds_to_string(value)
400
def header_line(self, format_string):
401
return format_string.format(**self.tableheaders)
403
def client_line(self, client, format_string):
404
return format_string.format(
405
**{key: self.string_from_client(client, key)
406
for key in self.keywords})
406
410
class DumpJSONCmd(PrintCmd):
407
411
def output(self, clients):
775
801
self.__dbus_object_path__ = "objpath_{}".format(name)
776
802
self.attributes = attributes
777
803
self.attributes["Name"] = name
778
def Set(interface, property, value,
779
properties_interface):
805
def Set(self, interface, property, value, dbus_interface):
780
806
testcase.assertEqual(interface, client_interface)
781
testcase.assertEqual(properties_interface,
807
testcase.assertEqual(dbus_interface,
782
808
dbus.PROPERTIES_IFACE)
783
809
self.attributes[property] = value
784
def Get(interface, property, properties_interface):
810
def Get(self, interface, property, dbus_interface):
785
811
testcase.assertEqual(interface, client_interface)
786
testcase.assertEqual(properties_interface,
812
testcase.assertEqual(dbus_interface,
787
813
dbus.PROPERTIES_IFACE)
788
814
return self.attributes[property]
789
def __getitem__(self, key):
790
return self.attributes[key]
791
self.clients = collections.OrderedDict([
795
KeyID=("92ed150794387c03ce684574b1139a65"
796
"94a34f895daaaf09fd8ea90a27cddb12"),
798
Host="foo.example.org",
799
Enabled=dbus.Boolean(True),
801
LastCheckedOK="2019-02-03T00:00:00",
802
Created="2019-01-02T00:00:00",
804
Fingerprint=("778827225BA7DE539C5A"
805
"7CFA59CFF7CDBD9A5920"),
806
CheckerRunning=dbus.Boolean(False),
807
LastEnabled="2019-01-03T00:00:00",
808
ApprovalPending=dbus.Boolean(False),
809
ApprovedByDefault=dbus.Boolean(True),
810
LastApprovalRequest="",
812
ApprovalDuration=1000,
813
Checker="fping -q -- %(host)s",
814
ExtendedTimeout=900000,
815
Expires="2019-02-04T00:00:00",
816
LastCheckerStatus=0)),
820
KeyID=("0558568eedd67d622f5c83b35a115f79"
821
"6ab612cff5ad227247e46c2b020f441c"),
824
Enabled=dbus.Boolean(True),
826
LastCheckedOK="2019-02-04T00:00:00",
827
Created="2019-01-03T00:00:00",
829
Fingerprint=("3E393AEAEFB84C7E89E2"
830
"F547B3A107558FCA3A27"),
831
CheckerRunning=dbus.Boolean(True),
832
LastEnabled="2019-01-04T00:00:00",
833
ApprovalPending=dbus.Boolean(False),
834
ApprovedByDefault=dbus.Boolean(False),
835
LastApprovalRequest="2019-01-03T00:00:00",
837
ApprovalDuration=1000,
839
ExtendedTimeout=900000,
840
Expires="2019-02-05T00:00:00",
841
LastCheckerStatus=-2)),
815
def Approve(self, approve, dbus_interface):
816
testcase.assertEqual(dbus_interface, client_interface)
817
self.calls.append(("Approve", (approve,
819
self.client = MockClient(
821
KeyID=("92ed150794387c03ce684574b1139a65"
822
"94a34f895daaaf09fd8ea90a27cddb12"),
824
Host="foo.example.org",
825
Enabled=dbus.Boolean(True),
827
LastCheckedOK="2019-02-03T00:00:00",
828
Created="2019-01-02T00:00:00",
830
Fingerprint=("778827225BA7DE539C5A"
831
"7CFA59CFF7CDBD9A5920"),
832
CheckerRunning=dbus.Boolean(False),
833
LastEnabled="2019-01-03T00:00:00",
834
ApprovalPending=dbus.Boolean(False),
835
ApprovedByDefault=dbus.Boolean(True),
836
LastApprovalRequest="",
838
ApprovalDuration=1000,
839
Checker="fping -q -- %(host)s",
840
ExtendedTimeout=900000,
841
Expires="2019-02-04T00:00:00",
843
self.other_client = MockClient(
845
KeyID=("0558568eedd67d622f5c83b35a115f79"
846
"6ab612cff5ad227247e46c2b020f441c"),
849
Enabled=dbus.Boolean(True),
851
LastCheckedOK="2019-02-04T00:00:00",
852
Created="2019-01-03T00:00:00",
854
Fingerprint=("3E393AEAEFB84C7E89E2"
855
"F547B3A107558FCA3A27"),
856
CheckerRunning=dbus.Boolean(True),
857
LastEnabled="2019-01-04T00:00:00",
858
ApprovalPending=dbus.Boolean(False),
859
ApprovedByDefault=dbus.Boolean(False),
860
LastApprovalRequest="2019-01-03T00:00:00",
862
ApprovalDuration=1000,
864
ExtendedTimeout=900000,
865
Expires="2019-02-05T00:00:00",
866
LastCheckerStatus=-2)
867
self.clients = collections.OrderedDict(
869
(self.client, self.client.attributes),
870
(self.other_client, self.other_client.attributes),
872
self.one_client = {self.client: self.client.attributes}
844
874
class TestPrintTableCmd(TestCmd):
845
875
def test_normal(self):
860
890
self.assertEqual(output, expected_output)
861
891
def test_one_client(self):
862
output = PrintTableCmd().output({"foo": self.clients["foo"]})
892
output = PrintTableCmd().output(self.one_client)
863
893
expected_output = """
864
894
Name Enabled Timeout Last Successful Check
865
895
foo Yes 00:05:00 2019-02-03T00:00:00
867
897
self.assertEqual(output, expected_output)
899
class TestDumpJSONCmd(TestCmd):
901
self.expected_json = {
904
"KeyID": ("92ed150794387c03ce684574b1139a65"
905
"94a34f895daaaf09fd8ea90a27cddb12"),
906
"Host": "foo.example.org",
909
"LastCheckedOK": "2019-02-03T00:00:00",
910
"Created": "2019-01-02T00:00:00",
912
"Fingerprint": ("778827225BA7DE539C5A"
913
"7CFA59CFF7CDBD9A5920"),
914
"CheckerRunning": False,
915
"LastEnabled": "2019-01-03T00:00:00",
916
"ApprovalPending": False,
917
"ApprovedByDefault": True,
918
"LastApprovalRequest": "",
920
"ApprovalDuration": 1000,
921
"Checker": "fping -q -- %(host)s",
922
"ExtendedTimeout": 900000,
923
"Expires": "2019-02-04T00:00:00",
924
"LastCheckerStatus": 0,
928
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
929
"6ab612cff5ad227247e46c2b020f441c"),
933
"LastCheckedOK": "2019-02-04T00:00:00",
934
"Created": "2019-01-03T00:00:00",
936
"Fingerprint": ("3E393AEAEFB84C7E89E2"
937
"F547B3A107558FCA3A27"),
938
"CheckerRunning": True,
939
"LastEnabled": "2019-01-04T00:00:00",
940
"ApprovalPending": False,
941
"ApprovedByDefault": False,
942
"LastApprovalRequest": "2019-01-03T00:00:00",
943
"ApprovalDelay": 30000,
944
"ApprovalDuration": 1000,
946
"ExtendedTimeout": 900000,
947
"Expires": "2019-02-05T00:00:00",
948
"LastCheckerStatus": -2,
951
return super(TestDumpJSONCmd, self).setUp()
952
def test_normal(self):
953
json_data = json.loads(DumpJSONCmd().output(self.clients))
954
self.assertDictEqual(json_data, self.expected_json)
955
def test_one_client(self):
956
clients = self.one_client
957
json_data = json.loads(DumpJSONCmd().output(clients))
958
expected_json = {"foo": self.expected_json["foo"]}
959
self.assertDictEqual(json_data, expected_json)
961
class TestIsEnabledCmd(TestCmd):
962
def test_is_enabled(self):
963
self.assertTrue(all(IsEnabledCmd().is_enabled(client, properties)
964
for client, properties in self.clients.items()))
965
def test_is_enabled_run_exits_successfully(self):
966
with self.assertRaises(SystemExit) as e:
967
IsEnabledCmd().run(None, self.one_client)
968
if e.exception.code is not None:
969
self.assertEqual(e.exception.code, 0)
971
self.assertIsNone(e.exception.code)
972
def test_is_enabled_run_exits_with_failure(self):
973
self.client.attributes["Enabled"] = dbus.Boolean(False)
974
with self.assertRaises(SystemExit) as e:
975
IsEnabledCmd().run(None, self.one_client)
976
if isinstance(e.exception.code, int):
977
self.assertNotEqual(e.exception.code, 0)
979
self.assertIsNotNone(e.exception.code)
981
class TestRemoveCmd(TestCmd):
982
def test_remove(self):
983
class MockMandos(object):
986
def RemoveClient(self, dbus_path):
987
self.calls.append(("RemoveClient", (dbus_path,)))
988
mandos = MockMandos()
989
super(TestRemoveCmd, self).setUp()
990
RemoveCmd().run(mandos, self.clients)
991
self.assertEqual(len(mandos.calls), 2)
992
for client in self.clients:
993
self.assertIn(("RemoveClient",
994
(client.__dbus_object_path__,)),
997
class TestApproveCmd(TestCmd):
998
def test_approve(self):
999
ApproveCmd().run(None, self.clients)
1000
for client in self.clients:
1001
self.assertIn(("Approve", (True, client_interface)),
1004
class TestDenyCmd(TestCmd):
1005
def test_deny(self):
1006
DenyCmd().run(None, self.clients)
1007
for client in self.clients:
1008
self.assertIn(("Approve", (False, client_interface)),
1011
class TestEnableCmd(TestCmd):
1012
def test_enable(self):
1013
for client in self.clients:
1014
client.attributes["Enabled"] = False
1016
EnableCmd().run(None, self.clients)
1018
for client in self.clients:
1019
self.assertTrue(client.attributes["Enabled"])
1021
class TestDisableCmd(TestCmd):
1022
def test_disable(self):
1023
DisableCmd().run(None, self.clients)
1025
for client in self.clients:
1026
self.assertFalse(client.attributes["Enabled"])
1028
class Unique(object):
1029
"""Class for objects which exist only to be unique objects, since
1030
unittest.mock.sentinel only exists in Python 3.3"""
1032
class TestPropertyCmd(TestCmd):
1033
"""Abstract class for tests of PropertyCmd classes"""
1035
if not hasattr(self, "command"):
1037
values_to_get = getattr(self, "values_to_get",
1039
for value_to_set, value_to_get in zip(self.values_to_set,
1041
for client in self.clients:
1042
old_value = client.attributes[self.property]
1043
self.assertNotIsInstance(old_value, Unique)
1044
client.attributes[self.property] = Unique()
1045
self.run_command(value_to_set, self.clients)
1046
for client in self.clients:
1047
value = client.attributes[self.property]
1048
self.assertNotIsInstance(value, Unique)
1049
self.assertEqual(value, value_to_get)
1050
def run_command(self, value, clients):
1051
self.command().run(None, clients)
1053
class TestBumpTimeoutCmd(TestPropertyCmd):
1054
command = BumpTimeoutCmd
1055
property = "LastCheckedOK"
1056
values_to_set = [""]
1058
class TestStartCheckerCmd(TestPropertyCmd):
1059
command = StartCheckerCmd
1060
property = "CheckerRunning"
1061
values_to_set = [dbus.Boolean(True)]
1063
class TestStopCheckerCmd(TestPropertyCmd):
1064
command = StopCheckerCmd
1065
property = "CheckerRunning"
1066
values_to_set = [dbus.Boolean(False)]
1068
class TestApproveByDefaultCmd(TestPropertyCmd):
1069
command = ApproveByDefaultCmd
1070
property = "ApprovedByDefault"
1071
values_to_set = [dbus.Boolean(True)]
1073
class TestDenyByDefaultCmd(TestPropertyCmd):
1074
command = DenyByDefaultCmd
1075
property = "ApprovedByDefault"
1076
values_to_set = [dbus.Boolean(False)]
1078
class TestValueArgumentPropertyCmd(TestPropertyCmd):
1079
"""Abstract class for tests of PropertyCmd classes using the
1080
ValueArgumentMixIn"""
1082
if type(self) is TestValueArgumentPropertyCmd:
1084
return super(TestValueArgumentPropertyCmd, self).runTest()
1085
def run_command(self, value, clients):
1086
self.command(value).run(None, clients)
1088
class TestSetCheckerCmd(TestValueArgumentPropertyCmd):
1089
command = SetCheckerCmd
1090
property = "Checker"
1091
values_to_set = ["", ":", "fping -q -- %s"]
1093
class TestSetHostCmd(TestValueArgumentPropertyCmd):
1094
command = SetHostCmd
1096
values_to_set = ["192.0.2.3", "foo.example.org"]
1098
class TestSetSecretCmd(TestValueArgumentPropertyCmd):
1099
command = SetSecretCmd
1101
values_to_set = [open("/dev/null", "rb"),
1102
io.BytesIO(b"secret\0xyzzy\nbar")]
1103
values_to_get = [b"", b"secret\0xyzzy\nbar"]
1105
class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):
1106
command = SetTimeoutCmd
1107
property = "Timeout"
1108
values_to_set = [datetime.timedelta(),
1109
datetime.timedelta(minutes=5),
1110
datetime.timedelta(seconds=1),
1111
datetime.timedelta(weeks=1),
1112
datetime.timedelta(weeks=52)]
1113
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1115
class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):
1116
command = SetExtendedTimeoutCmd
1117
property = "ExtendedTimeout"
1118
values_to_set = [datetime.timedelta(),
1119
datetime.timedelta(minutes=5),
1120
datetime.timedelta(seconds=1),
1121
datetime.timedelta(weeks=1),
1122
datetime.timedelta(weeks=52)]
1123
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1125
class TestSetIntervalCmd(TestValueArgumentPropertyCmd):
1126
command = SetIntervalCmd
1127
property = "Interval"
1128
values_to_set = [datetime.timedelta(),
1129
datetime.timedelta(minutes=5),
1130
datetime.timedelta(seconds=1),
1131
datetime.timedelta(weeks=1),
1132
datetime.timedelta(weeks=52)]
1133
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1135
class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):
1136
command = SetApprovalDelayCmd
1137
property = "ApprovalDelay"
1138
values_to_set = [datetime.timedelta(),
1139
datetime.timedelta(minutes=5),
1140
datetime.timedelta(seconds=1),
1141
datetime.timedelta(weeks=1),
1142
datetime.timedelta(weeks=52)]
1143
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1145
class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):
1146
command = SetApprovalDurationCmd
1147
property = "ApprovalDuration"
1148
values_to_set = [datetime.timedelta(),
1149
datetime.timedelta(minutes=5),
1150
datetime.timedelta(seconds=1),
1151
datetime.timedelta(weeks=1),
1152
datetime.timedelta(weeks=52)]
1153
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1155
class Test_command_from_options(unittest.TestCase):
1157
self.parser = argparse.ArgumentParser()
1158
add_command_line_options(self.parser)
1159
def assert_command_from_args(self, args, command_cls, **cmd_attrs):
1160
"""Assert that parsing ARGS should result in an instance of
1161
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1162
options = self.parser.parse_args(args)
1163
check_option_syntax(self.parser, options)
1164
commands = commands_from_options(options)
1165
self.assertEqual(len(commands), 1)
1166
command = commands[0]
1167
self.assertIsInstance(command, command_cls)
1168
for key, value in cmd_attrs.items():
1169
self.assertEqual(getattr(command, key), value)
1170
def test_print_table(self):
1171
self.assert_command_from_args([], PrintTableCmd,
1174
def test_print_table_verbose(self):
1175
self.assert_command_from_args(["--verbose"], PrintTableCmd,
1178
def test_print_table_verbose_short(self):
1179
self.assert_command_from_args(["-v"], PrintTableCmd,
1182
def test_enable(self):
1183
self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1185
def test_enable_short(self):
1186
self.assert_command_from_args(["-e", "foo"], EnableCmd)
1188
def test_disable(self):
1189
self.assert_command_from_args(["--disable", "foo"],
1192
def test_disable_short(self):
1193
self.assert_command_from_args(["-d", "foo"], DisableCmd)
1195
def test_bump_timeout(self):
1196
self.assert_command_from_args(["--bump-timeout", "foo"],
1199
def test_bump_timeout_short(self):
1200
self.assert_command_from_args(["-b", "foo"], BumpTimeoutCmd)
1202
def test_start_checker(self):
1203
self.assert_command_from_args(["--start-checker", "foo"],
1206
def test_stop_checker(self):
1207
self.assert_command_from_args(["--stop-checker", "foo"],
1210
def test_remove(self):
1211
self.assert_command_from_args(["--remove", "foo"],
1214
def test_remove_short(self):
1215
self.assert_command_from_args(["-r", "foo"], RemoveCmd)
1217
def test_checker(self):
1218
self.assert_command_from_args(["--checker", ":", "foo"],
1219
SetCheckerCmd, value_to_set=":")
1221
def test_checker_empty(self):
1222
self.assert_command_from_args(["--checker", "", "foo"],
1223
SetCheckerCmd, value_to_set="")
1225
def test_checker_short(self):
1226
self.assert_command_from_args(["-c", ":", "foo"],
1227
SetCheckerCmd, value_to_set=":")
1229
def test_timeout(self):
1230
self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1232
value_to_set=300000)
1234
def test_timeout_short(self):
1235
self.assert_command_from_args(["-t", "PT5M", "foo"],
1237
value_to_set=300000)
1239
def test_extended_timeout(self):
1240
self.assert_command_from_args(["--extended-timeout", "PT15M",
1242
SetExtendedTimeoutCmd,
1243
value_to_set=900000)
1245
def test_interval(self):
1246
self.assert_command_from_args(["--interval", "PT2M", "foo"],
1248
value_to_set=120000)
1250
def test_interval_short(self):
1251
self.assert_command_from_args(["-i", "PT2M", "foo"],
1253
value_to_set=120000)
1255
def test_approve_by_default(self):
1256
self.assert_command_from_args(["--approve-by-default", "foo"],
1257
ApproveByDefaultCmd)
1259
def test_deny_by_default(self):
1260
self.assert_command_from_args(["--deny-by-default", "foo"],
1263
def test_approval_delay(self):
1264
self.assert_command_from_args(["--approval-delay", "PT30S",
1265
"foo"], SetApprovalDelayCmd,
1268
def test_approval_duration(self):
1269
self.assert_command_from_args(["--approval-duration", "PT1S",
1270
"foo"], SetApprovalDurationCmd,
1273
def test_host(self):
1274
self.assert_command_from_args(["--host", "foo.example.org",
1276
value_to_set="foo.example.org")
1278
def test_host_short(self):
1279
self.assert_command_from_args(["-H", "foo.example.org",
1281
value_to_set="foo.example.org")
1283
def test_secret_devnull(self):
1284
self.assert_command_from_args(["--secret", os.path.devnull,
1285
"foo"], SetSecretCmd,
1288
def test_secret_tempfile(self):
1289
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1290
value = b"secret\0xyzzy\nbar"
1293
self.assert_command_from_args(["--secret", f.name,
1294
"foo"], SetSecretCmd,
1297
def test_secret_devnull_short(self):
1298
self.assert_command_from_args(["-s", os.path.devnull, "foo"],
1299
SetSecretCmd, value_to_set=b"")
1301
def test_secret_tempfile_short(self):
1302
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1303
value = b"secret\0xyzzy\nbar"
1306
self.assert_command_from_args(["-s", f.name, "foo"],
1310
def test_approve(self):
1311
self.assert_command_from_args(["--approve", "foo"],
1314
def test_approve_short(self):
1315
self.assert_command_from_args(["-A", "foo"], ApproveCmd)
1317
def test_deny(self):
1318
self.assert_command_from_args(["--deny", "foo"], DenyCmd)
1320
def test_deny_short(self):
1321
self.assert_command_from_args(["-D", "foo"], DenyCmd)
1323
def test_dump_json(self):
1324
self.assert_command_from_args(["--dump-json"], DumpJSONCmd)
1326
def test_is_enabled(self):
1327
self.assert_command_from_args(["--is-enabled", "foo"],
1330
def test_is_enabled_short(self):
1331
self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)
871
1335
def should_only_run_tests():