126
110
datetime.timedelta(0, 330)
127
111
>>> rfc3339_duration_to_delta("P1DT3M20S")
128
112
datetime.timedelta(1, 200)
129
>>> # Can not be empty:
130
>>> rfc3339_duration_to_delta("")
131
Traceback (most recent call last):
133
ValueError: Invalid RFC 3339 duration: u''
134
>>> # Must start with "P":
135
>>> rfc3339_duration_to_delta("1D")
136
Traceback (most recent call last):
138
ValueError: Invalid RFC 3339 duration: u'1D'
139
>>> # Must use correct order
140
>>> rfc3339_duration_to_delta("PT1S2M")
141
Traceback (most recent call last):
143
ValueError: Invalid RFC 3339 duration: u'PT1S2M'
144
>>> # Time needs time marker
145
>>> rfc3339_duration_to_delta("P1H2S")
146
Traceback (most recent call last):
148
ValueError: Invalid RFC 3339 duration: u'P1H2S'
149
>>> # Weeks can not be combined with anything else
150
>>> rfc3339_duration_to_delta("P1D2W")
151
Traceback (most recent call last):
153
ValueError: Invalid RFC 3339 duration: u'P1D2W'
154
>>> rfc3339_duration_to_delta("P2W2H")
155
Traceback (most recent call last):
157
ValueError: Invalid RFC 3339 duration: u'P2W2H'
160
115
# Parsing an RFC 3339 duration with regular expressions is not
240
195
def string_to_delta(interval):
241
196
"""Parse a string and return a datetime.timedelta
198
>>> string_to_delta('7d')
199
datetime.timedelta(7)
200
>>> string_to_delta('60s')
201
datetime.timedelta(0, 60)
202
>>> string_to_delta('60m')
203
datetime.timedelta(0, 3600)
204
>>> string_to_delta('24h')
205
datetime.timedelta(1)
206
>>> string_to_delta('1w')
207
datetime.timedelta(7)
208
>>> string_to_delta('5m 30s')
209
datetime.timedelta(0, 330)
245
213
return rfc3339_duration_to_delta(interval)
246
except ValueError as e:
247
log.warning("%s - Parsing as pre-1.6.1 interval instead",
249
return parse_pre_1_6_1_interval(interval)
252
def parse_pre_1_6_1_interval(interval):
253
"""Parse an interval string as documented by Mandos before 1.6.1, and
254
return a datetime.timedelta
255
>>> parse_pre_1_6_1_interval('7d')
256
datetime.timedelta(7)
257
>>> parse_pre_1_6_1_interval('60s')
258
datetime.timedelta(0, 60)
259
>>> parse_pre_1_6_1_interval('60m')
260
datetime.timedelta(0, 3600)
261
>>> parse_pre_1_6_1_interval('24h')
262
datetime.timedelta(1)
263
>>> parse_pre_1_6_1_interval('1w')
264
datetime.timedelta(7)
265
>>> parse_pre_1_6_1_interval('5m 30s')
266
datetime.timedelta(0, 330)
267
>>> parse_pre_1_6_1_interval('')
268
datetime.timedelta(0)
269
>>> # Ignore unknown characters, allow any order and repetitions
270
>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')
271
datetime.timedelta(2, 480, 18000)
275
217
value = datetime.timedelta(0)
276
218
regexp = re.compile(r"(\d+)([dsmhw]?)")
294
236
def print_clients(clients, keywords):
295
print('\n'.join(TableOfClients(clients, keywords).rows()))
297
class TableOfClients(object):
298
def __init__(self, clients, keywords):
299
self.clients = clients
300
self.keywords = keywords
303
237
def valuetostring(value, keyword):
304
if isinstance(value, dbus.Boolean):
238
if type(value) is dbus.Boolean:
305
239
return "Yes" if value else "No"
306
240
if keyword in ("Timeout", "Interval", "ApprovalDelay",
307
241
"ApprovalDuration", "ExtendedTimeout"):
308
242
return milliseconds_to_string(value)
309
243
return str(value)
312
# Create format string to format table rows
313
format_string = " ".join("{{{key}:{width}}}".format(
314
width=max(len(tablewords[key]),
315
max(len(self.valuetostring(client[key], key))
316
for client in self.clients)),
318
for key in self.keywords)
319
# Start with header line
320
rows = [format_string.format(**tablewords)]
321
for client in self.clients:
322
rows.append(format_string
323
.format(**{key: self.valuetostring(client[key], key)
324
for key in self.keywords}))
245
# Create format string to print table rows
246
format_string = " ".join("{{{key}:{width}}}".format(
247
width=max(len(tablewords[key]),
248
max(len(valuetostring(client[key], key))
249
for client in clients)),
253
print(format_string.format(**tablewords))
254
for client in clients:
256
.format(**{key: valuetostring(client[key], key)
257
for key in keywords}))
328
260
def has_actions(options):
459
397
clients[client_objc] = client
462
log.critical("Client not found on server: %r", name)
400
print("Client not found on server: {!r}"
401
.format(name), file=sys.stderr)
465
404
if not has_actions(options) and clients:
466
405
if options.verbose or options.dump_json:
467
406
keywords = ("Name", "Enabled", "Timeout", "LastCheckedOK",
468
"Created", "Interval", "Host", "KeyID",
469
"Fingerprint", "CheckerRunning",
470
"LastEnabled", "ApprovalPending",
471
"ApprovedByDefault", "LastApprovalRequest",
472
"ApprovalDelay", "ApprovalDuration",
473
"Checker", "ExtendedTimeout", "Expires",
407
"Created", "Interval", "Host", "Fingerprint",
408
"CheckerRunning", "LastEnabled",
409
"ApprovalPending", "ApprovedByDefault",
410
"LastApprovalRequest", "ApprovalDelay",
411
"ApprovalDuration", "Checker",
412
"ExtendedTimeout", "Expires",
474
413
"LastCheckerStatus")
476
415
keywords = defaultkeywords
553
492
client.Approve(dbus.Boolean(False),
554
493
dbus_interface=client_interface)
557
class Test_milliseconds_to_string(unittest.TestCase):
559
self.assertEqual(milliseconds_to_string(93785000),
561
def test_no_days(self):
562
self.assertEqual(milliseconds_to_string(7385000), "02:03:05")
563
def test_all_zero(self):
564
self.assertEqual(milliseconds_to_string(0), "00:00:00")
565
def test_no_fractional_seconds(self):
566
self.assertEqual(milliseconds_to_string(400), "00:00:00")
567
self.assertEqual(milliseconds_to_string(900), "00:00:00")
568
self.assertEqual(milliseconds_to_string(1900), "00:00:01")
570
class Test_string_to_delta(unittest.TestCase):
571
def test_handles_basic_rfc3339(self):
572
self.assertEqual(string_to_delta("PT2H"),
573
datetime.timedelta(0, 7200))
574
def test_falls_back_to_pre_1_6_1_with_warning(self):
575
# assertLogs only exists in Python 3.4
576
if hasattr(self, "assertLogs"):
577
with self.assertLogs(log, logging.WARNING):
578
value = string_to_delta("2h")
580
value = string_to_delta("2h")
581
self.assertEqual(value, datetime.timedelta(0, 7200))
583
class Test_TableOfClients(unittest.TestCase):
586
self.old_tablewords = tablewords
591
"Bool": "A D-BUS Boolean",
592
"NonDbusBoolean": "A Non-D-BUS Boolean",
593
"Integer": "An Integer",
594
"Timeout": "Timedelta 1",
595
"Interval": "Timedelta 2",
596
"ApprovalDelay": "Timedelta 3",
597
"ApprovalDuration": "Timedelta 4",
598
"ExtendedTimeout": "Timedelta 5",
599
"String": "A String",
601
self.keywords = ["Attr1", "AttrTwo"]
607
"Bool": dbus.Boolean(False),
608
"NonDbusBoolean": False,
612
"ApprovalDelay": 2000,
613
"ApprovalDuration": 3000,
614
"ExtendedTimeout": 4000,
621
"Bool": dbus.Boolean(True),
622
"NonDbusBoolean": True,
625
"Interval": 93786000,
626
"ApprovalDelay": 93787000,
627
"ApprovalDuration": 93788000,
628
"ExtendedTimeout": 93789000,
629
"String": "A huge string which will not fit," * 10,
634
tablewords = self.old_tablewords
635
def test_short_header(self):
636
rows = TableOfClients(self.clients, self.keywords).rows()
641
self.assertEqual(rows, expected_rows)
642
def test_booleans(self):
643
keywords = ["Bool", "NonDbusBoolean"]
644
rows = TableOfClients(self.clients, keywords).rows()
646
"A D-BUS Boolean A Non-D-BUS Boolean",
650
self.assertEqual(rows, expected_rows)
651
def test_milliseconds_detection(self):
652
keywords = ["Integer", "Timeout", "Interval", "ApprovalDelay",
653
"ApprovalDuration", "ExtendedTimeout"]
654
rows = TableOfClients(self.clients, keywords).rows()
656
An Integer Timedelta 1 Timedelta 2 Timedelta 3 Timedelta 4 Timedelta 5
657
0 00:00:00 00:00:01 00:00:02 00:00:03 00:00:04
658
1 1T02:03:05 1T02:03:06 1T02:03:07 1T02:03:08 1T02:03:09
661
self.assertEqual(rows, expected_rows)
662
def test_empty_and_long_string_values(self):
663
keywords = ["String"]
664
rows = TableOfClients(self.clients, keywords).rows()
668
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,
671
self.assertEqual(rows, expected_rows)
675
def should_only_run_tests():
676
parser = argparse.ArgumentParser(add_help=False)
677
parser.add_argument("--check", action='store_true')
678
args, unknown_args = parser.parse_known_args()
679
run_tests = args.check
681
# Remove --check argument from sys.argv
682
sys.argv[1:] = unknown_args
685
# Add all tests from doctest strings
686
def load_tests(loader, tests, none):
688
tests.addTests(doctest.DocTestSuite())
691
496
if __name__ == "__main__":
692
if should_only_run_tests():
693
# Call using ./tdd-python-script --check [--verbose]