[#505] create an atomic ACLs endpoint

irods/access.py

@@ -102,7 +102,7 @@
     def __hash__(self):
         return hash((self.access_name, iRODSPath(self.path), self.user_name, self.user_zone))
 
-    def copy(self, decanonicalize=False):
+    def copy(self, decanonicalize=False, ref_zone=''):
         other = copy.deepcopy(self)
         if decanonicalize:
             replacement_string = {
@@ -112,6 +112,9 @@
                 "modify_object": "write",
             }.get(self.access_name)
             other.access_name = replacement_string if replacement_string is not None else self.access_name
+            if '' != ref_zone == other.user_zone:
+                other.user_zone = ''
+
         return other
 
     def __repr__(self):

irods/exception.py

@@ -650,6 +650,10 @@
     code = -130000
 
 
+class SYS_INTERNAL_ERR(SystemException):
+    code = -154000
+
+
 class SYS_BAD_INPUT(iRODSException):
     code = -158000
 

irods/manager/access_manager.py

@@ -1,19 +1,20 @@
 from os.path import basename, dirname
 
 from irods.manager import Manager
 from irods.api_number import api_number
-from irods.message import ModAclRequest, iRODSMessage
+from irods.message import ModAclRequest, iRODSMessage, JSON_Message
 from irods.data_object import iRODSDataObject, irods_dirname, irods_basename
 from irods.collection import iRODSCollection
 from irods.models import (
     DataObject,
     Collection,
     User,
     CollectionUser,
     DataAccess,
     CollectionAccess,
 )
 from irods.access import iRODSAccess
+import irods.exception as ex
 from irods.column import In
 from irods.user import iRODSUser
 
@@ -36,6 +37,33 @@
 
 
 class AccessManager(Manager):
+
+    def _ACL_operation(self, op_input: iRODSAccess):
+        return {
+            "acl": op_input.access_name,
+            "entity_name": op_input.user_name,
+            **(
+                {} if not (z := op_input.user_zone)
+                else {"zone": z}
+            )
+        }
+
+    def _call_atomic_acl_api(self, logical_path : str, *operations, admin=False):
+        request_text = {"logical_path": logical_path}
+        request_text["admin_mode"] = admin
+        request_text["operations"] = [self._ACL_operation(op) for op in operations]
+
+        with self.sess.pool.get_connection() as conn:
+            request_msg = iRODSMessage(
+                "RODS_API_REQ",
+                JSON_Message(request_text, conn.server_version),
+                int_info=20005,
+            )
+            conn.send(request_msg)
+            response = conn.recv()
+        response_msg = response.get_json_encoded_struct()
+        logger.debug("in atomic ACL api, server responded with: %r", response_msg)
+
     def get(self, target, report_raw_acls=True, **kw):
 
         if report_raw_acls:

irods/test/access_test.py

@@ -497,6 +497,42 @@
             self.sess,
         )
 
+    def test_atomic_acls_505(self):
+        ses = self.sess
+        zone = user1 = user2 = group = None
+        try:
+            zone = ses.zones.create("twilight","remote")
+            user1 = ses.users.create("test_user_505", "rodsuser")
+            user2 = ses.users.create("rod_serling_505#twilight", "rodsuser")
+            group = ses.groups.create("test_group_505")
+            ses.acls._call_atomic_acl_api(
+                self.coll_path,
+                a1:=iRODSAccess("write", "",  user1.name,  user1.zone),
+                a2:=iRODSAccess("read", "", user2.name, user2.zone),
+                a3:=iRODSAccess("read", "", group.name),
+            )
+
+            accesses = ses.acls.get(self.coll)
+
+            # For purposes of equality tests, assign the path name of interest into each ACL.
+            for p in (a1, a2, a3):
+                p.path = self.coll_path
+
+            # Assert that the ACLs we added are among those listed for the object in the catalog.
+            normalize = lambda access: access.copy(decanonicalize=True, ref_zone=ses.zone)
+            self.assertLess(
+                set(normalize(_) for _ in (a1,a2,a3)),
+                set(normalize(_) for _ in accesses)
+            )
+        finally:
+            if user1:
+                user1.remove()
+            if user2:
+                user2.remove()
+            if group:
+                group.remove()
+            if zone:
+                zone.remove()
 
 if __name__ == "__main__":
     # let the tests find the parent irods lib