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):