from future import standard_library
standard_library.install_aliases()
-from builtins import str
-from builtins import range
-from builtins import object
import arvados
import contextlib
import errno
import http.client
import httplib2
import io
-import mock
import os
import pycurl
import queue
import tempfile
import unittest
-if sys.version_info >= (3, 0):
- from io import StringIO, BytesIO
-else:
- from cStringIO import StringIO
- BytesIO = StringIO
+from io import StringIO, BytesIO
+from unittest import mock
# Use this hostname when you want to make sure the traffic will be
# instantly refused. 100::/64 is a dedicated black hole.
return mock.patch('httplib2.Http.request', side_effect=queue_with((
(fake_httplib2_response(code, **headers), body) for code in codes)))
-def mock_api_responses(api_client, body, codes, headers={}):
+def mock_api_responses(api_client, body, codes, headers={}, method='request'):
if not isinstance(body, bytes) and hasattr(body, 'encode'):
body = body.encode()
- return mock.patch.object(api_client._http, 'request', side_effect=queue_with((
+ return mock.patch.object(api_client._http, method, side_effect=queue_with((
(fake_httplib2_response(code, **headers), body) for code in codes)))
def str_keep_locator(s):
shutil.rmtree(disk_cache_dir, ignore_errors=True)
block_cache = arvados.keep.KeepBlockCache(disk_cache=disk_cache)
return block_cache
+
+
+class DiskCacheBase:
+ def make_block_cache(self, disk_cache):
+ self.disk_cache_dir = tempfile.mkdtemp() if disk_cache else None
+ block_cache = arvados.keep.KeepBlockCache(disk_cache=disk_cache,
+ disk_cache_dir=self.disk_cache_dir)
+ return block_cache
+
+ def tearDown(self):
+ if self.disk_cache_dir:
+ shutil.rmtree(self.disk_cache_dir)