-import md5
-import shutil
+# Copyright (C) The Arvados Authors. All rights reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import print_function
+from __future__ import absolute_import
+
+from builtins import str
+from builtins import range
+import hashlib
+import mock
+import os
import random
+import shutil
+import sys
import tempfile
import threading
import unittest
+import arvados
import arvados.cache
+from . import run_test_server
def _random(n):
- return bytearray(random.getrandbits(8) for _ in xrange(n))
+ return bytearray(random.getrandbits(8) for _ in range(n))
class CacheTestThread(threading.Thread):
def run(self):
c = arvados.cache.SafeHTTPCache(self._dir)
url = 'http://example.com/foo'
+ self.ok = True
for x in range(16):
- data_in = _random(128)
- data_in = md5.new(data_in).hexdigest() + "\n" + str(data_in)
- c.set(url, data_in)
- data_out = c.get(url)
- digest, content = data_out.split("\n", 1)
- self.ok = (digest == md5.new(content).hexdigest())
+ try:
+ data_in = _random(128)
+ data_in = hashlib.md5(data_in).hexdigest().encode() + b"\n" + data_in
+ c.set(url, data_in)
+ data_out = c.get(url)
+ digest, _, content = data_out.partition(b"\n")
+ if digest != hashlib.md5(content).hexdigest().encode():
+ self.ok = False
+ except Exception as err:
+ self.ok = False
+ print("cache failed: {}: {}".format(type(err), err), file=sys.stderr)
+ raise
class CacheTest(unittest.TestCase):
def tearDown(self):
shutil.rmtree(self._dir)
+ def test_cache_create_error(self):
+ _, filename = tempfile.mkstemp()
+ home_was = os.environ['HOME']
+ os.environ['HOME'] = filename
+ try:
+ c = arvados.http_cache('test')
+ self.assertEqual(None, c)
+ finally:
+ os.environ['HOME'] = home_was
+ os.unlink(filename)
+
def test_cache_crud(self):
- c = arvados.cache.SafeHTTPCache(self._dir)
+ c = arvados.cache.SafeHTTPCache(self._dir, max_age=0)
url = 'https://example.com/foo?bar=baz'
data1 = _random(256)
data2 = _random(128)
for t in threads:
t.join()
self.assertTrue(t.ok)
+
+
+class CacheIntegrationTest(run_test_server.TestCaseWithServers):
+ MAIN_SERVER = {}
+
+ def test_cache_used_by_default_client(self):
+ with mock.patch('arvados.cache.SafeHTTPCache.get') as getter:
+ arvados.api('v1')._rootDesc.get('foobar')
+ getter.assert_called()