import copy
import mock
import os
-import pprint
import random
import re
import sys
-import tempfile
import datetime
import ciso8601
import time
self.assertEqual(c1.manifest_text, c2.manifest_text)
self.assertNotEqual(c1.replication_desired, c2.replication_desired)
+ def test_storage_classes_desired_kept_on_load(self):
+ m = '. 781e5e245d69b566979b86e28d23f2c7+10 0:10:count1.txt 0:10:count2.txt\n'
+ c1 = Collection(m, storage_classes_desired=['archival'])
+ c1.save_new()
+ loc = c1.manifest_locator()
+ c2 = Collection(loc)
+ self.assertEqual(c1.manifest_text, c2.manifest_text)
+ self.assertEqual(c1.storage_classes_desired(), c2.storage_classes_desired())
+
+ def test_storage_classes_change_after_save(self):
+ m = '. 781e5e245d69b566979b86e28d23f2c7+10 0:10:count1.txt 0:10:count2.txt\n'
+ c1 = Collection(m, storage_classes_desired=['archival'])
+ c1.save_new()
+ loc = c1.manifest_locator()
+ c2 = Collection(loc)
+ self.assertEqual(['archival'], c2.storage_classes_desired())
+ 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(['highIO'], c3.storage_classes_desired())
+
+ def test_storage_classes_desired_not_loaded_if_provided(self):
+ m = '. 781e5e245d69b566979b86e28d23f2c7+10 0:10:count1.txt 0:10:count2.txt\n'
+ c1 = Collection(m, storage_classes_desired=['archival'])
+ c1.save_new()
+ loc = c1.manifest_locator()
+ c2 = Collection(loc, storage_classes_desired=['default'])
+ self.assertEqual(c1.manifest_text, c2.manifest_text)
+ self.assertNotEqual(c1.storage_classes_desired(), c2.storage_classes_desired())
+
def test_init_manifest(self):
m1 = """. 5348b82a029fd9e971a811ce1f71360b+43 0:43:md5sum.txt
. 085c37f02916da1cad16f93c54d899b7+41 0:41:md5sum.txt
c = Collection(manifest)
self.assertEqual(c.portable_manifest_text(), manifest)
+ def test_other_special_chars_on_file_token(self):
+ cases = [
+ ('\\000', '\0'),
+ ('\\011', '\t'),
+ ('\\012', '\n'),
+ ('\\072', ':'),
+ ('\\134400', '\\400'),
+ ]
+ for encoded, decoded in cases:
+ manifest = '. d41d8cd98f00b204e9800998ecf8427e+0 0:0:some%sfile.txt\n' % encoded
+ c = Collection(manifest)
+ self.assertEqual(c.portable_manifest_text(), manifest)
+ self.assertIn('some%sfile.txt' % decoded, c.keys())
+
def test_escaped_paths_do_get_unescaped_on_listing(self):
# Dir & file names are literally '\056' (escaped form: \134056)
manifest = './\\134056\\040Test d41d8cd98f00b204e9800998ecf8427e+0 0:0:\\134056\n'
def setUp(self):
self.keep_put = getattr(arvados.keep.KeepClient, 'put')
+ @mock.patch('arvados.keep.KeepClient.put', autospec=True)
+ def test_storage_classes_desired(self, put_mock):
+ put_mock.side_effect = self.keep_put
+ c = Collection(storage_classes_desired=['default'])
+ with c.open("file.txt", 'wb') as f:
+ f.write('content')
+ c.save_new()
+ _, kwargs = put_mock.call_args
+ self.assertEqual(['default'], kwargs['classes'])
+
@mock.patch('arvados.keep.KeepClient.put', autospec=True)
def test_repacked_block_submission_get_permission_token(self, mocked_put):
'''