1 from __future__ import print_function
2 from __future__ import absolute_import
4 from builtins import str
5 from builtins import range
18 from . import run_test_server
22 return bytearray(random.getrandbits(8) for _ in range(n))
25 class CacheTestThread(threading.Thread):
26 def __init__(self, dir):
27 super(CacheTestThread, self).__init__()
31 c = arvados.cache.SafeHTTPCache(self._dir)
32 url = 'http://example.com/foo'
36 data_in = _random(128)
37 data_in = md5.new(data_in).hexdigest() + bytes("\n") + bytes(data_in)
40 digest, _, content = data_out.partition("\n")
41 if digest != md5.new(content).hexdigest():
43 except Exception as err:
45 print("cache failed: {}".format(err), file=sys.stderr)
48 class CacheTest(unittest.TestCase):
50 self._dir = tempfile.mkdtemp()
53 shutil.rmtree(self._dir)
55 def test_cache_create_error(self):
56 _, filename = tempfile.mkstemp()
57 home_was = os.environ['HOME']
58 os.environ['HOME'] = filename
60 c = arvados.http_cache('test')
61 self.assertEqual(None, c)
63 os.environ['HOME'] = home_was
66 def test_cache_crud(self):
67 c = arvados.cache.SafeHTTPCache(self._dir, max_age=0)
68 url = 'https://example.com/foo?bar=baz'
71 self.assertEqual(None, c.get(url))
74 self.assertEqual(data1, c.get(url))
76 self.assertEqual(None, c.get(url))
79 self.assertEqual(data2, c.get(url))
81 def test_cache_threads(self):
84 t = CacheTestThread(dir=self._dir)
92 class CacheIntegrationTest(run_test_server.TestCaseWithServers):
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()