sdk/python/tests/test_cache.py

   1 import md5
   2 import shutil
   3 import random
   4 import tempfile
   5 import threading
   6 import unittest
   7
   8 import arvados.cache
   9
  10
  11 def _random(n):
  12     return bytearray(random.getrandbits(8) for _ in xrange(n))
  13
  14
  15 class CacheTestThread(threading.Thread):
  16     def __init__(self, dir):
  17         super(CacheTestThread, self).__init__()
  18         self._dir = dir
  19
  20     def run(self):
  21         c = arvados.cache.SafeHTTPCache(self._dir)
  22         url = 'http://example.com/foo'
  23         for x in range(16):
  24             data_in = _random(128)
  25             data_in = md5.new(data_in).hexdigest() + "\n" + str(data_in)
  26             c.set(url, data_in)
  27             data_out = c.get(url)
  28             digest, content = data_out.split("\n", 1)
  29             self.ok = (digest == md5.new(content).hexdigest())
  30
  31
  32 class CacheTest(unittest.TestCase):
  33     def setUp(self):
  34         self._dir = tempfile.mkdtemp()
  35
  36     def tearDown(self):
  37         shutil.rmtree(self._dir)
  38
  39     def test_cache_crud(self):
  40         c = arvados.cache.SafeHTTPCache(self._dir)
  41         url = 'https://example.com/foo?bar=baz'
  42         data1 = _random(256)
  43         data2 = _random(128)
  44         self.assertEqual(None, c.get(url))
  45         c.delete(url)
  46         c.set(url, data1)
  47         self.assertEqual(data1, c.get(url))
  48         c.delete(url)
  49         self.assertEqual(None, c.get(url))
  50         c.set(url, data1)
  51         c.set(url, data2)
  52         self.assertEqual(data2, c.get(url))
  53
  54     def test_cache_threads(self):
  55         threads = []
  56         for _ in range(64):
  57             t = CacheTestThread(dir=self._dir)
  58             t.start()
  59             threads.append(t)
  60         for t in threads:
  61             t.join()
  62             self.assertTrue(t.ok)