Merge branch 'master' of git.curoverse.com:arvados into 11876-r-sdk
[arvados.git] / sdk / python / tests / test_keep_locator.py
1 # Copyright (C) The Arvados Authors. All rights reserved.
2 #
3 # SPDX-License-Identifier: Apache-2.0
4
5 from builtins import next
6 from builtins import zip
7 from builtins import str
8 from builtins import range
9 import datetime
10 import itertools
11 import random
12 import unittest
13
14 from arvados.keep import KeepLocator
15
16 class ArvadosKeepLocatorTest(unittest.TestCase):
17     DEFAULT_TEST_COUNT = 10
18
19     def numstrs(fmtstr, base, exponent):
20         def genstrs(self, count=None):
21             return (fmtstr.format(random.randint(0, base ** exponent))
22                     for c in range(count or self.DEFAULT_TEST_COUNT))
23         return genstrs
24
25     checksums = numstrs('{:032x}', 16, 32)
26     sizes = numstrs('{:d}', 2, 26)
27     signatures = numstrs('{:040x}', 16, 40)
28     timestamps = numstrs('{:08x}', 16, 8)
29
30     def base_locators(self, count=DEFAULT_TEST_COUNT):
31         return ('+'.join(pair) for pair in
32                 zip(self.checksums(count), self.sizes(count)))
33
34     def perm_hints(self, count=DEFAULT_TEST_COUNT):
35         for sig, ts in zip(self.signatures(count),
36                                       self.timestamps(count)):
37             yield 'A{}@{}'.format(sig, ts)
38
39     def test_good_locators_returned(self):
40         for hint_gens in [(), (self.sizes(),),
41                           (self.sizes(), self.perm_hints())]:
42             for loc_data in zip(self.checksums(), *hint_gens):
43                 locator = '+'.join(loc_data)
44                 self.assertEqual(locator, str(KeepLocator(locator)))
45
46     def test_nonchecksum_rejected(self):
47         for badstr in ['', 'badbadbad', '8f9e68d957b504a29ba76c526c3145dj',
48                        '+8f9e68d957b504a29ba76c526c3145d9',
49                        '3+8f9e68d957b504a29ba76c526c3145d9']:
50             self.assertRaises(ValueError, KeepLocator, badstr)
51
52     def test_unknown_hints_accepted(self):
53         base = next(self.base_locators(1))
54         for weirdhint in ['Zfoo', 'Ybar234', 'Xa@b_c-372', 'W99']:
55             locator = '+'.join([base, weirdhint])
56             self.assertEqual(locator, str(KeepLocator(locator)))
57
58     def test_bad_hints_rejected(self):
59         base = next(self.base_locators(1))
60         for badhint in ['', 'A', 'lowercase', '+32']:
61             self.assertRaises(ValueError, KeepLocator,
62                               '+'.join([base, badhint]))
63
64     def test_multiple_locator_hints_accepted(self):
65         base = next(self.base_locators(1))
66         for loc_hints in itertools.permutations(['Kab1cd', 'Kef2gh', 'Kij3kl']):
67             locator = '+'.join((base,) + loc_hints)
68             self.assertEqual(locator, str(KeepLocator(locator)))
69
70     def test_str_type(self):
71         base = next(self.base_locators(1))
72         locator = KeepLocator(base)
73         self.assertEqual(type(''), type(locator.__str__()))
74
75     def test_expiry_passed(self):
76         base = next(self.base_locators(1))
77         signature = next(self.signatures(1))
78         dt1980 = datetime.datetime(1980, 1, 1)
79         dt2000 = datetime.datetime(2000, 2, 2)
80         dt2080 = datetime.datetime(2080, 3, 3)
81         locator = KeepLocator(base)
82         self.assertFalse(locator.permission_expired())
83         self.assertFalse(locator.permission_expired(dt1980))
84         self.assertFalse(locator.permission_expired(dt2080))
85         # Timestamped to 1987-01-05 18:48:32.
86         locator = KeepLocator('{}+A{}@20000000'.format(base, signature))
87         self.assertTrue(locator.permission_expired())
88         self.assertTrue(locator.permission_expired(dt2000))
89         self.assertFalse(locator.permission_expired(dt1980))
90
91
92 if __name__ == '__main__':
93     unittest.main()