+# Copyright (C) The Arvados Authors. All rights reserved.
+#
+# SPDX-License-Identifier: AGPL-3.0
+
+from builtins import range
+from six import assertRegex
import arvados
import arvados_fuse
import arvados_fuse.command
def current_manifest(tmpdir):
- return json.load(open(
- os.path.join(tmpdir, '.arvados#collection')
- ))['manifest_text']
+ with open(os.path.join(tmpdir, '.arvados#collection')) as tmp:
+ return json.load(tmp)['manifest_text']
+def storage_classes_desired(tmpdir):
+ with open(os.path.join(tmpdir, '.arvados#collection')) as tmp:
+ return json.load(tmp)['storage_classes_desired']
class TmpCollectionTest(IntegrationTest):
mnt_args = [
'--mount-tmp', 'zzz',
]
+ @IntegrationTest.mount(argv=mnt_args+['--storage-classes', 'foo, bar'])
+ def test_storage_classes(self):
+ self.pool_test(os.path.join(self.mnt, 'zzz'))
+ @staticmethod
+ def _test_storage_classes(self, zzz):
+ self.assertEqual(storage_classes_desired(zzz), ['foo', 'bar'])
+
@IntegrationTest.mount(argv=mnt_args+['--mount-tmp', 'yyy'])
def test_two_tmp(self):
self.pool_test(os.path.join(self.mnt, 'zzz'),
def _test_tmp_onefile(self, tmpdir):
with open(os.path.join(tmpdir, 'foo'), 'w') as f:
f.write('foo')
- self.assertRegexpMatches(
+ assertRegex(
+ self,
current_manifest(tmpdir),
r'^\. acbd18db4cc2f85cedef654fccc4a4d8\+3(\+\S+)? 0:3:foo\n$')
r'^\. 37b51d194a7513e45b56f6524f2d51f2\+3(\+\S+)? acbd18db4cc2f85cedef654fccc4a4d8\+3(\+\S+)? 0:3:bar 3:3:foo\n$'),
('foo', None,
r'^\. 37b51d194a7513e45b56f6524f2d51f2\+3(\+\S+)? 0:3:bar\n$'),
+ ('bar', None,
+ r'^$'),
]
- for fn, content, expect in ops:
- path = os.path.join(tmpdir, fn)
- if content is None:
- os.unlink(path)
- else:
- with open(path, 'w') as f:
- f.write(content)
- self.assertRegexpMatches(current_manifest(tmpdir), expect)
+ for _ in range(10):
+ for fn, content, expect in ops:
+ path = os.path.join(tmpdir, fn)
+ if content is None:
+ os.unlink(path)
+ else:
+ with open(path, 'w') as f:
+ f.write(content)
+ assertRegex(self, current_manifest(tmpdir), expect)
+
+ @IntegrationTest.mount(argv=mnt_args)
+ def test_tmp_rewrite(self):
+ self.pool_test(os.path.join(self.mnt, 'zzz'))
+ @staticmethod
+ def _test_tmp_rewrite(self, tmpdir):
+ with open(os.path.join(tmpdir, "b1"), 'w') as f:
+ f.write("b1")
+ with open(os.path.join(tmpdir, "b2"), 'w') as f:
+ f.write("b2")
+ with open(os.path.join(tmpdir, "b1"), 'w') as f:
+ f.write("1b")
+ assertRegex(self, current_manifest(tmpdir), "^\. ed4f3f67c70b02b29c50ce1ea26666bd\+4(\+\S+)? 0:2:b1 2:2:b2\n$")