826
777
self.__dbus_object_path__ = "objpath_{}".format(name)
827
778
self.attributes = attributes
828
779
self.attributes["Name"] = name
830
def Set(self, interface, property, value, dbus_interface):
780
def Set(interface, property, value,
781
properties_interface):
831
782
testcase.assertEqual(interface, client_interface)
832
testcase.assertEqual(dbus_interface,
783
testcase.assertEqual(properties_interface,
833
784
dbus.PROPERTIES_IFACE)
834
785
self.attributes[property] = value
835
def Get(self, interface, property, dbus_interface):
786
def Get(interface, property, properties_interface):
836
787
testcase.assertEqual(interface, client_interface)
837
testcase.assertEqual(dbus_interface,
788
testcase.assertEqual(properties_interface,
838
789
dbus.PROPERTIES_IFACE)
839
790
return self.attributes[property]
840
def Approve(self, approve, dbus_interface):
841
testcase.assertEqual(dbus_interface, client_interface)
842
self.calls.append(("Approve", (approve,
844
self.client = MockClient(
846
KeyID=("92ed150794387c03ce684574b1139a65"
847
"94a34f895daaaf09fd8ea90a27cddb12"),
849
Host="foo.example.org",
850
Enabled=dbus.Boolean(True),
852
LastCheckedOK="2019-02-03T00:00:00",
853
Created="2019-01-02T00:00:00",
855
Fingerprint=("778827225BA7DE539C5A"
856
"7CFA59CFF7CDBD9A5920"),
857
CheckerRunning=dbus.Boolean(False),
858
LastEnabled="2019-01-03T00:00:00",
859
ApprovalPending=dbus.Boolean(False),
860
ApprovedByDefault=dbus.Boolean(True),
861
LastApprovalRequest="",
863
ApprovalDuration=1000,
864
Checker="fping -q -- %(host)s",
865
ExtendedTimeout=900000,
866
Expires="2019-02-04T00:00:00",
868
self.other_client = MockClient(
870
KeyID=("0558568eedd67d622f5c83b35a115f79"
871
"6ab612cff5ad227247e46c2b020f441c"),
874
Enabled=dbus.Boolean(True),
876
LastCheckedOK="2019-02-04T00:00:00",
877
Created="2019-01-03T00:00:00",
879
Fingerprint=("3E393AEAEFB84C7E89E2"
880
"F547B3A107558FCA3A27"),
881
CheckerRunning=dbus.Boolean(True),
882
LastEnabled="2019-01-04T00:00:00",
883
ApprovalPending=dbus.Boolean(False),
884
ApprovedByDefault=dbus.Boolean(False),
885
LastApprovalRequest="2019-01-03T00:00:00",
887
ApprovalDuration=1000,
889
ExtendedTimeout=900000,
890
Expires="2019-02-05T00:00:00",
891
LastCheckerStatus=-2)
892
self.clients = collections.OrderedDict(
894
(self.client, self.client.attributes),
895
(self.other_client, self.other_client.attributes),
791
def __getitem__(self, key):
792
return self.attributes[key]
793
self.clients = collections.OrderedDict([
797
KeyID=("92ed150794387c03ce684574b1139a65"
798
"94a34f895daaaf09fd8ea90a27cddb12"),
800
Host="foo.example.org",
801
Enabled=dbus.Boolean(True),
803
LastCheckedOK="2019-02-03T00:00:00",
804
Created="2019-01-02T00:00:00",
806
Fingerprint=("778827225BA7DE539C5A"
807
"7CFA59CFF7CDBD9A5920"),
808
CheckerRunning=dbus.Boolean(False),
809
LastEnabled="2019-01-03T00:00:00",
810
ApprovalPending=dbus.Boolean(False),
811
ApprovedByDefault=dbus.Boolean(True),
812
LastApprovalRequest="",
814
ApprovalDuration=1000,
815
Checker="fping -q -- %(host)s",
816
ExtendedTimeout=900000,
817
Expires="2019-02-04T00:00:00",
818
LastCheckerStatus=0)),
822
KeyID=("0558568eedd67d622f5c83b35a115f79"
823
"6ab612cff5ad227247e46c2b020f441c"),
826
Enabled=dbus.Boolean(True),
828
LastCheckedOK="2019-02-04T00:00:00",
829
Created="2019-01-03T00:00:00",
831
Fingerprint=("3E393AEAEFB84C7E89E2"
832
"F547B3A107558FCA3A27"),
833
CheckerRunning=dbus.Boolean(True),
834
LastEnabled="2019-01-04T00:00:00",
835
ApprovalPending=dbus.Boolean(False),
836
ApprovedByDefault=dbus.Boolean(False),
837
LastApprovalRequest="2019-01-03T00:00:00",
839
ApprovalDuration=1000,
841
ExtendedTimeout=900000,
842
Expires="2019-02-05T00:00:00",
843
LastCheckerStatus=-2)),
897
self.one_client = {self.client: self.client.attributes}
899
846
class TestPrintTableCmd(TestCmd):
900
847
def test_normal(self):
978
925
json_data = json.loads(DumpJSONCmd().output(self.clients))
979
926
self.assertDictEqual(json_data, self.expected_json)
980
927
def test_one_client(self):
981
clients = self.one_client
928
clients = {"foo": self.clients["foo"]}
982
929
json_data = json.loads(DumpJSONCmd().output(clients))
983
930
expected_json = {"foo": self.expected_json["foo"]}
984
931
self.assertDictEqual(json_data, expected_json)
986
class TestIsEnabledCmd(TestCmd):
987
def test_is_enabled(self):
988
self.assertTrue(all(IsEnabledCmd().is_enabled(client, properties)
989
for client, properties in self.clients.items()))
990
def test_is_enabled_run_exits_successfully(self):
991
with self.assertRaises(SystemExit) as e:
992
IsEnabledCmd().run(None, self.one_client)
993
if e.exception.code is not None:
994
self.assertEqual(e.exception.code, 0)
996
self.assertIsNone(e.exception.code)
997
def test_is_enabled_run_exits_with_failure(self):
998
self.client.attributes["Enabled"] = dbus.Boolean(False)
999
with self.assertRaises(SystemExit) as e:
1000
IsEnabledCmd().run(None, self.one_client)
1001
if isinstance(e.exception.code, int):
1002
self.assertNotEqual(e.exception.code, 0)
1004
self.assertIsNotNone(e.exception.code)
1006
class TestRemoveCmd(TestCmd):
1007
def test_remove(self):
1008
class MockMandos(object):
1011
def RemoveClient(self, dbus_path):
1012
self.calls.append(("RemoveClient", (dbus_path,)))
1013
mandos = MockMandos()
1014
super(TestRemoveCmd, self).setUp()
1015
RemoveCmd().run(mandos, self.clients)
1016
self.assertEqual(len(mandos.calls), 2)
1017
for client in self.clients:
1018
self.assertIn(("RemoveClient",
1019
(client.__dbus_object_path__,)),
1022
class TestApproveCmd(TestCmd):
1023
def test_approve(self):
1024
ApproveCmd().run(None, self.clients)
1025
for client in self.clients:
1026
self.assertIn(("Approve", (True, client_interface)),
1029
class TestDenyCmd(TestCmd):
1030
def test_deny(self):
1031
DenyCmd().run(None, self.clients)
1032
for client in self.clients:
1033
self.assertIn(("Approve", (False, client_interface)),
1036
class TestEnableCmd(TestCmd):
1037
def test_enable(self):
1038
for client in self.clients:
1039
client.attributes["Enabled"] = False
1041
EnableCmd().run(None, self.clients)
1043
for client in self.clients:
1044
self.assertTrue(client.attributes["Enabled"])
1046
class TestDisableCmd(TestCmd):
1047
def test_disable(self):
1048
DisableCmd().run(None, self.clients)
1050
for client in self.clients:
1051
self.assertFalse(client.attributes["Enabled"])
1053
class Unique(object):
1054
"""Class for objects which exist only to be unique objects, since
1055
unittest.mock.sentinel only exists in Python 3.3"""
1057
class TestPropertyCmd(TestCmd):
1058
"""Abstract class for tests of PropertyCmd classes"""
1060
if not hasattr(self, "command"):
1062
values_to_get = getattr(self, "values_to_get",
1064
for value_to_set, value_to_get in zip(self.values_to_set,
1066
for client in self.clients:
1067
old_value = client.attributes[self.property]
1068
self.assertNotIsInstance(old_value, Unique)
1069
client.attributes[self.property] = Unique()
1070
self.run_command(value_to_set, self.clients)
1071
for client in self.clients:
1072
value = client.attributes[self.property]
1073
self.assertNotIsInstance(value, Unique)
1074
self.assertEqual(value, value_to_get)
1075
def run_command(self, value, clients):
1076
self.command().run(None, clients)
1078
class TestBumpTimeoutCmd(TestPropertyCmd):
1079
command = BumpTimeoutCmd
1080
property = "LastCheckedOK"
1081
values_to_set = [""]
1083
class TestStartCheckerCmd(TestPropertyCmd):
1084
command = StartCheckerCmd
1085
property = "CheckerRunning"
1086
values_to_set = [dbus.Boolean(True)]
1088
class TestStopCheckerCmd(TestPropertyCmd):
1089
command = StopCheckerCmd
1090
property = "CheckerRunning"
1091
values_to_set = [dbus.Boolean(False)]
1093
class TestApproveByDefaultCmd(TestPropertyCmd):
1094
command = ApproveByDefaultCmd
1095
property = "ApprovedByDefault"
1096
values_to_set = [dbus.Boolean(True)]
1098
class TestDenyByDefaultCmd(TestPropertyCmd):
1099
command = DenyByDefaultCmd
1100
property = "ApprovedByDefault"
1101
values_to_set = [dbus.Boolean(False)]
1103
class TestValueArgumentPropertyCmd(TestPropertyCmd):
1104
"""Abstract class for tests of PropertyCmd classes using the
1105
ValueArgumentMixIn"""
1107
if type(self) is TestValueArgumentPropertyCmd:
1109
return super(TestValueArgumentPropertyCmd, self).runTest()
1110
def run_command(self, value, clients):
1111
self.command(value).run(None, clients)
1113
class TestSetCheckerCmd(TestValueArgumentPropertyCmd):
1114
command = SetCheckerCmd
1115
property = "Checker"
1116
values_to_set = ["", ":", "fping -q -- %s"]
1118
class TestSetHostCmd(TestValueArgumentPropertyCmd):
1119
command = SetHostCmd
1121
values_to_set = ["192.0.2.3", "foo.example.org"]
1123
class TestSetSecretCmd(TestValueArgumentPropertyCmd):
1124
command = SetSecretCmd
1126
values_to_set = [io.BytesIO(b""),
1127
io.BytesIO(b"secret\0xyzzy\nbar")]
1128
values_to_get = [b"", b"secret\0xyzzy\nbar"]
1130
class TestSetTimeoutCmd(TestValueArgumentPropertyCmd):
1131
command = SetTimeoutCmd
1132
property = "Timeout"
1133
values_to_set = [datetime.timedelta(),
1134
datetime.timedelta(minutes=5),
1135
datetime.timedelta(seconds=1),
1136
datetime.timedelta(weeks=1),
1137
datetime.timedelta(weeks=52)]
1138
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1140
class TestSetExtendedTimeoutCmd(TestValueArgumentPropertyCmd):
1141
command = SetExtendedTimeoutCmd
1142
property = "ExtendedTimeout"
1143
values_to_set = [datetime.timedelta(),
1144
datetime.timedelta(minutes=5),
1145
datetime.timedelta(seconds=1),
1146
datetime.timedelta(weeks=1),
1147
datetime.timedelta(weeks=52)]
1148
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1150
class TestSetIntervalCmd(TestValueArgumentPropertyCmd):
1151
command = SetIntervalCmd
1152
property = "Interval"
1153
values_to_set = [datetime.timedelta(),
1154
datetime.timedelta(minutes=5),
1155
datetime.timedelta(seconds=1),
1156
datetime.timedelta(weeks=1),
1157
datetime.timedelta(weeks=52)]
1158
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1160
class TestSetApprovalDelayCmd(TestValueArgumentPropertyCmd):
1161
command = SetApprovalDelayCmd
1162
property = "ApprovalDelay"
1163
values_to_set = [datetime.timedelta(),
1164
datetime.timedelta(minutes=5),
1165
datetime.timedelta(seconds=1),
1166
datetime.timedelta(weeks=1),
1167
datetime.timedelta(weeks=52)]
1168
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1170
class TestSetApprovalDurationCmd(TestValueArgumentPropertyCmd):
1171
command = SetApprovalDurationCmd
1172
property = "ApprovalDuration"
1173
values_to_set = [datetime.timedelta(),
1174
datetime.timedelta(minutes=5),
1175
datetime.timedelta(seconds=1),
1176
datetime.timedelta(weeks=1),
1177
datetime.timedelta(weeks=52)]
1178
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1180
class Test_command_from_options(unittest.TestCase):
1182
self.parser = argparse.ArgumentParser()
1183
add_command_line_options(self.parser)
1184
def assert_command_from_args(self, args, command_cls, **cmd_attrs):
1185
"""Assert that parsing ARGS should result in an instance of
1186
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1187
options = self.parser.parse_args(args)
1188
check_option_syntax(self.parser, options)
1189
commands = commands_from_options(options)
1190
self.assertEqual(len(commands), 1)
1191
command = commands[0]
1192
self.assertIsInstance(command, command_cls)
1193
for key, value in cmd_attrs.items():
1194
self.assertEqual(getattr(command, key), value)
1195
def test_print_table(self):
1196
self.assert_command_from_args([], PrintTableCmd,
1199
def test_print_table_verbose(self):
1200
self.assert_command_from_args(["--verbose"], PrintTableCmd,
1203
def test_print_table_verbose_short(self):
1204
self.assert_command_from_args(["-v"], PrintTableCmd,
1207
def test_enable(self):
1208
self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1210
def test_enable_short(self):
1211
self.assert_command_from_args(["-e", "foo"], EnableCmd)
1213
def test_disable(self):
1214
self.assert_command_from_args(["--disable", "foo"],
1217
def test_disable_short(self):
1218
self.assert_command_from_args(["-d", "foo"], DisableCmd)
1220
def test_bump_timeout(self):
1221
self.assert_command_from_args(["--bump-timeout", "foo"],
1224
def test_bump_timeout_short(self):
1225
self.assert_command_from_args(["-b", "foo"], BumpTimeoutCmd)
1227
def test_start_checker(self):
1228
self.assert_command_from_args(["--start-checker", "foo"],
1231
def test_stop_checker(self):
1232
self.assert_command_from_args(["--stop-checker", "foo"],
1235
def test_remove(self):
1236
self.assert_command_from_args(["--remove", "foo"],
1239
def test_remove_short(self):
1240
self.assert_command_from_args(["-r", "foo"], RemoveCmd)
1242
def test_checker(self):
1243
self.assert_command_from_args(["--checker", ":", "foo"],
1244
SetCheckerCmd, value_to_set=":")
1246
def test_checker_empty(self):
1247
self.assert_command_from_args(["--checker", "", "foo"],
1248
SetCheckerCmd, value_to_set="")
1250
def test_checker_short(self):
1251
self.assert_command_from_args(["-c", ":", "foo"],
1252
SetCheckerCmd, value_to_set=":")
1254
def test_timeout(self):
1255
self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1257
value_to_set=300000)
1259
def test_timeout_short(self):
1260
self.assert_command_from_args(["-t", "PT5M", "foo"],
1262
value_to_set=300000)
1264
def test_extended_timeout(self):
1265
self.assert_command_from_args(["--extended-timeout", "PT15M",
1267
SetExtendedTimeoutCmd,
1268
value_to_set=900000)
1270
def test_interval(self):
1271
self.assert_command_from_args(["--interval", "PT2M", "foo"],
1273
value_to_set=120000)
1275
def test_interval_short(self):
1276
self.assert_command_from_args(["-i", "PT2M", "foo"],
1278
value_to_set=120000)
1280
def test_approve_by_default(self):
1281
self.assert_command_from_args(["--approve-by-default", "foo"],
1282
ApproveByDefaultCmd)
1284
def test_deny_by_default(self):
1285
self.assert_command_from_args(["--deny-by-default", "foo"],
1288
def test_approval_delay(self):
1289
self.assert_command_from_args(["--approval-delay", "PT30S",
1290
"foo"], SetApprovalDelayCmd,
1293
def test_approval_duration(self):
1294
self.assert_command_from_args(["--approval-duration", "PT1S",
1295
"foo"], SetApprovalDurationCmd,
1298
def test_host(self):
1299
self.assert_command_from_args(["--host", "foo.example.org",
1301
value_to_set="foo.example.org")
1303
def test_host_short(self):
1304
self.assert_command_from_args(["-H", "foo.example.org",
1306
value_to_set="foo.example.org")
1308
def test_secret_devnull(self):
1309
self.assert_command_from_args(["--secret", os.path.devnull,
1310
"foo"], SetSecretCmd,
1313
def test_secret_tempfile(self):
1314
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1315
value = b"secret\0xyzzy\nbar"
1318
self.assert_command_from_args(["--secret", f.name,
1319
"foo"], SetSecretCmd,
1322
def test_secret_devnull_short(self):
1323
self.assert_command_from_args(["-s", os.path.devnull, "foo"],
1324
SetSecretCmd, value_to_set=b"")
1326
def test_secret_tempfile_short(self):
1327
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1328
value = b"secret\0xyzzy\nbar"
1331
self.assert_command_from_args(["-s", f.name, "foo"],
1335
def test_approve(self):
1336
self.assert_command_from_args(["--approve", "foo"],
1339
def test_approve_short(self):
1340
self.assert_command_from_args(["-A", "foo"], ApproveCmd)
1342
def test_deny(self):
1343
self.assert_command_from_args(["--deny", "foo"], DenyCmd)
1345
def test_deny_short(self):
1346
self.assert_command_from_args(["-D", "foo"], DenyCmd)
1348
def test_dump_json(self):
1349
self.assert_command_from_args(["--dump-json"], DumpJSONCmd)
1351
def test_is_enabled(self):
1352
self.assert_command_from_args(["--is-enabled", "foo"],
1355
def test_is_enabled_short(self):
1356
self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)
1359
class Test_check_option_syntax(unittest.TestCase):
1360
# This mostly corresponds to the definition from has_actions() in
1361
# check_option_syntax()
1363
# The actual values set here are not that important, but we do
1364
# at least stick to the correct types, even though they are
1368
"bump_timeout": True,
1369
"start_checker": True,
1370
"stop_checker": True,
1374
"timeout": datetime.timedelta(),
1375
"extended_timeout": datetime.timedelta(),
1376
"interval": datetime.timedelta(),
1377
"approved_by_default": True,
1378
"approval_delay": datetime.timedelta(),
1379
"approval_duration": datetime.timedelta(),
1381
"secret": io.BytesIO(b"x"),
1387
self.parser = argparse.ArgumentParser()
1388
add_command_line_options(self.parser)
1390
@contextlib.contextmanager
1391
def assertParseError(self):
1392
with self.assertRaises(SystemExit) as e:
1393
with self.temporarily_suppress_stderr():
1395
# Exit code from argparse is guaranteed to be "2". Reference:
1396
# https://docs.python.org/3/library/argparse.html#exiting-methods
1397
self.assertEqual(e.exception.code, 2)
1400
@contextlib.contextmanager
1401
def temporarily_suppress_stderr():
1402
null = os.open(os.path.devnull, os.O_RDWR)
1403
stderrcopy = os.dup(sys.stderr.fileno())
1404
os.dup2(null, sys.stderr.fileno())
1410
os.dup2(stderrcopy, sys.stderr.fileno())
1411
os.close(stderrcopy)
1413
def check_option_syntax(self, options):
1414
check_option_syntax(self.parser, options)
1416
def test_actions_requires_client_or_all(self):
1417
for action, value in self.actions.items():
1418
options = self.parser.parse_args()
1419
setattr(options, action, value)
1420
with self.assertParseError():
1421
self.check_option_syntax(options)
1423
def test_actions_conflicts_with_verbose(self):
1424
for action, value in self.actions.items():
1425
options = self.parser.parse_args()
1426
setattr(options, action, value)
1427
options.verbose = True
1428
with self.assertParseError():
1429
self.check_option_syntax(options)
1431
def test_dump_json_conflicts_with_verbose(self):
1432
options = self.parser.parse_args()
1433
options.dump_json = True
1434
options.verbose = True
1435
with self.assertParseError():
1436
self.check_option_syntax(options)
1438
def test_dump_json_conflicts_with_action(self):
1439
for action, value in self.actions.items():
1440
options = self.parser.parse_args()
1441
setattr(options, action, value)
1442
options.dump_json = True
1443
with self.assertParseError():
1444
self.check_option_syntax(options)
1446
def test_all_can_not_be_alone(self):
1447
options = self.parser.parse_args()
1449
with self.assertParseError():
1450
self.check_option_syntax(options)
1452
def test_all_is_ok_with_any_action(self):
1453
for action, value in self.actions.items():
1454
options = self.parser.parse_args()
1455
setattr(options, action, value)
1457
self.check_option_syntax(options)
1459
def test_is_enabled_fails_without_client(self):
1460
options = self.parser.parse_args()
1461
options.is_enabled = True
1462
with self.assertParseError():
1463
self.check_option_syntax(options)
1465
def test_is_enabled_works_with_one_client(self):
1466
options = self.parser.parse_args()
1467
options.is_enabled = True
1468
options.client = ["foo"]
1469
self.check_option_syntax(options)
1471
def test_is_enabled_fails_with_two_clients(self):
1472
options = self.parser.parse_args()
1473
options.is_enabled = True
1474
options.client = ["foo", "barbar"]
1475
with self.assertParseError():
1476
self.check_option_syntax(options)
1480
934
def should_only_run_tests():
1481
935
parser = argparse.ArgumentParser(add_help=False)