sdk/python/tests/test_keep_locator.py

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