sdk/python/tests/test_cache.py

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