sdk/python/tests/test_cache.py

   1 from __future__ import print_function
   2
   3 import md5
   4 import mock
   5 import shutil
   6 import random
   7 import sys
   8 import tempfile
   9 import threading
  10 import unittest
  11
  12 import arvados.cache
  13 import run_test_server
  14
  15
  16 def _random(n):
  17     return bytearray(random.getrandbits(8) for _ in xrange(n))
  18
  19
  20 class CacheTestThread(threading.Thread):
  21     def __init__(self, dir):
  22         super(CacheTestThread, self).__init__()
  23         self._dir = dir
  24
  25     def run(self):
  26         c = arvados.cache.SafeHTTPCache(self._dir)
  27         url = 'http://example.com/foo'
  28         self.ok = True
  29         for x in range(16):
  30             try:
  31                 data_in = _random(128)
  32                 data_in = md5.new(data_in).hexdigest() + "\n" + str(data_in)
  33                 c.set(url, data_in)
  34                 data_out = c.get(url)
  35                 digest, content = data_out.split("\n", 1)
  36                 if digest != md5.new(content).hexdigest():
  37                     self.ok = False
  38             except Exception as err:
  39                 self.ok = False
  40                 print("cache failed: {}".format(err), file=sys.stderr)
  41
  42
  43 class CacheTest(unittest.TestCase):
  44     def setUp(self):
  45         self._dir = tempfile.mkdtemp()
  46
  47     def tearDown(self):
  48         shutil.rmtree(self._dir)
  49
  50     def test_cache_crud(self):
  51         c = arvados.cache.SafeHTTPCache(self._dir, max_age=0)
  52         url = 'https://example.com/foo?bar=baz'
  53         data1 = _random(256)
  54         data2 = _random(128)
  55         self.assertEqual(None, c.get(url))
  56         c.delete(url)
  57         c.set(url, data1)
  58         self.assertEqual(data1, c.get(url))
  59         c.delete(url)
  60         self.assertEqual(None, c.get(url))
  61         c.set(url, data1)
  62         c.set(url, data2)
  63         self.assertEqual(data2, c.get(url))
  64
  65     def test_cache_threads(self):
  66         threads = []
  67         for _ in range(64):
  68             t = CacheTestThread(dir=self._dir)
  69             t.start()
  70             threads.append(t)
  71         for t in threads:
  72             t.join()
  73             self.assertTrue(t.ok)
  74
  75
  76 class CacheIntegrationTest(run_test_server.TestCaseWithServers):
  77     MAIN_SERVER = {}
  78
  79     def test_cache_used_by_default_client(self):
  80         with mock.patch('arvados.cache.SafeHTTPCache.get') as getter:
  81             arvados.api('v1')._rootDesc.get('foobar')
  82             getter.assert_called()