sdk/python/tests/test_cache.py

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