2 # -*- coding: utf-8 -*-
4 from builtins import next
5 from builtins import zip
6 from builtins import str
7 from builtins import range
13 from arvados.keep import KeepLocator
15 class ArvadosKeepLocatorTest(unittest.TestCase):
16 DEFAULT_TEST_COUNT = 10
18 def numstrs(fmtstr, base, exponent):
19 def genstrs(self, count=None):
20 return (fmtstr.format(random.randint(0, base ** exponent))
21 for c in range(count or self.DEFAULT_TEST_COUNT))
24 checksums = numstrs('{:032x}', 16, 32)
25 sizes = numstrs('{:d}', 2, 26)
26 signatures = numstrs('{:040x}', 16, 40)
27 timestamps = numstrs('{:08x}', 16, 8)
29 def base_locators(self, count=DEFAULT_TEST_COUNT):
30 return ('+'.join(pair) for pair in
31 zip(self.checksums(count), self.sizes(count)))
33 def perm_hints(self, count=DEFAULT_TEST_COUNT):
34 for sig, ts in zip(self.signatures(count),
35 self.timestamps(count)):
36 yield 'A{}@{}'.format(sig, ts)
38 def test_good_locators_returned(self):
39 for hint_gens in [(), (self.sizes(),),
40 (self.sizes(), self.perm_hints())]:
41 for loc_data in zip(self.checksums(), *hint_gens):
42 locator = '+'.join(loc_data)
43 self.assertEqual(locator, str(KeepLocator(locator)))
45 def test_nonchecksum_rejected(self):
46 for badstr in ['', 'badbadbad', '8f9e68d957b504a29ba76c526c3145dj',
47 '+8f9e68d957b504a29ba76c526c3145d9',
48 '3+8f9e68d957b504a29ba76c526c3145d9']:
49 self.assertRaises(ValueError, KeepLocator, badstr)
51 def test_unknown_hints_accepted(self):
52 base = next(self.base_locators(1))
53 for weirdhint in ['Zfoo', 'Ybar234', 'Xa@b_c-372', 'W99']:
54 locator = '+'.join([base, weirdhint])
55 self.assertEqual(locator, str(KeepLocator(locator)))
57 def test_bad_hints_rejected(self):
58 base = next(self.base_locators(1))
59 for badhint in ['', 'A', 'lowercase', '+32']:
60 self.assertRaises(ValueError, KeepLocator,
61 '+'.join([base, badhint]))
63 def test_multiple_locator_hints_accepted(self):
64 base = next(self.base_locators(1))
65 for loc_hints in itertools.permutations(['Kab1cd', 'Kef2gh', 'Kij3kl']):
66 locator = '+'.join((base,) + loc_hints)
67 self.assertEqual(locator, str(KeepLocator(locator)))
69 def test_expiry_passed(self):
70 base = next(self.base_locators(1))
71 signature = next(self.signatures(1))
72 dt1980 = datetime.datetime(1980, 1, 1)
73 dt2000 = datetime.datetime(2000, 2, 2)
74 dt2080 = datetime.datetime(2080, 3, 3)
75 locator = KeepLocator(base)
76 self.assertFalse(locator.permission_expired())
77 self.assertFalse(locator.permission_expired(dt1980))
78 self.assertFalse(locator.permission_expired(dt2080))
79 # Timestamped to 1987-01-05 18:48:32.
80 locator = KeepLocator('{}+A{}@20000000'.format(base, signature))
81 self.assertTrue(locator.permission_expired())
82 self.assertTrue(locator.permission_expired(dt2000))
83 self.assertFalse(locator.permission_expired(dt1980))
86 if __name__ == '__main__':