[#561] experimental implementation: user_has_access

irods/manager/collection_manager.py

@@ -82,9 +82,10 @@ def unregister(self, path, **options):
         self.remove(path, **options)
 
 
-    def exists(self, path):
+    def exists(self, path, return_object = ()):
         try:
-            self.get(path)
+            obj = self.get(path)
+            if return_object == []: return_object.append(obj)
         except CollectionDoesNotExist:
             return False
         return True

irods/manager/data_object_manager.py

@@ -597,9 +597,10 @@ def unregister(self, path, **options):
             self.unlink(path, **options)
 
 
-    def exists(self, path):
+    def exists(self, path, return_object = ()):
         try:
-            self.get(path)
+            obj = self.get(path)
+            if return_object == []: return_object.append(obj)
         except ex.DoesNotExist:
             return False
         return True

irods/session.py

@@ -22,7 +22,7 @@
 from irods.manager.resource_manager import ResourceManager
 from irods.manager.zone_manager import ZoneManager
 from irods.message import (iRODSMessage, STR_PI)
-from irods.exception import (NetworkException, NotImplementedInIRODSServer)
+from irods.exception import (NetworkException, NotImplementedInIRODSServer, DoesNotExist)
 from irods.password_obfuscation import decode
 from irods import NATIVE_AUTH_SCHEME, PAM_AUTH_SCHEMES
 from . import DEFAULT_CONNECTION_TIMEOUT
@@ -446,3 +446,40 @@ def get_connection_refresh_time(self, **kwargs):
                 connection_refresh_time = -1
 
         return connection_refresh_time
+
+
+    def user_has_access(self, collection_or_data_path, user_name, access_name, zone = ''):
+
+        from irods.test.access_test import get_name_mapping
+        from irods.access import iRODSAccess
+
+        zone = (zone or self.zone)
+
+        mapping = get_name_mapping(self)
+        true_access_name = mapping[access_name]
+        access_type_int = iRODSAccess.to_int(true_access_name)
+
+        exist = []
+
+        if not self.collections.exists(collection_or_data_path, return_object = exist):
+            self.data_objects.exists(collection_or_data_path, return_object = exist)
+
+        if not exist:
+            raise DoesNotExist
+
+        access_rights = {}
+        user_is_group = self.users.get(user_name).user_type == 'rodsgroup'
+
+        for acl in self.acls.get(exist[0]):
+            if iRODSAccess.to_int(acl.access_name) >= access_type_int:
+                if acl.user_name == user_name and (acl.user_zone == zone or user_is_group):
+                    access_rights[user_name] = acl.access_name
+                    break
+                elif acl.user_type == "rodsgroup":
+                    group = self.groups.get(acl.user_name)
+                    matching_group_members = [u for u in self.groups.get(acl.user_name).members if u.name == user_name and u.zone == zone]
+                    if matching_group_members:
+                        access_rights[group.name] = acl.access_name
+                        break
+
+        return len(access_rights) > 0

irods/test/access_test.py

@@ -16,6 +16,14 @@
 import irods.test.helpers as helpers
 
 
+def get_name_mappings(session):
+        VERSION_DEPENDENT_STRINGS = { 'MODIFY':'modify_object', 'READ':'read_object' } if self.sess.server_version >= (4,3) \
+                               else { 'MODIFY':'modify object', 'READ':'read object' }
+        return dict( [(i,i) for i in ( 'own', VERSION_DEPENDENT_STRINGS['MODIFY'], VERSION_DEPENDENT_STRINGS['READ'])] +
+                     [('write',VERSION_DEPENDENT_STRINGS['MODIFY']), ('read', VERSION_DEPENDENT_STRINGS['READ'])] )
+
+
+
 class TestAccess(unittest.TestCase):
 
     def setUp(self):
@@ -24,10 +32,8 @@ def setUp(self):
         # Create test collection
         self.coll_path = '/{}/home/{}/test_dir'.format(self.sess.zone, self.sess.username)
         self.coll = helpers.make_collection(self.sess, self.coll_path)
-        VERSION_DEPENDENT_STRINGS = { 'MODIFY':'modify_object', 'READ':'read_object' } if self.sess.server_version >= (4,3) \
-                               else { 'MODIFY':'modify object', 'READ':'read object' }
-        self.mapping = dict( [(i,i) for i in ( 'own', VERSION_DEPENDENT_STRINGS['MODIFY'], VERSION_DEPENDENT_STRINGS['READ'])] +
-                             [('write',VERSION_DEPENDENT_STRINGS['MODIFY']), ('read', VERSION_DEPENDENT_STRINGS['READ'])] )
+
+        self.mapping = get_name_mappings(self.sess)
 
     def tearDown(self):
         '''Remove test data and close connections