misc corrections / ruff lint and format

Author: d-w-moore

README.md

@@ -2118,6 +2118,34 @@ membership, this can be achieved with another query.
 `<session>.permissions` was therefore removed in v2.0.0
 in favor of `<session>.acls`.
 
+Atomic ACLs
+-----------
+
+A list of permissions may be added to an object atomically using
+the AccessManager's apply_atomic_operations method:
+```
+from irods.access import ACLOperation
+from irods.helpers import home_collection
+session = irods.helpers.make_session()
+myCollection = session.collections.create(f"{home_collection(session).path}/newCollection")
+
+session.acls.apply_atomic_operations(myCollection.path,
+   *[ACLOperation("read", "public"),
+     ACLOperation("write", "bob", "otherZone")
+    ])
+```
+ACLOperation objects form a linear order with iRODSAccess objects, and
+indeed are subclassed from them as well, allowing equivalency testing:
+
+Thus, for example:
+```
+ACLOperation('read','public') in sess.acls.get(object)
+```
+is a valid operation. Consequently, any client application that habitually
+caches object permissions could use similar code to check new ACLOperations against the cache
+and conceivably be able to optimize size of an atomic ACLs request by eliminating
+any ACLOperations that might have been redundant.
+
 Quotas (v2.0.0)
 ---------------
 

irods/access.py

@@ -5,6 +5,22 @@
 from irods.path import iRODSPath
 
 
+_ichmod_listed_permissions = (
+    "own",
+    "delete_object",
+    "write",
+    "modify_object",
+    "create_object",
+    "delete_metadata",
+    "modify_metadata",
+    "create_metadata",
+    "read",
+    "read_object",
+    "read_metadata",
+    "null",
+)
+
+
 class _Access_LookupMeta(type):
     def __getitem__(self, key):
         return self.codes[key]
@@ -28,6 +44,7 @@ def to_int(cls, key):
     def to_string(cls, key):
         return cls.strings[key]
 
+    # noqa: RUF012 - Cannot change in minor release
     codes = collections.OrderedDict(
         (key_, value_)
         for key_, value_ in sorted(
@@ -55,24 +72,10 @@ def to_string(cls, key):
             ).items(),
             key=lambda _: _[1],
         )
-        if key_
-        in (
-            # These are copied from ichmod help text.
-            "own",
-            "delete_object",
-            "write",
-            "modify_object",
-            "create_object",
-            "delete_metadata",
-            "modify_metadata",
-            "create_metadata",
-            "read",
-            "read_object",
-            "read_metadata",
-            "null",
-        )
+        if key_ in _ichmod_listed_permissions
     )
 
+    # noqa: RUF012 - Cannot change in minor release
     strings = collections.OrderedDict((number, string) for string, number in codes.items())
 
     def __init__(self, access_name, path, user_name="", user_zone="", user_type=None):
