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 ArvadosPutResumeCacheTest(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 perm_hints(self, count=DEFAULT_TEST_COUNT):
  26         for sig, ts in itertools.izip(self.signatures(count),
  27                                       self.timestamps(count)):
  28             yield 'A{}@{}'.format(sig, ts)
  29
  30     def test_good_locators_returned(self):
  31         for hint_gens in [(), (self.sizes(),), (self.perm_hints(),),
  32                           (self.sizes(), self.perm_hints())]:
  33             for loc_data in itertools.izip(self.checksums(), *hint_gens):
  34                 locator = '+'.join(loc_data)
  35                 self.assertEquals(locator, str(KeepLocator(locator)))
  36
  37     def test_nonchecksum_rejected(self):
  38         for badstr in ['', 'badbadbad', '8f9e68d957b504a29ba76c526c3145dj',
  39                        '+8f9e68d957b504a29ba76c526c3145d9',
  40                        '3+8f9e68d957b504a29ba76c526c3145d9']:
  41             self.assertRaises(ValueError, KeepLocator, badstr)
  42
  43     def test_bad_hints_rejected(self):
  44         checksum = next(self.checksums(1))
  45         for badhint in ['', 'nonsense', '+32', checksum]:
  46             self.assertRaises(ValueError, KeepLocator,
  47                               '+'.join([checksum, badhint]))
  48
  49     def test_expiry_passed(self):
  50         checksum = next(self.checksums(1))
  51         signature = next(self.signatures(1))
  52         dt1980 = datetime.datetime(1980, 1, 1)
  53         dt2000 = datetime.datetime(2000, 2, 2)
  54         dt2080 = datetime.datetime(2080, 3, 3)
  55         locator = KeepLocator(checksum)
  56         self.assertFalse(locator.permission_expired())
  57         self.assertFalse(locator.permission_expired(dt1980))
  58         self.assertFalse(locator.permission_expired(dt2080))
  59         # Timestamped to 1987-01-05 18:48:32.
  60         locator = KeepLocator('{}+A{}@20000000'.format(checksum, signature))
  61         self.assertTrue(locator.permission_expired())
  62         self.assertTrue(locator.permission_expired(dt2000))
  63         self.assertFalse(locator.permission_expired(dt1980))
  64
  65
  66 if __name__ == '__main__':
  67     unittest.main()