import ciso8601
import time
import unittest
+import parameterized
from . import run_test_server
from arvados._ranges import Range, LocatorAndRange
from arvados.collection import Collection, CollectionReader
from . import arvados_testutil as tutil
+from .arvados_testutil import make_block_cache
class TestResumableWriter(arvados.ResumableCollectionWriter):
KEEP_BLOCK_SIZE = 1024 # PUT to Keep every 1K.
def current_state(self):
return self.dump_state(copy.deepcopy)
-
+@parameterized.parameterized_class([{"disk_cache": True}, {"disk_cache": False}])
class ArvadosCollectionsTest(run_test_server.TestCaseWithServers,
tutil.ArvadosBaseTestCase):
+ disk_cache = False
MAIN_SERVER = {}
@classmethod
run_test_server.authorize_with('admin')
cls.api_client = arvados.api('v1')
cls.keep_client = arvados.KeepClient(api_client=cls.api_client,
- local_store=cls.local_store)
+ local_store=cls.local_store,
+ block_cache=make_block_cache(cls.disk_cache))
def write_foo_bar_baz(self):
cw = arvados.CollectionWriter(self.api_client)
class MockKeep(object):
def __init__(self, content, num_retries=0):
self.content = content
+ self.num_prefetch_threads = 1
def get(self, locator, num_retries=0, prefetch=False):
return self.content[locator]
self.mock_get_collection(client, status, 'foo_file')
return client
- def test_init_no_default_retries(self):
+ def test_init_default_retries(self):
client = self.api_client_mock(200)
reader = arvados.CollectionReader(self.DEFAULT_UUID, api_client=client)
reader.manifest_text()
- client.collections().get().execute.assert_called_with(num_retries=0)
+ client.collections().get().execute.assert_called_with(num_retries=10)
def test_uuid_init_success(self):
client = self.api_client_mock(200)
# Ensure stripped_manifest() doesn't mangle our manifest in
# any way other than stripping hints.
self.assertEqual(
- re.sub('\+[^\d\s\+]+', '', nonnormal),
+ re.sub(r'\+[^\d\s\+]+', '', nonnormal),
reader.stripped_manifest())
# Ensure stripped_manifest() didn't mutate our reader.
self.assertEqual(nonnormal, reader.manifest_text())
c1.save_new()
loc = c1.manifest_locator()
c2 = Collection(loc)
- self.assertEqual(c1.manifest_text, c2.manifest_text)
+ self.assertEqual(c1.manifest_text(strip=True), c2.manifest_text(strip=True))
self.assertEqual(c1.replication_desired, c2.replication_desired)
def test_replication_desired_not_loaded_if_provided(self):
c1.save_new()
loc = c1.manifest_locator()
c2 = Collection(loc, replication_desired=2)
- self.assertEqual(c1.manifest_text, c2.manifest_text)
+ self.assertEqual(c1.manifest_text(strip=True), c2.manifest_text(strip=True))
self.assertNotEqual(c1.replication_desired, c2.replication_desired)
def test_storage_classes_desired_kept_on_load(self):
c1.save_new()
loc = c1.manifest_locator()
c2 = Collection(loc)
- self.assertEqual(c1.manifest_text, c2.manifest_text)
+ self.assertEqual(c1.manifest_text(strip=True), c2.manifest_text(strip=True))
self.assertEqual(c1.storage_classes_desired(), c2.storage_classes_desired())
def test_storage_classes_change_after_save(self):
c2.save(storage_classes=['highIO'])
self.assertEqual(['highIO'], c2.storage_classes_desired())
c3 = Collection(loc)
- self.assertEqual(c1.manifest_text, c3.manifest_text)
+ self.assertEqual(c1.manifest_text(strip=True), c3.manifest_text(strip=True))
self.assertEqual(['highIO'], c3.storage_classes_desired())
def test_storage_classes_desired_not_loaded_if_provided(self):
c1.save_new()
loc = c1.manifest_locator()
c2 = Collection(loc, storage_classes_desired=['default'])
- self.assertEqual(c1.manifest_text, c2.manifest_text)
+ self.assertEqual(c1.manifest_text(strip=True), c2.manifest_text(strip=True))
self.assertNotEqual(c1.storage_classes_desired(), c2.storage_classes_desired())
def test_init_manifest(self):