@@ -91,6 +94,14 @@ def __init__(self, access_name, path, user_name="", user_zone="", user_type=None
         self.user_zone = user_zone
         self.user_type = user_type
 
+    def __lt__(self, other):
+        return (self.access_name, self.user_name, self.user_zone, iRODSPath(self.path)) < (
+            other.access_name,
+            other.user_name,
+            other.user_zone,
+            iRODSPath(other.path),
+        )
+
     def __eq__(self, other):
         return (
             self.access_name == other.access_name
@@ -102,8 +113,9 @@ def __eq__(self, other):
     def __hash__(self):
         return hash((self.access_name, iRODSPath(self.path), self.user_name, self.user_zone))
 
-    def copy(self, decanonicalize=False, ref_zone=''):
+    def copy(self, decanonicalize=False, implied_zone=''):
         other = copy.deepcopy(self)
+
         if decanonicalize:
             replacement_string = {
                 "read object": "read",
@@ -112,8 +124,10 @@ def copy(self, decanonicalize=False, ref_zone=''):
                 "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 = ''
+
+        # Useful if we wish to force an explicitly specified local zone to an implicit zone spec in the copy, for equality testing:
+        if '' != implied_zone == other.user_zone:
+            other.user_zone = ''
 
         return other
 
@@ -124,6 +138,56 @@ def __repr__(self):
         return f"<iRODSAccess {access_name} {self.path} {self.user_name}{user_type_hint} {self.user_zone}>"
 
 
+class ACLOperation(iRODSAccess):
+    def __init__(self, access_name: str, user_name: str = "", user_zone: str = ""):
+        super().__init__(
+            access_name=access_name,
+            path="",
+            user_name=user_name,
+            user_zone=user_zone,
+        )
+
+    def __eq__(self, other):
+        return (
+            self.access_name,
+            self.user_name,
+            self.user_zone,
+        ) == (
+            other.access_name,
+            other.user_name,
+            other.user_zone,
+        )
+
+    def __lt__(self, other):
+        return (
+            self.access_name,
+            self.user_name,
+            self.user_zone,
+        ) < (
+            other.access_name,
+            other.user_name,
+            other.user_zone,
+        )
+
+    def __repr__(self):
+        return f"<ACLOperation {self.access_name} {self.user_name} {self.user_zone}>"
+
+
+(
+    _ichmod_synonym_mapping := {
+        # syn : canonical
+        "write": "modify_object",
+        "read": "read_object",
+    }
+).update((key.replace("_", " "), key) for key in iRODSAccess.codes.keys())
+
+
+all_permissions = {
+    **iRODSAccess.codes,
+    **{key: iRODSAccess.codes[_ichmod_synonym_mapping[key]] for key in _ichmod_synonym_mapping},
+}
+
+
 class _iRODSAccess_pre_4_3_0(iRODSAccess):
     codes = collections.OrderedDict(
         (key.replace("_", " "), value)

irods/api_number.py

@@ -177,6 +177,7 @@
     "ATOMIC_APPLY_METADATA_OPERATIONS_APN": 20002,
     "GET_FILE_DESCRIPTOR_INFO_APN": 20000,
     "REPLICA_CLOSE_APN": 20004,
+    "ATOMIC_APPLY_ACL_OPERATIONS_APN": 20005,
     "TOUCH_APN": 20007,
     "AUTH_PLUG_REQ_AN": 1201,
     "AUTHENTICATION_APN": 110000,

irods/manager/access_manager.py

@@ -14,7 +14,6 @@
     CollectionAccess,
 )
 from irods.access import iRODSAccess
-import irods.exception as ex
 from irods.column import In
 from irods.user import iRODSUser
 
@@ -37,27 +36,26 @@ def users_by_ids(session, ids=()):
 
 
 class AccessManager(Manager):
-
-    def _ACL_operation(self, op_input: iRODSAccess):
+    @staticmethod
+    def _to_acl_operation_json(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}
-            )
+            **({} 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]
+    def apply_atomic_operations(self, logical_path: str, *operations, admin=False):
+        request_text = {
+            "logical_path": logical_path,
+            "admin_mode": admin,
+            "operations": [self._to_acl_operation_json(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,
+                int_info=api_number["ATOMIC_APPLY_ACL_OPERATIONS_APN"],
             )
             conn.send(request_msg)
             response = conn.recv()

irods/test/access_test.py

@@ -4,7 +4,7 @@
 import sys
 import unittest
 
-from irods.access import iRODSAccess
+from irods.access import iRODSAccess, ACLOperation
 from irods.collection import iRODSCollection
 from irods.column import In, Like
 from irods.exception import UserDoesNotExist
@@ -498,43 +498,45 @@ def test_iRODSAccess_cannot_be_constructed_using_unsupported_type__issue_558(sel
         )
 
     def test_atomic_acls_505(self):
-        #import pdb;pdb.set_trace()
         ses = self.sess
-        zone = user1 = user2 = group = None
+        zone = user1 = user2 = user3 = group = None
         try:
-            zone = ses.zones.create("twilight","remote")
+            zone = ses.zones.create("twilight", "remote")
             user1 = ses.users.create("test_user_505", "rodsuser")
             user2 = ses.users.create("rod_serling_505#twilight", "rodsuser")
+            user3 = ses.users.create("local_test_user_505", "rodsuser")
             group = ses.groups.create("test_group_505")
-            ses.acls._call_atomic_acl_api(
+            ses.acls.apply_atomic_operations(
                 self.coll_path,
-                a1:=iRODSAccess("write", "",  user1.name,  user1.zone),
-                a2:=iRODSAccess("read", "", user2.name, user2.zone),
-                a3:=iRODSAccess("read", "", group.name),
+                a1:=ACLOperation("write", user1.name, user1.zone),
+                a2:=ACLOperation("read", user2.name, user2.zone),
+                a3:=ACLOperation("read", user3.name),
+                a4:=ACLOperation("read", group.name),
             )
 
-            accesses = ses.acls.get(self.coll)
+            normalize = lambda access: access.copy(decanonicalize=True, implied_zone=ses.zone)
 
-            # 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
+            accesses = [normalize(acl) for acl in ses.acls.get(self.coll)]
 
             # 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)
-            )
+            self.assertIn(normalize(a1), accesses)
+            self.assertIn(normalize(a2), accesses)
+            self.assertIn(normalize(a3), accesses)
+            self.assertIn(normalize(a4), accesses)
+
         finally:
             if user1:
                 user1.remove()
             if user2:
                 user2.remove()
+            if user3:
+                user3.remove()
             if group:
                 group.remove()
             if zone:
                 zone.remove()
 
+
 if __name__ == "__main__":
     # let the tests find the parent irods lib
     sys.path.insert(0, os.path.abspath("../.."))