+
+class StorageClassesTest(IntegrationTest):
+ mnt_args = [
+ '--read-write',
+ '--mount-home', 'homedir',
+ ]
+
+ def setUp(self):
+ super(StorageClassesTest, self).setUp()
+ self.api = arvados.safeapi.ThreadSafeApiCache(
+ arvados.config.settings(),
+ version='v1',
+ )
+
+ @IntegrationTest.mount(argv=mnt_args)
+ def test_collection_default_storage_classes(self):
+ coll_path = os.path.join(self.mnt, 'homedir', 'a_collection')
+ self.api.collections().create(body={'name':'a_collection'}).execute()
+ self.pool_test(coll_path)
+ @staticmethod
+ def _test_collection_default_storage_classes(self, coll):
+ self.assertEqual(storage_classes_desired(coll), ['default'])
+
+ @IntegrationTest.mount(argv=mnt_args+['--storage-classes', 'foo'])
+ def test_collection_custom_storage_classes(self):
+ coll_path = os.path.join(self.mnt, 'homedir', 'new_coll')
+ os.mkdir(coll_path)
+ self.pool_test(coll_path)
+ @staticmethod
+ def _test_collection_custom_storage_classes(self, coll):
+ self.assertEqual(storage_classes_desired(coll), ['foo'])
+
+def _readonlyCollectionTestHelper(mounttmp):
+ f = open(os.path.join(mounttmp, 'thing1.txt'), 'rt')
+ # Testing that close() doesn't raise an error.
+ f.close()
+
+class ReadonlyCollectionTest(MountTestBase):
+ def setUp(self):
+ super(ReadonlyCollectionTest, self).setUp()
+ cw = arvados.collection.Collection()
+ with cw.open('thing1.txt', 'wt') as f:
+ f.write("data 1")
+ cw.save_new(owner_uuid=run_test_server.fixture("groups")["aproject"]["uuid"])
+ self.testcollection = cw.api_response()
+
+ def runTest(self):
+ settings = arvados.config.settings().copy()
+ settings["ARVADOS_API_TOKEN"] = run_test_server.fixture("api_client_authorizations")["project_viewer"]["api_token"]
+ self.api = arvados.safeapi.ThreadSafeApiCache(settings, version='v1')
+ self.make_mount(fuse.CollectionDirectory, collection_record=self.testcollection, enable_write=False)
+
+ self.pool.apply(_readonlyCollectionTestHelper, (self.mounttmp,))
+
+
+@parameterized.parameterized_class([
+ {'root_class': fusedir.ProjectDirectory, 'root_kwargs': {
+ 'project_object': run_test_server.fixture('users')['admin'],
+ }},
+ {'root_class': fusedir.ProjectDirectory, 'root_kwargs': {
+ 'project_object': run_test_server.fixture('groups')['public'],
+ }},
+])
+class UnsupportedCreateTest(MountTestBase):
+ root_class = None
+ root_kwargs = {}
+
+ def setUp(self):
+ super().setUp()
+ if 'prefs' in self.root_kwargs.get('project_object', ()):
+ self.root_kwargs['project_object']['prefs'] = {}
+ self.make_mount(self.root_class, **self.root_kwargs)
+ # Make sure the directory knows about its top-level ents.
+ os.listdir(self.mounttmp)
+
+ def test_create(self):
+ test_path = Path(self.mounttmp, 'test_create')
+ with self.assertRaises(OSError) as exc_check:
+ with test_path.open('w'):
+ pass
+ self.assertEqual(exc_check.exception.errno, errno.ENOTSUP)
+
+
+# FIXME: IMO, for consistency with the "create inside a project" case,
+# these operations should also return ENOTSUP instead of EPERM.
+# Right now they're returning EPERM because the clasess' writable() method
+# usually returns False, and the Operations class transforms that accordingly.
+# However, for cases where the mount will never be writable, I think ENOTSUP
+# is a clearer error: it lets the user know they can't fix the problem by
+# adding permissions in Arvados, etc.
+@parameterized.parameterized_class([
+ {'root_class': fusedir.MagicDirectory,
+ 'preset_dir': 'by_id',
+ 'preset_file': 'README',
+ },
+
+ {'root_class': fusedir.SharedDirectory,
+ 'root_kwargs': {
+ 'exclude': run_test_server.fixture('users')['admin']['uuid'],
+ },
+ 'preset_dir': 'Active User',
+ },
+
+ {'root_class': fusedir.TagDirectory,
+ 'root_kwargs': {
+ 'tag': run_test_server.fixture('links')['foo_collection_tag']['name'],
+ },
+ 'preset_dir': run_test_server.fixture('collections')['foo_collection_in_aproject']['uuid'],
+ },
+
+ {'root_class': fusedir.TagsDirectory,
+ 'preset_dir': run_test_server.fixture('links')['foo_collection_tag']['name'],
+ },
+])
+class UnsupportedOperationsTest(UnsupportedCreateTest):
+ preset_dir = None
+ preset_file = None
+
+ def test_create(self):
+ test_path = Path(self.mounttmp, 'test_create')
+ with self.assertRaises(OSError) as exc_check:
+ with test_path.open('w'):
+ pass
+ self.assertEqual(exc_check.exception.errno, errno.EPERM)
+
+ def test_mkdir(self):
+ test_path = Path(self.mounttmp, 'test_mkdir')
+ with self.assertRaises(OSError) as exc_check:
+ test_path.mkdir()
+ self.assertEqual(exc_check.exception.errno, errno.EPERM)
+
+ def test_rename(self):
+ src_name = self.preset_dir or self.preset_file
+ if src_name is None:
+ return
+ test_src = Path(self.mounttmp, src_name)
+ test_dst = test_src.with_name('test_dst')
+ with self.assertRaises(OSError) as exc_check:
+ test_src.rename(test_dst)
+ self.assertEqual(exc_check.exception.errno, errno.EPERM)
+
+ def test_rmdir(self):
+ if self.preset_dir is None:
+ return
+ test_path = Path(self.mounttmp, self.preset_dir)
+ with self.assertRaises(OSError) as exc_check:
+ test_path.rmdir()
+ self.assertEqual(exc_check.exception.errno, errno.EPERM)
+
+ def test_unlink(self):
+ if self.preset_file is None:
+ return
+ test_path = Path(self.mounttmp, self.preset_file)
+ with self.assertRaises(OSError) as exc_check:
+ test_path.unlink()
+ self.assertEqual(exc_check.exception.errno, errno.EPERM)