Merge branch '19954-permission-dedup-doc'

[arvados.git] / services / fuse / tests / test_mount.py

diff --git a/services/fuse/tests/test_mount.py b/services/fuse/tests/test_mount.py
index df3d4263417bcc271b77c05dc75aec0ee8343aea..f4e5138e2ce0fd5d1559046754f2c50a4f1c2ddb 100644 (file)
--- a/services/fuse/tests/test_mount.py
+++ b/services/fuse/tests/test_mount.py
@@ -6,7 +6,9 @@ from __future__ import absolute_import
 from future.utils import viewitems
 from builtins import str
 from builtins import object
+from pathlib import Path
 from six import assertRegex
+import errno
 import json
 import llfuse
 import logging
@@ -20,6 +22,7 @@ import parameterized
 
 import arvados
 import arvados_fuse as fuse
+from arvados_fuse import fusedir
 from . import run_test_server
 
 from .integration_test import IntegrationTest
@@ -1225,7 +1228,10 @@ class SlashSubstitutionTest(IntegrationTest):
 
     def setUp(self):
         super(SlashSubstitutionTest, self).setUp()
-        self.api = arvados.safeapi.ThreadSafeApiCache(arvados.config.settings())
+        self.api = arvados.safeapi.ThreadSafeApiCache(
+            arvados.config.settings(),
+            version='v1',
+        )
         self.api.config = lambda: {"Collections": {"ForwardSlashNameSubstitution": "[SLASH]"}}
         self.testcoll = self.api.collections().create(body={"name": "foo/bar/baz"}).execute()
         self.testcolleasy = self.api.collections().create(body={"name": "foo-bar-baz"}).execute()
@@ -1284,7 +1290,10 @@ class StorageClassesTest(IntegrationTest):
 
     def setUp(self):
         super(StorageClassesTest, self).setUp()
-        self.api = arvados.safeapi.ThreadSafeApiCache(arvados.config.settings())
+        self.api = arvados.safeapi.ThreadSafeApiCache(
+            arvados.config.settings(),
+            version='v1',
+        )
 
     @IntegrationTest.mount(argv=mnt_args)
     def test_collection_default_storage_classes(self):
@@ -1321,7 +1330,110 @@ class ReadonlyCollectionTest(MountTestBase):
     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)
+        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)