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 os
  12 import random
  13 import shutil
  14 import sys
  15 import tempfile
  16 import threading
  17 import unittest
  18
  19 from unittest import mock
  20
  21 import arvados
  22 import arvados.cache
  23 from . import run_test_server
  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()