sdk/python/tests/test_cache.py

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