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