1 from __future__ import print_function
15 import run_test_server
19 return bytearray(random.getrandbits(8) for _ in xrange(n))
22 class CacheTestThread(threading.Thread):
23 def __init__(self, dir):
24 super(CacheTestThread, self).__init__()
28 c = arvados.cache.SafeHTTPCache(self._dir)
29 url = 'http://example.com/foo'
33 data_in = _random(128)
34 data_in = md5.new(data_in).hexdigest() + "\n" + str(data_in)
37 digest, content = data_out.split("\n", 1)
38 if digest != md5.new(content).hexdigest():
40 except Exception as err:
42 print("cache failed: {}".format(err), file=sys.stderr)
45 class CacheTest(unittest.TestCase):
47 self._dir = tempfile.mkdtemp()
50 shutil.rmtree(self._dir)
52 def test_cache_create_error(self):
53 _, filename = tempfile.mkstemp()
54 home_was = os.environ['HOME']
55 os.environ['HOME'] = filename
57 c = arvados.http_cache('test')
58 self.assertEqual(None, c)
60 os.environ['HOME'] = home_was
63 def test_cache_crud(self):
64 c = arvados.cache.SafeHTTPCache(self._dir, max_age=0)
65 url = 'https://example.com/foo?bar=baz'
68 self.assertEqual(None, c.get(url))
71 self.assertEqual(data1, c.get(url))
73 self.assertEqual(None, c.get(url))
76 self.assertEqual(data2, c.get(url))
78 def test_cache_threads(self):
81 t = CacheTestThread(dir=self._dir)
89 class CacheIntegrationTest(run_test_server.TestCaseWithServers):
92 def test_cache_used_by_default_client(self):
93 with mock.patch('arvados.cache.SafeHTTPCache.get') as getter:
94 arvados.api('v1')._rootDesc.get('foobar')
95 getter.assert_called()