sdk/python/tests/test_cache.py

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