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