364
296
"LastApprovalRequest", "ApprovalDelay",
365
297
"ApprovalDuration", "Checker", "ExtendedTimeout",
366
298
"Expires", "LastCheckerStatus")
367
def run(self, clients):
368
print(self.output(clients))
299
def run(self, clients, bus=None, mandos=None):
300
print(self.output(clients.values()))
301
def output(self, clients):
302
raise NotImplementedError()
370
304
class PropertyCmd(Command):
371
305
"""Abstract class for Actions for setting one client property"""
372
def run_on_one_client(self, client):
306
def run_on_one_client(self, client, properties):
373
307
"""Set the Client's D-Bus property"""
374
client.Set(client_interface, self.property, self.value_to_set,
308
log.debug("D-Bus: %s:%s:%s.Set(%r, %r, %r)", busname,
309
client.__dbus_object_path__,
310
dbus.PROPERTIES_IFACE, client_interface,
311
self.propname, self.value_to_set
312
if not isinstance(self.value_to_set, dbus.Boolean)
313
else bool(self.value_to_set))
314
client.Set(client_interface, self.propname, self.value_to_set,
375
315
dbus_interface=dbus.PROPERTIES_IFACE)
318
raise NotImplementedError()
377
class ValueArgumentMixIn(object):
378
"""Mixin class for commands taking a value as argument"""
320
class PropertyValueCmd(PropertyCmd):
321
"""Abstract class for PropertyCmd recieving a value as argument"""
379
322
def __init__(self, value):
380
323
self.value_to_set = value
382
class MillisecondsValueArgumentMixIn(ValueArgumentMixIn):
383
"""Mixin class for commands taking a value argument as
325
class MillisecondsPropertyValueArgumentCmd(PropertyValueCmd):
326
"""Abstract class for PropertyValueCmd taking a value argument as
327
a datetime.timedelta() but should store it as milliseconds."""
386
329
def value_to_set(self):
388
331
@value_to_set.setter
389
332
def value_to_set(self, value):
390
"""When setting, convert value to a datetime.timedelta"""
391
self._vts = string_to_delta(value).total_seconds() * 1000
333
"""When setting, convert value from a datetime.timedelta"""
334
self._vts = int(round(value.total_seconds() * 1000))
393
336
# Actual (non-abstract) command classes
395
338
class PrintTableCmd(PrintCmd):
396
339
def __init__(self, verbose=False):
397
340
self.verbose = verbose
398
342
def output(self, clients):
343
default_keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
344
keywords = default_keywords
400
346
keywords = self.all_keywords
402
keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK")
403
return str(TableOfClients(clients.values(), keywords))
347
return str(self.TableOfClients(clients, keywords))
349
class TableOfClients(object):
352
"Enabled": "Enabled",
353
"Timeout": "Timeout",
354
"LastCheckedOK": "Last Successful Check",
355
"LastApprovalRequest": "Last Approval Request",
356
"Created": "Created",
357
"Interval": "Interval",
359
"Fingerprint": "Fingerprint",
361
"CheckerRunning": "Check Is Running",
362
"LastEnabled": "Last Enabled",
363
"ApprovalPending": "Approval Is Pending",
364
"ApprovedByDefault": "Approved By Default",
365
"ApprovalDelay": "Approval Delay",
366
"ApprovalDuration": "Approval Duration",
367
"Checker": "Checker",
368
"ExtendedTimeout": "Extended Timeout",
369
"Expires": "Expires",
370
"LastCheckerStatus": "Last Checker Status",
373
def __init__(self, clients, keywords, tableheaders=None):
374
self.clients = clients
375
self.keywords = keywords
376
if tableheaders is not None:
377
self.tableheaders = tableheaders
380
return "\n".join(self.rows())
382
if sys.version_info.major == 2:
383
__unicode__ = __str__
385
return str(self).encode(locale.getpreferredencoding())
388
format_string = self.row_formatting_string()
389
rows = [self.header_line(format_string)]
390
rows.extend(self.client_line(client, format_string)
391
for client in self.clients)
394
def row_formatting_string(self):
395
"Format string used to format table rows"
396
return " ".join("{{{key}:{width}}}".format(
397
width=max(len(self.tableheaders[key]),
398
*(len(self.string_from_client(client, key))
399
for client in self.clients)),
401
for key in self.keywords)
403
def string_from_client(self, client, key):
404
return self.valuetostring(client[key], key)
407
def valuetostring(value, keyword):
408
if isinstance(value, dbus.Boolean):
409
return "Yes" if value else "No"
410
if keyword in ("Timeout", "Interval", "ApprovalDelay",
411
"ApprovalDuration", "ExtendedTimeout"):
412
return milliseconds_to_string(value)
415
def header_line(self, format_string):
416
return format_string.format(**self.tableheaders)
418
def client_line(self, client, format_string):
419
return format_string.format(
420
**{key: self.string_from_client(client, key)
421
for key in self.keywords})
405
425
class DumpJSONCmd(PrintCmd):
406
426
def output(self, clients):
741
810
with self.assertLogs(log, logging.WARNING):
742
811
value = string_to_delta("2h")
744
value = string_to_delta("2h")
813
class WarningFilter(logging.Filter):
814
"""Don't show, but record the presence of, warnings"""
815
def filter(self, record):
816
is_warning = record.levelno >= logging.WARNING
817
self.found = is_warning or getattr(self, "found",
819
return not is_warning
820
warning_filter = WarningFilter()
821
log.addFilter(warning_filter)
823
value = string_to_delta("2h")
825
log.removeFilter(warning_filter)
826
self.assertTrue(getattr(warning_filter, "found", False))
745
827
self.assertEqual(value, datetime.timedelta(0, 7200))
747
class Test_TableOfClients(unittest.TestCase):
753
"Bool": "A D-BUS Boolean",
754
"NonDbusBoolean": "A Non-D-BUS Boolean",
755
"Integer": "An Integer",
756
"Timeout": "Timedelta 1",
757
"Interval": "Timedelta 2",
758
"ApprovalDelay": "Timedelta 3",
759
"ApprovalDuration": "Timedelta 4",
760
"ExtendedTimeout": "Timedelta 5",
761
"String": "A String",
830
class TestCmd(unittest.TestCase):
831
"""Abstract class for tests of command classes"""
834
class MockClient(object):
835
def __init__(self, name, **attributes):
836
self.__dbus_object_path__ = "/clients/{}".format(name)
837
self.attributes = attributes
838
self.attributes["Name"] = name
840
def Set(self, interface, propname, value, dbus_interface):
841
testcase.assertEqual(interface, client_interface)
842
testcase.assertEqual(dbus_interface,
843
dbus.PROPERTIES_IFACE)
844
self.attributes[propname] = value
845
def Get(self, interface, propname, dbus_interface):
846
testcase.assertEqual(interface, client_interface)
847
testcase.assertEqual(dbus_interface,
848
dbus.PROPERTIES_IFACE)
849
return self.attributes[propname]
850
def Approve(self, approve, dbus_interface):
851
testcase.assertEqual(dbus_interface, client_interface)
852
self.calls.append(("Approve", (approve,
854
self.client = MockClient(
856
KeyID=("92ed150794387c03ce684574b1139a65"
857
"94a34f895daaaf09fd8ea90a27cddb12"),
859
Host="foo.example.org",
860
Enabled=dbus.Boolean(True),
862
LastCheckedOK="2019-02-03T00:00:00",
863
Created="2019-01-02T00:00:00",
865
Fingerprint=("778827225BA7DE539C5A"
866
"7CFA59CFF7CDBD9A5920"),
867
CheckerRunning=dbus.Boolean(False),
868
LastEnabled="2019-01-03T00:00:00",
869
ApprovalPending=dbus.Boolean(False),
870
ApprovedByDefault=dbus.Boolean(True),
871
LastApprovalRequest="",
873
ApprovalDuration=1000,
874
Checker="fping -q -- %(host)s",
875
ExtendedTimeout=900000,
876
Expires="2019-02-04T00:00:00",
878
self.other_client = MockClient(
880
KeyID=("0558568eedd67d622f5c83b35a115f79"
881
"6ab612cff5ad227247e46c2b020f441c"),
884
Enabled=dbus.Boolean(True),
886
LastCheckedOK="2019-02-04T00:00:00",
887
Created="2019-01-03T00:00:00",
889
Fingerprint=("3E393AEAEFB84C7E89E2"
890
"F547B3A107558FCA3A27"),
891
CheckerRunning=dbus.Boolean(True),
892
LastEnabled="2019-01-04T00:00:00",
893
ApprovalPending=dbus.Boolean(False),
894
ApprovedByDefault=dbus.Boolean(False),
895
LastApprovalRequest="2019-01-03T00:00:00",
897
ApprovalDuration=1000,
899
ExtendedTimeout=900000,
900
Expires="2019-02-05T00:00:00",
901
LastCheckerStatus=-2)
902
self.clients = collections.OrderedDict(
904
("/clients/foo", self.client.attributes),
905
("/clients/barbar", self.other_client.attributes),
907
self.one_client = {"/clients/foo": self.client.attributes}
912
def get_object(client_bus_name, path):
913
self.assertEqual(client_bus_name, busname)
915
"/clients/foo": self.client,
916
"/clients/barbar": self.other_client,
920
class TestPrintTableCmd(TestCmd):
921
def test_normal(self):
922
output = PrintTableCmd().output(self.clients.values())
923
expected_output = """
924
Name Enabled Timeout Last Successful Check
925
foo Yes 00:05:00 2019-02-03T00:00:00
926
barbar Yes 00:05:00 2019-02-04T00:00:00
928
self.assertEqual(output, expected_output)
929
def test_verbose(self):
930
output = PrintTableCmd(verbose=True).output(
931
self.clients.values())
932
expected_output = """
933
Name Enabled Timeout Last Successful Check Created Interval Host Key ID Fingerprint Check Is Running Last Enabled Approval Is Pending Approved By Default Last Approval Request Approval Delay Approval Duration Checker Extended Timeout Expires Last Checker Status
934
foo Yes 00:05:00 2019-02-03T00:00:00 2019-01-02T00:00:00 00:02:00 foo.example.org 92ed150794387c03ce684574b1139a6594a34f895daaaf09fd8ea90a27cddb12 778827225BA7DE539C5A7CFA59CFF7CDBD9A5920 No 2019-01-03T00:00:00 No Yes 00:00:00 00:00:01 fping -q -- %(host)s 00:15:00 2019-02-04T00:00:00 0
935
barbar Yes 00:05:00 2019-02-04T00:00:00 2019-01-03T00:00:00 00:02:00 192.0.2.3 0558568eedd67d622f5c83b35a115f796ab612cff5ad227247e46c2b020f441c 3E393AEAEFB84C7E89E2F547B3A107558FCA3A27 Yes 2019-01-04T00:00:00 No No 2019-01-03T00:00:00 00:00:30 00:00:01 : 00:15:00 2019-02-05T00:00:00 -2
937
self.assertEqual(output, expected_output)
938
def test_one_client(self):
939
output = PrintTableCmd().output(self.one_client.values())
940
expected_output = """
941
Name Enabled Timeout Last Successful Check
942
foo Yes 00:05:00 2019-02-03T00:00:00
944
self.assertEqual(output, expected_output)
946
class TestDumpJSONCmd(TestCmd):
948
self.expected_json = {
951
"KeyID": ("92ed150794387c03ce684574b1139a65"
952
"94a34f895daaaf09fd8ea90a27cddb12"),
953
"Host": "foo.example.org",
956
"LastCheckedOK": "2019-02-03T00:00:00",
957
"Created": "2019-01-02T00:00:00",
959
"Fingerprint": ("778827225BA7DE539C5A"
960
"7CFA59CFF7CDBD9A5920"),
961
"CheckerRunning": False,
962
"LastEnabled": "2019-01-03T00:00:00",
963
"ApprovalPending": False,
964
"ApprovedByDefault": True,
965
"LastApprovalRequest": "",
967
"ApprovalDuration": 1000,
968
"Checker": "fping -q -- %(host)s",
969
"ExtendedTimeout": 900000,
970
"Expires": "2019-02-04T00:00:00",
971
"LastCheckerStatus": 0,
975
"KeyID": ("0558568eedd67d622f5c83b35a115f79"
976
"6ab612cff5ad227247e46c2b020f441c"),
980
"LastCheckedOK": "2019-02-04T00:00:00",
981
"Created": "2019-01-03T00:00:00",
983
"Fingerprint": ("3E393AEAEFB84C7E89E2"
984
"F547B3A107558FCA3A27"),
985
"CheckerRunning": True,
986
"LastEnabled": "2019-01-04T00:00:00",
987
"ApprovalPending": False,
988
"ApprovedByDefault": False,
989
"LastApprovalRequest": "2019-01-03T00:00:00",
990
"ApprovalDelay": 30000,
991
"ApprovalDuration": 1000,
993
"ExtendedTimeout": 900000,
994
"Expires": "2019-02-05T00:00:00",
995
"LastCheckerStatus": -2,
763
self.keywords = ["Attr1", "AttrTwo"]
769
"Bool": dbus.Boolean(False),
770
"NonDbusBoolean": False,
774
"ApprovalDelay": 2000,
775
"ApprovalDuration": 3000,
776
"ExtendedTimeout": 4000,
783
"Bool": dbus.Boolean(True),
784
"NonDbusBoolean": True,
787
"Interval": 93786000,
788
"ApprovalDelay": 93787000,
789
"ApprovalDuration": 93788000,
790
"ExtendedTimeout": 93789000,
791
"String": "A huge string which will not fit," * 10,
794
def test_short_header(self):
795
text = str(TableOfClients(self.clients, self.keywords,
802
self.assertEqual(text, expected_text)
803
def test_booleans(self):
804
keywords = ["Bool", "NonDbusBoolean"]
805
text = str(TableOfClients(self.clients, keywords,
808
A D-BUS Boolean A Non-D-BUS Boolean
812
self.assertEqual(text, expected_text)
813
def test_milliseconds_detection(self):
814
keywords = ["Integer", "Timeout", "Interval", "ApprovalDelay",
815
"ApprovalDuration", "ExtendedTimeout"]
816
text = str(TableOfClients(self.clients, keywords,
819
An Integer Timedelta 1 Timedelta 2 Timedelta 3 Timedelta 4 Timedelta 5
820
0 00:00:00 00:00:01 00:00:02 00:00:03 00:00:04
821
1 1T02:03:05 1T02:03:06 1T02:03:07 1T02:03:08 1T02:03:09
823
self.assertEqual(text, expected_text)
824
def test_empty_and_long_string_values(self):
825
keywords = ["String"]
826
text = str(TableOfClients(self.clients, keywords,
831
A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,A huge string which will not fit,
833
self.assertEqual(text, expected_text)
998
return super(TestDumpJSONCmd, self).setUp()
999
def test_normal(self):
1000
json_data = json.loads(DumpJSONCmd().output(self.clients))
1001
self.assertDictEqual(json_data, self.expected_json)
1002
def test_one_client(self):
1003
clients = self.one_client
1004
json_data = json.loads(DumpJSONCmd().output(clients))
1005
expected_json = {"foo": self.expected_json["foo"]}
1006
self.assertDictEqual(json_data, expected_json)
1008
class TestIsEnabledCmd(TestCmd):
1009
def test_is_enabled(self):
1010
self.assertTrue(all(IsEnabledCmd().is_enabled(client, properties)
1011
for client, properties in self.clients.items()))
1012
def test_is_enabled_run_exits_successfully(self):
1013
with self.assertRaises(SystemExit) as e:
1014
IsEnabledCmd().run(self.one_client)
1015
if e.exception.code is not None:
1016
self.assertEqual(e.exception.code, 0)
1018
self.assertIsNone(e.exception.code)
1019
def test_is_enabled_run_exits_with_failure(self):
1020
self.client.attributes["Enabled"] = dbus.Boolean(False)
1021
with self.assertRaises(SystemExit) as e:
1022
IsEnabledCmd().run(self.one_client)
1023
if isinstance(e.exception.code, int):
1024
self.assertNotEqual(e.exception.code, 0)
1026
self.assertIsNotNone(e.exception.code)
1028
class TestRemoveCmd(TestCmd):
1029
def test_remove(self):
1030
class MockMandos(object):
1033
def RemoveClient(self, dbus_path):
1034
self.calls.append(("RemoveClient", (dbus_path,)))
1035
mandos = MockMandos()
1036
super(TestRemoveCmd, self).setUp()
1037
RemoveCmd().run(self.clients, self.bus, mandos)
1038
self.assertEqual(len(mandos.calls), 2)
1039
for clientpath in self.clients:
1040
self.assertIn(("RemoveClient", (clientpath,)),
1043
class TestApproveCmd(TestCmd):
1044
def test_approve(self):
1045
ApproveCmd().run(self.clients, self.bus)
1046
for clientpath in self.clients:
1047
client = self.bus.get_object(busname, clientpath)
1048
self.assertIn(("Approve", (True, client_interface)),
1051
class TestDenyCmd(TestCmd):
1052
def test_deny(self):
1053
DenyCmd().run(self.clients, self.bus)
1054
for clientpath in self.clients:
1055
client = self.bus.get_object(busname, clientpath)
1056
self.assertIn(("Approve", (False, client_interface)),
1059
class TestEnableCmd(TestCmd):
1060
def test_enable(self):
1061
for clientpath in self.clients:
1062
client = self.bus.get_object(busname, clientpath)
1063
client.attributes["Enabled"] = False
1065
EnableCmd().run(self.clients, self.bus)
1067
for clientpath in self.clients:
1068
client = self.bus.get_object(busname, clientpath)
1069
self.assertTrue(client.attributes["Enabled"])
1071
class TestDisableCmd(TestCmd):
1072
def test_disable(self):
1073
DisableCmd().run(self.clients, self.bus)
1074
for clientpath in self.clients:
1075
client = self.bus.get_object(busname, clientpath)
1076
self.assertFalse(client.attributes["Enabled"])
1078
class Unique(object):
1079
"""Class for objects which exist only to be unique objects, since
1080
unittest.mock.sentinel only exists in Python 3.3"""
1082
class TestPropertyCmd(TestCmd):
1083
"""Abstract class for tests of PropertyCmd classes"""
1085
if not hasattr(self, "command"):
1087
values_to_get = getattr(self, "values_to_get",
1089
for value_to_set, value_to_get in zip(self.values_to_set,
1091
for clientpath in self.clients:
1092
client = self.bus.get_object(busname, clientpath)
1093
old_value = client.attributes[self.propname]
1094
self.assertNotIsInstance(old_value, Unique)
1095
client.attributes[self.propname] = Unique()
1096
self.run_command(value_to_set, self.clients)
1097
for clientpath in self.clients:
1098
client = self.bus.get_object(busname, clientpath)
1099
value = client.attributes[self.propname]
1100
self.assertNotIsInstance(value, Unique)
1101
self.assertEqual(value, value_to_get)
1102
def run_command(self, value, clients):
1103
self.command().run(clients, self.bus)
1105
class TestBumpTimeoutCmd(TestPropertyCmd):
1106
command = BumpTimeoutCmd
1107
propname = "LastCheckedOK"
1108
values_to_set = [""]
1110
class TestStartCheckerCmd(TestPropertyCmd):
1111
command = StartCheckerCmd
1112
propname = "CheckerRunning"
1113
values_to_set = [dbus.Boolean(True)]
1115
class TestStopCheckerCmd(TestPropertyCmd):
1116
command = StopCheckerCmd
1117
propname = "CheckerRunning"
1118
values_to_set = [dbus.Boolean(False)]
1120
class TestApproveByDefaultCmd(TestPropertyCmd):
1121
command = ApproveByDefaultCmd
1122
propname = "ApprovedByDefault"
1123
values_to_set = [dbus.Boolean(True)]
1125
class TestDenyByDefaultCmd(TestPropertyCmd):
1126
command = DenyByDefaultCmd
1127
propname = "ApprovedByDefault"
1128
values_to_set = [dbus.Boolean(False)]
1130
class TestPropertyValueCmd(TestPropertyCmd):
1131
"""Abstract class for tests of PropertyValueCmd classes"""
1133
if type(self) is TestPropertyValueCmd:
1135
return super(TestPropertyValueCmd, self).runTest()
1136
def run_command(self, value, clients):
1137
self.command(value).run(clients, self.bus)
1139
class TestSetCheckerCmd(TestPropertyValueCmd):
1140
command = SetCheckerCmd
1141
propname = "Checker"
1142
values_to_set = ["", ":", "fping -q -- %s"]
1144
class TestSetHostCmd(TestPropertyValueCmd):
1145
command = SetHostCmd
1147
values_to_set = ["192.0.2.3", "foo.example.org"]
1149
class TestSetSecretCmd(TestPropertyValueCmd):
1150
command = SetSecretCmd
1152
values_to_set = [io.BytesIO(b""),
1153
io.BytesIO(b"secret\0xyzzy\nbar")]
1154
values_to_get = [b"", b"secret\0xyzzy\nbar"]
1156
class TestSetTimeoutCmd(TestPropertyValueCmd):
1157
command = SetTimeoutCmd
1158
propname = "Timeout"
1159
values_to_set = [datetime.timedelta(),
1160
datetime.timedelta(minutes=5),
1161
datetime.timedelta(seconds=1),
1162
datetime.timedelta(weeks=1),
1163
datetime.timedelta(weeks=52)]
1164
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1166
class TestSetExtendedTimeoutCmd(TestPropertyValueCmd):
1167
command = SetExtendedTimeoutCmd
1168
propname = "ExtendedTimeout"
1169
values_to_set = [datetime.timedelta(),
1170
datetime.timedelta(minutes=5),
1171
datetime.timedelta(seconds=1),
1172
datetime.timedelta(weeks=1),
1173
datetime.timedelta(weeks=52)]
1174
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1176
class TestSetIntervalCmd(TestPropertyValueCmd):
1177
command = SetIntervalCmd
1178
propname = "Interval"
1179
values_to_set = [datetime.timedelta(),
1180
datetime.timedelta(minutes=5),
1181
datetime.timedelta(seconds=1),
1182
datetime.timedelta(weeks=1),
1183
datetime.timedelta(weeks=52)]
1184
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1186
class TestSetApprovalDelayCmd(TestPropertyValueCmd):
1187
command = SetApprovalDelayCmd
1188
propname = "ApprovalDelay"
1189
values_to_set = [datetime.timedelta(),
1190
datetime.timedelta(minutes=5),
1191
datetime.timedelta(seconds=1),
1192
datetime.timedelta(weeks=1),
1193
datetime.timedelta(weeks=52)]
1194
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1196
class TestSetApprovalDurationCmd(TestPropertyValueCmd):
1197
command = SetApprovalDurationCmd
1198
propname = "ApprovalDuration"
1199
values_to_set = [datetime.timedelta(),
1200
datetime.timedelta(minutes=5),
1201
datetime.timedelta(seconds=1),
1202
datetime.timedelta(weeks=1),
1203
datetime.timedelta(weeks=52)]
1204
values_to_get = [0, 300000, 1000, 604800000, 31449600000]
1206
class Test_command_from_options(unittest.TestCase):
1208
self.parser = argparse.ArgumentParser()
1209
add_command_line_options(self.parser)
1210
def assert_command_from_args(self, args, command_cls, **cmd_attrs):
1211
"""Assert that parsing ARGS should result in an instance of
1212
COMMAND_CLS with (optionally) all supplied attributes (CMD_ATTRS)."""
1213
options = self.parser.parse_args(args)
1214
check_option_syntax(self.parser, options)
1215
commands = commands_from_options(options)
1216
self.assertEqual(len(commands), 1)
1217
command = commands[0]
1218
self.assertIsInstance(command, command_cls)
1219
for key, value in cmd_attrs.items():
1220
self.assertEqual(getattr(command, key), value)
1221
def test_print_table(self):
1222
self.assert_command_from_args([], PrintTableCmd,
1225
def test_print_table_verbose(self):
1226
self.assert_command_from_args(["--verbose"], PrintTableCmd,
1229
def test_print_table_verbose_short(self):
1230
self.assert_command_from_args(["-v"], PrintTableCmd,
1233
def test_enable(self):
1234
self.assert_command_from_args(["--enable", "foo"], EnableCmd)
1236
def test_enable_short(self):
1237
self.assert_command_from_args(["-e", "foo"], EnableCmd)
1239
def test_disable(self):
1240
self.assert_command_from_args(["--disable", "foo"],
1243
def test_disable_short(self):
1244
self.assert_command_from_args(["-d", "foo"], DisableCmd)
1246
def test_bump_timeout(self):
1247
self.assert_command_from_args(["--bump-timeout", "foo"],
1250
def test_bump_timeout_short(self):
1251
self.assert_command_from_args(["-b", "foo"], BumpTimeoutCmd)
1253
def test_start_checker(self):
1254
self.assert_command_from_args(["--start-checker", "foo"],
1257
def test_stop_checker(self):
1258
self.assert_command_from_args(["--stop-checker", "foo"],
1261
def test_remove(self):
1262
self.assert_command_from_args(["--remove", "foo"],
1265
def test_remove_short(self):
1266
self.assert_command_from_args(["-r", "foo"], RemoveCmd)
1268
def test_checker(self):
1269
self.assert_command_from_args(["--checker", ":", "foo"],
1270
SetCheckerCmd, value_to_set=":")
1272
def test_checker_empty(self):
1273
self.assert_command_from_args(["--checker", "", "foo"],
1274
SetCheckerCmd, value_to_set="")
1276
def test_checker_short(self):
1277
self.assert_command_from_args(["-c", ":", "foo"],
1278
SetCheckerCmd, value_to_set=":")
1280
def test_timeout(self):
1281
self.assert_command_from_args(["--timeout", "PT5M", "foo"],
1283
value_to_set=300000)
1285
def test_timeout_short(self):
1286
self.assert_command_from_args(["-t", "PT5M", "foo"],
1288
value_to_set=300000)
1290
def test_extended_timeout(self):
1291
self.assert_command_from_args(["--extended-timeout", "PT15M",
1293
SetExtendedTimeoutCmd,
1294
value_to_set=900000)
1296
def test_interval(self):
1297
self.assert_command_from_args(["--interval", "PT2M", "foo"],
1299
value_to_set=120000)
1301
def test_interval_short(self):
1302
self.assert_command_from_args(["-i", "PT2M", "foo"],
1304
value_to_set=120000)
1306
def test_approve_by_default(self):
1307
self.assert_command_from_args(["--approve-by-default", "foo"],
1308
ApproveByDefaultCmd)
1310
def test_deny_by_default(self):
1311
self.assert_command_from_args(["--deny-by-default", "foo"],
1314
def test_approval_delay(self):
1315
self.assert_command_from_args(["--approval-delay", "PT30S",
1316
"foo"], SetApprovalDelayCmd,
1319
def test_approval_duration(self):
1320
self.assert_command_from_args(["--approval-duration", "PT1S",
1321
"foo"], SetApprovalDurationCmd,
1324
def test_host(self):
1325
self.assert_command_from_args(["--host", "foo.example.org",
1327
value_to_set="foo.example.org")
1329
def test_host_short(self):
1330
self.assert_command_from_args(["-H", "foo.example.org",
1332
value_to_set="foo.example.org")
1334
def test_secret_devnull(self):
1335
self.assert_command_from_args(["--secret", os.path.devnull,
1336
"foo"], SetSecretCmd,
1339
def test_secret_tempfile(self):
1340
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1341
value = b"secret\0xyzzy\nbar"
1344
self.assert_command_from_args(["--secret", f.name,
1345
"foo"], SetSecretCmd,
1348
def test_secret_devnull_short(self):
1349
self.assert_command_from_args(["-s", os.path.devnull, "foo"],
1350
SetSecretCmd, value_to_set=b"")
1352
def test_secret_tempfile_short(self):
1353
with tempfile.NamedTemporaryFile(mode="r+b") as f:
1354
value = b"secret\0xyzzy\nbar"
1357
self.assert_command_from_args(["-s", f.name, "foo"],
1361
def test_approve(self):
1362
self.assert_command_from_args(["--approve", "foo"],
1365
def test_approve_short(self):
1366
self.assert_command_from_args(["-A", "foo"], ApproveCmd)
1368
def test_deny(self):
1369
self.assert_command_from_args(["--deny", "foo"], DenyCmd)
1371
def test_deny_short(self):
1372
self.assert_command_from_args(["-D", "foo"], DenyCmd)
1374
def test_dump_json(self):
1375
self.assert_command_from_args(["--dump-json"], DumpJSONCmd)
1377
def test_is_enabled(self):
1378
self.assert_command_from_args(["--is-enabled", "foo"],
1381
def test_is_enabled_short(self):
1382
self.assert_command_from_args(["-V", "foo"], IsEnabledCmd)
1384
def test_deny_before_remove(self):
1385
options = self.parser.parse_args(["--deny", "--remove", "foo"])
1386
check_option_syntax(self.parser, options)
1387
commands = commands_from_options(options)
1388
self.assertEqual(len(commands), 2)
1389
self.assertIsInstance(commands[0], DenyCmd)
1390
self.assertIsInstance(commands[1], RemoveCmd)
1392
def test_deny_before_remove_reversed(self):
1393
options = self.parser.parse_args(["--remove", "--deny", "--all"])
1394
check_option_syntax(self.parser, options)
1395
commands = commands_from_options(options)
1396
self.assertEqual(len(commands), 2)
1397
self.assertIsInstance(commands[0], DenyCmd)
1398
self.assertIsInstance(commands[1], RemoveCmd)
1401
class Test_check_option_syntax(unittest.TestCase):
1402
# This mostly corresponds to the definition from has_actions() in
1403
# check_option_syntax()
1405
# The actual values set here are not that important, but we do
1406
# at least stick to the correct types, even though they are
1410
"bump_timeout": True,
1411
"start_checker": True,
1412
"stop_checker": True,
1416
"timeout": datetime.timedelta(),
1417
"extended_timeout": datetime.timedelta(),
1418
"interval": datetime.timedelta(),
1419
"approved_by_default": True,
1420
"approval_delay": datetime.timedelta(),
1421
"approval_duration": datetime.timedelta(),
1423
"secret": io.BytesIO(b"x"),
1429
self.parser = argparse.ArgumentParser()
1430
add_command_line_options(self.parser)
1432
@contextlib.contextmanager
1433
def assertParseError(self):
1434
with self.assertRaises(SystemExit) as e:
1435
with self.temporarily_suppress_stderr():
1437
# Exit code from argparse is guaranteed to be "2". Reference:
1438
# https://docs.python.org/3/library/argparse.html#exiting-methods
1439
self.assertEqual(e.exception.code, 2)
1442
@contextlib.contextmanager
1443
def temporarily_suppress_stderr():
1444
null = os.open(os.path.devnull, os.O_RDWR)
1445
stderrcopy = os.dup(sys.stderr.fileno())
1446
os.dup2(null, sys.stderr.fileno())
1452
os.dup2(stderrcopy, sys.stderr.fileno())
1453
os.close(stderrcopy)
1455
def check_option_syntax(self, options):
1456
check_option_syntax(self.parser, options)
1458
def test_actions_requires_client_or_all(self):
1459
for action, value in self.actions.items():
1460
options = self.parser.parse_args()
1461
setattr(options, action, value)
1462
with self.assertParseError():
1463
self.check_option_syntax(options)
1465
def test_actions_conflicts_with_verbose(self):
1466
for action, value in self.actions.items():
1467
options = self.parser.parse_args()
1468
setattr(options, action, value)
1469
options.verbose = True
1470
with self.assertParseError():
1471
self.check_option_syntax(options)
1473
def test_dump_json_conflicts_with_verbose(self):
1474
options = self.parser.parse_args()
1475
options.dump_json = True
1476
options.verbose = True
1477
with self.assertParseError():
1478
self.check_option_syntax(options)
1480
def test_dump_json_conflicts_with_action(self):
1481
for action, value in self.actions.items():
1482
options = self.parser.parse_args()
1483
setattr(options, action, value)
1484
options.dump_json = True
1485
with self.assertParseError():
1486
self.check_option_syntax(options)
1488
def test_all_can_not_be_alone(self):
1489
options = self.parser.parse_args()
1491
with self.assertParseError():
1492
self.check_option_syntax(options)
1494
def test_all_is_ok_with_any_action(self):
1495
for action, value in self.actions.items():
1496
options = self.parser.parse_args()
1497
setattr(options, action, value)
1499
self.check_option_syntax(options)
1501
def test_is_enabled_fails_without_client(self):
1502
options = self.parser.parse_args()
1503
options.is_enabled = True
1504
with self.assertParseError():
1505
self.check_option_syntax(options)
1507
def test_is_enabled_works_with_one_client(self):
1508
options = self.parser.parse_args()
1509
options.is_enabled = True
1510
options.client = ["foo"]
1511
self.check_option_syntax(options)
1513
def test_is_enabled_fails_with_two_clients(self):
1514
options = self.parser.parse_args()
1515
options.is_enabled = True
1516
options.client = ["foo", "barbar"]
1517
with self.assertParseError():
1518
self.check_option_syntax(options)
1520
def test_remove_can_only_be_combined_with_action_deny(self):
1521
for action, value in self.actions.items():
1522
if action in {"remove", "deny"}:
1524
options = self.parser.parse_args()
1525
setattr(options, action, value)
1527
options.remove = True
1528
with self.assertParseError():
1529
self.check_option_syntax(options)