From f93a93c053219caa4a2273534381dae6bd373a6f Mon Sep 17 00:00:00 2001 From: d-w-moore Date: Sun, 25 Oct 2026 13:14:54 -0400 Subject: [PATCH 1/7] user equality operator (for group membership test) --- irods/user.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/irods/user.py b/irods/user.py index 4312eef64..d0a048fca 100644 --- a/irods/user.py +++ b/irods/user.py @@ -17,6 +17,10 @@ def remove_quota(self, resource="total"): def set_quota(self, amount, resource="total"): self.manager.set_quota(self.name, amount, resource=resource) + # TODO: add issue for iRODSUser & iRODSGroup equality operators + def __eq__(self, other_user): + return (self.name, self.zone) == (other_user.name, other_user.zone) + def __init__(self, manager, result=None): self.manager = manager if result: @@ -82,6 +86,9 @@ def remove_quota(self, resource="total"): def set_quota(self, amount, resource="total"): self.manager.set_quota(self.name, amount, resource=resource) + def __eq__(self, other_group): + return self.name == other_group.name + def __init__(self, manager, result=None): self.manager = manager if result: From a295d7ab5e81e679074bcdca64068aa27d2cf4db Mon Sep 17 00:00:00 2001 From: d-w-moore Date: Thu, 22 Oct 2026 12:09:26 -0400 Subject: [PATCH 2/7] [_561] implement get_maximum_permissions_for_path functionality and test A query is done to determine the highest permission level the given user (or group) has on the target collection/data_object. A specific query will be used if the calling user has the availability, otherwise a reasonably efficient general query is used instead. --- irods/access.py | 4 + irods/manager/access_manager.py | 167 +++++++++++++++++++++++++++++++- irods/query.py | 48 ++++++++- irods/test/access_test.py | 51 ++++++++++ 4 files changed, 267 insertions(+), 3 deletions(-) diff --git a/irods/access.py b/irods/access.py index 1ad6660a7..682b1ee10 100644 --- a/irods/access.py +++ b/irods/access.py @@ -1,5 +1,6 @@ import collections import copy +import enum import warnings from irods.collection import iRODSCollection @@ -20,6 +21,9 @@ "null", ) +class return_type(enum.Enum): + INTEGER = 1 + STRING = 2 class _Access_LookupMeta(type): @staticmethod diff --git a/irods/manager/access_manager.py b/irods/manager/access_manager.py index 5873c1e15..640a08348 100644 --- a/irods/manager/access_manager.py +++ b/irods/manager/access_manager.py @@ -3,11 +3,13 @@ import logging from os.path import basename, dirname -from irods.access import iRODSAccess +import irods.access +from irods.access import iRODSAccess, return_type from irods.api_number import api_number from irods.collection import iRODSCollection from irods.column import In from irods.data_object import irods_basename, irods_dirname, iRODSDataObject +from irods.exception import CAT_NO_ROWS_FOUND from irods.manager import Manager from irods.message import JSON_Message, ModAclRequest, iRODSMessage from irods.models import ( @@ -16,9 +18,11 @@ CollectionUser, DataAccess, DataObject, + Group, User, ) -from irods.user import iRODSUser +from irods.query import AutoAliasedSpecificQuery, SpecificQuery +from irods.user import iRODSUser, iRODSGroup logger = logging.getLogger(__name__) @@ -34,8 +38,167 @@ def users_by_ids(session, ids=()): cond = () if not ids else ((In(User.id, list(map(int, ids))),) if len(ids) > 1 else (User.id == int(ids[0]),)) return [iRODSUser(session.users, i) for i in session.query(User.id, User.name, User.type, User.zone).filter(*cond)] +class UserAccessTypeQuery(AutoAliasedSpecificQuery): + SQL=''' +WITH target_user AS ( + SELECT user_id + FROM r_user_main + WHERE user_name = ? + AND zone_name = ? +), +resolved_object AS ( + -- The path names a collection + SELECT coll_id AS object_id + FROM r_coll_main + WHERE coll_name = ? + + UNION + + -- The path names a data object: collection path + '/' + data name + SELECT d.data_id AS object_id + FROM r_data_main AS d + JOIN r_coll_main AS c ON c.coll_id = d.coll_id + WHERE CONCAT(c.coll_name, '/', d.data_name) = ? +), +grants AS ( + -- Direct grants + SELECT a.access_type_id + FROM r_objt_access AS a + JOIN target_user AS u ON u.user_id = a.user_id + JOIN resolved_object AS o ON o.object_id = a.object_id + + UNION ALL + + -- Grants to groups the user belongs to + SELECT a.access_type_id + FROM r_user_group AS ug + JOIN target_user AS u ON u.user_id = ug.user_id + JOIN r_objt_access AS a ON a.user_id = ug.group_user_id + JOIN resolved_object AS o ON o.object_id = a.object_id +) +SELECT access_type_id +FROM grants +ORDER BY access_type_id DESC +LIMIT 1; +''' + +class QueryParameterError(ValueError): pass class AccessManager(Manager): + + def get_maximum_permission_for_path(self, + user, + coll=None, + data=None, + force_genquery=False, + feedback=None, + return_type = irods.access.return_type.INTEGER + ): + ses = self.sess + specific_query = None + + # Mainly for testing to assure success in guiding the query method to be 'general' or 'specific' + def feedback_method(string): + if isinstance(feedback,list): feedback[:0]=[string] + + # Convert user or group to object. Need to know if we are dealing with a rodsadmin + if isinstance(user, (iRODSUser, iRODSGroup)): + pass + else: + msg = "'user' argument should be of type iRODSUser or iRODSGroup." + raise QueryParameterError(msg) + + denullify = lambda s: s if s else "x" + + if not force_genquery: + specific_query_args = [ + user.name, + getattr(user,'zone',ses.zone), + denullify(coll.path if isinstance(coll, iRODSCollection) else coll), + denullify(data.path if isinstance(data, iRODSDataObject) else data), + ] + + # Find out if the query is already registered. + try: + specific_query = SpecificQuery(ses, alias='findQueryByAlias', args=('getUserAccessType',)) + specific_query.execute() + except CAT_NO_ROWS_FOUND: + specific_query = None + else: + specific_query = SpecificQuery(ses, alias='getUserAccessType', args=specific_query_args) + + # If there is no canned query available, and we are a rodsadmin, then install our own: + if not specific_query and ses.users.get(ses.username).type == 'rodsadmin': + specific_query = UserAccessTypeQuery(ses, args=specific_query_args) + + # Dispatch to the available/chosen query method. + if specific_query: + feedback_method('specific') + max_permission = self._get_maximum_permission_for_path_via_SpecificQuery(specific_query) + res = [max_permission] + else: + feedback_method('general') + res = None + groups = {g[Group.id] for g in ses.query(Group, User).filter(Group.id != user.id, User.id == user.id)} + groups |= {user.id} + + if data: + if isinstance(data, str): + collN = irods_dirname(data) + dataN = irods_basename(data) + join_criteria = [DataObject.name == dataN, Collection.name == collN] + elif isinstance(data, iRODSDataObject): + join_criteria = [ DataAccess.data_id == data.id] + else: + msg="data must be full logical path or data object" + raise ValueError(msg) + + q = ses.query(DataAccess.type).filter( + In(DataAccess.user_id, groups), + *join_criteria + ).order_by(DataAccess.type,order='desc') + + res = [_[DataAccess.type] for _ in q] + + elif coll: + if isinstance(coll, str): + join_criteria = [Collection.name == coll] + elif isinstance(coll, iRODSCollection): + join_criteria = [CollectionAccess.access_id == coll.id] # TODO: access_id is not an accurate name, should be coll_id + else: + msg="data must be full logical path or collection object" + raise ValueError(msg) + + q=ses.query(CollectionAccess).filter( + In(CollectionAccess.user_id, groups), + *join_criteria + ).order_by(CollectionAccess.type,order='desc') + + res = [_[CollectionAccess.type] for _ in q] + + else: + msg = 'Need a collection or data object' + raise QueryParameterError(msg) + + retval_transform = ( + (lambda _: iRODSAccess.strings[_]) if return_type == irods.access.return_type.STRING else (lambda _:_) + ) + res = res[0] if res else None + return retval_transform(res) if isinstance(res, int) else None + + + @staticmethod + def _get_maximum_permission_for_path_via_SpecificQuery(spec_query): + try: + row = next(iter( + spec_query.execute() + )) + except StopIteration: + row = {} + + highest_perm = row.get(0) + return None if highest_perm is None else int(highest_perm) + @staticmethod def _to_acl_operation_json(op_input: iRODSAccess): return { diff --git a/irods/query.py b/irods/query.py index aa6e7a514..121761cbf 100644 --- a/irods/query.py +++ b/irods/query.py @@ -1,4 +1,6 @@ from collections import OrderedDict +import copy +import uuid from irods import MAX_SQL_ROWS from irods.models import Model @@ -313,7 +315,10 @@ def remove(self): response = conn.recv() return response - def execute(self, limit=MAX_SQL_ROWS, offset=0, options=0, conditions=None): + def execute(self, limit=MAX_SQL_ROWS, offset=0, options=0, conditions=None, instance=None): + if instance is not None: + self = instance + target = self._alias or self._sql if conditions is None: @@ -363,6 +368,47 @@ def get_results(self): yield result +class AutoAliasedSpecificQuery(SpecificQuery): + # Subclasses should set this to a nonzero length string in valid SQL. + SQL = None + + def __init__(self, sess, **kw): + self.__alias = None + + if 'sql' not in kw: + kw['sql'] = getattr(self,'SQL') or 'select 1' + + if kw.get('alias') is None: + kw['alias'] = self.__alias = str(uuid.uuid1()) + + super().__init__(sess,**kw) + + if self.__alias: + self.register() + + def _clone(self): + q = copy.copy(self) + + # So that the _clone's destruction does not remove the query from the catalog: + q.__alias = None + return q + + def __del__(self): + if self.__alias is not None: + SpecificQuery(self.session, alias=self.__alias).remove() + + def execute(self, args=(), **kw): + # Support independent iterators. + q = self._clone() + if not q._args: + q._args = list(args) + results = super().execute(instance=q, **kw) + return results + + @property + def auto_alias(self): return self.__alias + + # Record a copy of the original value of IRODS_QUERY_LIMIT for possible future access. import irods.helpers as helpers diff --git a/irods/test/access_test.py b/irods/test/access_test.py index 142bb2f5c..2a9131f05 100644 --- a/irods/test/access_test.py +++ b/irods/test/access_test.py @@ -532,6 +532,57 @@ def test_atomic_acls__issue_505(self): if zone: zone.remove() + def test_get_maximum_permission_for_path__issue_561(self): + user = group = coll = None + ses = self.sess + try: + coll = ses.collections.create('/tempZone/home/public/folder') + data = ses.data_objects.create('/tempZone/home/public/folder/file') + user = ses.users.create('bob','rodsuser') + group = ses.groups.create('lab') + + # Set up test scenario with a new user and group, the group having deliberately higher permission + # so as to be able to account for the effect of group membership in the query results. + USER_PERM = 'modify_metadata' + GROUP_PERM = 'modify_object' + test_perms = { user.name: USER_PERM, group.name: GROUP_PERM } + for obj in (coll, data): + ses.acls.set(iRODSAccess(USER_PERM, obj.path, user.name)) + ses.acls.set(iRODSAccess(GROUP_PERM, obj.path, group.name)) + + for is_member in (False, True): + if is_member: + group.addmember(user.name,user.zone) + for usr in (user,group): + maxusr = user if user not in group.members else group + expected_permission = test_perms[maxusr.name] + for obj_type, obj in ( + ("coll", coll), + ("coll", coll.path), + ("data", data), + ("data", data.path), + ): + for query_type in ('general', 'specific'): + with self.subTest(f"{is_member=}, testing {usr} {obj} {query_type}"): + force_genquery_option = {'general': True, 'specific': False}[query_type] + feedback = [] + permission = ses.acls.get_maximum_permission_for_path( + usr, + feedback=feedback, + force_genquery=force_genquery_option, + **{obj_type: obj}, + ) + self.assertEqual(perm, expected_permission) + self.assertEqual(feedback[0], query_type) + finally: + if coll: + # Removes test data object as well. + coll.remove(force=True, recurse=True) + if user: + user.remove() + if group: + group.remove() + if __name__ == "__main__": # let the tests find the parent irods lib From 66676cac0abadad981e5cf25e75431eaf991ebb3 Mon Sep 17 00:00:00 2001 From: d-w-moore Date: Mon, 26 Oct 2026 11:43:01 -0400 Subject: [PATCH 3/7] [_561] test is now passing, yay --- irods/manager/access_manager.py | 4 ++-- irods/test/access_test.py | 14 ++++++++------ irods/user.py | 2 +- 3 files changed, 11 insertions(+), 9 deletions(-) diff --git a/irods/manager/access_manager.py b/irods/manager/access_manager.py index 640a08348..6c153533a 100644 --- a/irods/manager/access_manager.py +++ b/irods/manager/access_manager.py @@ -150,7 +150,7 @@ def feedback_method(string): elif isinstance(data, iRODSDataObject): join_criteria = [ DataAccess.data_id == data.id] else: - msg="data must be full logical path or data object" + msg="'data' parameter must be full logical path or data object" raise ValueError(msg) q = ses.query(DataAccess.type).filter( @@ -166,7 +166,7 @@ def feedback_method(string): elif isinstance(coll, iRODSCollection): join_criteria = [CollectionAccess.access_id == coll.id] # TODO: access_id is not an accurate name, should be coll_id else: - msg="data must be full logical path or collection object" + msg="'coll' parameter must be full logical path or collection object" raise ValueError(msg) q=ses.query(CollectionAccess).filter( diff --git a/irods/test/access_test.py b/irods/test/access_test.py index 2a9131f05..9b918fc3f 100644 --- a/irods/test/access_test.py +++ b/irods/test/access_test.py @@ -4,6 +4,7 @@ import sys import unittest +import irods.access from irods.access import ACLOperation, iRODSAccess from irods.collection import iRODSCollection from irods.column import In, Like @@ -536,8 +537,8 @@ def test_get_maximum_permission_for_path__issue_561(self): user = group = coll = None ses = self.sess try: - coll = ses.collections.create('/tempZone/home/public/folder') - data = ses.data_objects.create('/tempZone/home/public/folder/file') + coll = ses.collections.create('/tempZone/home/public/folder_561') + data = ses.data_objects.create('/tempZone/home/public/folder_561/file') user = ses.users.create('bob','rodsuser') group = ses.groups.create('lab') @@ -553,9 +554,9 @@ def test_get_maximum_permission_for_path__issue_561(self): for is_member in (False, True): if is_member: group.addmember(user.name,user.zone) - for usr in (user,group): - maxusr = user if user not in group.members else group - expected_permission = test_perms[maxusr.name] + for usr in (user, group): + max_user = usr if usr not in group.members else group + expected_permission = test_perms[max_user.name] for obj_type, obj in ( ("coll", coll), ("coll", coll.path), @@ -570,9 +571,10 @@ def test_get_maximum_permission_for_path__issue_561(self): usr, feedback=feedback, force_genquery=force_genquery_option, + return_type=irods.access.return_type.STRING, **{obj_type: obj}, ) - self.assertEqual(perm, expected_permission) + self.assertEqual(permission, expected_permission) self.assertEqual(feedback[0], query_type) finally: if coll: diff --git a/irods/user.py b/irods/user.py index d0a048fca..484eb6407 100644 --- a/irods/user.py +++ b/irods/user.py @@ -19,7 +19,7 @@ def set_quota(self, amount, resource="total"): # TODO: add issue for iRODSUser & iRODSGroup equality operators def __eq__(self, other_user): - return (self.name, self.zone) == (other_user.name, other_user.zone) + return isinstance(other_user, iRODSUser) and (self.name, self.zone) == (other_user.name, other_user.zone) def __init__(self, manager, result=None): self.manager = manager From 36bba51c21cd2fcc4dad48a3d6b379f55ecca1fc Mon Sep 17 00:00:00 2001 From: d-w-moore Date: Mon, 26 Oct 2026 13:08:39 -0400 Subject: [PATCH 4/7] whitespace --- irods/manager/access_manager.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/irods/manager/access_manager.py b/irods/manager/access_manager.py index 6c153533a..9635851d1 100644 --- a/irods/manager/access_manager.py +++ b/irods/manager/access_manager.py @@ -87,10 +87,10 @@ class QueryParameterError(ValueError): pass class AccessManager(Manager): def get_maximum_permission_for_path(self, - user, - coll=None, - data=None, - force_genquery=False, + user, + coll=None, + data=None, + force_genquery=False, feedback=None, return_type = irods.access.return_type.INTEGER ): @@ -100,7 +100,7 @@ def get_maximum_permission_for_path(self, # Mainly for testing to assure success in guiding the query method to be 'general' or 'specific' def feedback_method(string): if isinstance(feedback,list): feedback[:0]=[string] - + # Convert user or group to object. Need to know if we are dealing with a rodsadmin if isinstance(user, (iRODSUser, iRODSGroup)): pass From 7744d93e7da0762a740e793f5eacd83f35d04eac Mon Sep 17 00:00:00 2001 From: d-w-moore Date: Mon, 26 Oct 2026 13:11:28 -0400 Subject: [PATCH 5/7] appropriate(?) name for registered specific query --- irods/manager/access_manager.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/irods/manager/access_manager.py b/irods/manager/access_manager.py index 9635851d1..289de3c9b 100644 --- a/irods/manager/access_manager.py +++ b/irods/manager/access_manager.py @@ -120,7 +120,7 @@ def feedback_method(string): # Find out if the query is already registered. try: - specific_query = SpecificQuery(ses, alias='findQueryByAlias', args=('getUserAccessType',)) + specific_query = SpecificQuery(ses, alias='findQueryByAlias', args=('getMaximumPermissionForPath',)) specific_query.execute() except CAT_NO_ROWS_FOUND: specific_query = None From 91cf0b11ff38c80c3eff19cf9c91a258859f63f5 Mon Sep 17 00:00:00 2001 From: d-w-moore Date: Thu, 29 Oct 2026 13:27:50 -0400 Subject: [PATCH 6/7] review comments --- irods/access.py | 4 - irods/manager/access_manager.py | 135 ++++++++++++-------------------- irods/query.py | 44 ++--------- irods/test/access_test.py | 69 ++++++++++++++-- 4 files changed, 120 insertions(+), 132 deletions(-) diff --git a/irods/access.py b/irods/access.py index 682b1ee10..ffd9d28bb 100644 --- a/irods/access.py +++ b/irods/access.py @@ -21,10 +21,6 @@ "null", ) -class return_type(enum.Enum): - INTEGER = 1 - STRING = 2 - class _Access_LookupMeta(type): @staticmethod def _codes(): diff --git a/irods/manager/access_manager.py b/irods/manager/access_manager.py index 289de3c9b..d596ba809 100644 --- a/irods/manager/access_manager.py +++ b/irods/manager/access_manager.py @@ -3,8 +3,7 @@ import logging from os.path import basename, dirname -import irods.access -from irods.access import iRODSAccess, return_type +from irods.access import iRODSAccess from irods.api_number import api_number from irods.collection import iRODSCollection from irods.column import In @@ -21,9 +20,10 @@ Group, User, ) -from irods.query import AutoAliasedSpecificQuery, SpecificQuery +from irods.query import SpecificQuery from irods.user import iRODSUser, iRODSGroup + logger = logging.getLogger(__name__) @@ -38,68 +38,40 @@ def users_by_ids(session, ids=()): cond = () if not ids else ((In(User.id, list(map(int, ids))),) if len(ids) > 1 else (User.id == int(ids[0]),)) return [iRODSUser(session.users, i) for i in session.query(User.id, User.name, User.type, User.zone).filter(*cond)] -class UserAccessTypeQuery(AutoAliasedSpecificQuery): - SQL=''' -WITH target_user AS ( - SELECT user_id - FROM r_user_main - WHERE user_name = ? - AND zone_name = ? -), -resolved_object AS ( - -- The path names a collection - SELECT coll_id AS object_id - FROM r_coll_main - WHERE coll_name = ? - - UNION - - -- The path names a data object: collection path + '/' + data name - SELECT d.data_id AS object_id - FROM r_data_main AS d - JOIN r_coll_main AS c ON c.coll_id = d.coll_id - WHERE CONCAT(c.coll_name, '/', d.data_name) = ? -), -grants AS ( - -- Direct grants - SELECT a.access_type_id - FROM r_objt_access AS a - JOIN target_user AS u ON u.user_id = a.user_id - JOIN resolved_object AS o ON o.object_id = a.object_id - - UNION ALL - - -- Grants to groups the user belongs to - SELECT a.access_type_id - FROM r_user_group AS ug - JOIN target_user AS u ON u.user_id = ug.user_id - JOIN r_objt_access AS a ON a.user_id = ug.group_user_id - JOIN resolved_object AS o ON o.object_id = a.object_id -) -SELECT access_type_id -FROM grants -ORDER BY access_type_id DESC -LIMIT 1; -''' -class QueryParameterError(ValueError): pass +class QueryParameterError(ValueError): + pass -class AccessManager(Manager): +# This is the alias where we expect to find a specific query +alias_for_get_maximum_permission_for_path = 'get_maximum_permission_for_path' +class AccessManager(Manager): def get_maximum_permission_for_path(self, - user, - coll=None, - data=None, - force_genquery=False, - feedback=None, - return_type = irods.access.return_type.INTEGER + user : [iRODSUser|iRODSGroup], + coll : [str|iRODSCollection|None]=None, + data : [str|iRODSDataObject|None]=None, + specific_query : [SpecificQuery|str|None]='', + return_type : type[int]|type[str]=int, ): - ses = self.sess - specific_query = None + """ + Return the highest permission the named user/group has with respect to the + given data object or collection. In calculating this result, all ACLs and + group memberships are factored in. + + Only one of 'data' or 'coll' may be specified. - # Mainly for testing to assure success in guiding the query method to be 'general' or 'specific' - def feedback_method(string): - if isinstance(feedback,list): feedback[:0]=[string] + Args: + user: an iRODSUser or iRODSGroup object for whom the permission is to be calculated. + coll: an iRODS logical path or iRODSCollection. + data: an iRODS logical path or iRODSDataObject. + specific_query (default ""): a SpecificQuery object or an alias specifying + the specific query used for the calculation. The default value of "" will be + interpreted to mean the designated specific query preconfigured for this express + purpose. If the server does not cache a query under that alias, or if None is + given as a value, then the calculation will be done via a general query. + return_type: the type expected for the return value, i.e. str or int. Defaults to int. + """ + ses = self.sess # Convert user or group to object. Need to know if we are dealing with a rodsadmin if isinstance(user, (iRODSUser, iRODSGroup)): @@ -108,36 +80,31 @@ def feedback_method(string): msg = "'user' argument should be of type iRODSUser or iRODSGroup." raise QueryParameterError(msg) - denullify = lambda s: s if s else "x" + _specific_query = None - if not force_genquery: - specific_query_args = [ - user.name, - getattr(user,'zone',ses.zone), - denullify(coll.path if isinstance(coll, iRODSCollection) else coll), - denullify(data.path if isinstance(data, iRODSDataObject) else data), - ] + if specific_query == '': + specific_query = alias_for_get_maximum_permission_for_path - # Find out if the query is already registered. - try: - specific_query = SpecificQuery(ses, alias='findQueryByAlias', args=('getMaximumPermissionForPath',)) - specific_query.execute() - except CAT_NO_ROWS_FOUND: - specific_query = None - else: - specific_query = SpecificQuery(ses, alias='getUserAccessType', args=specific_query_args) - - # If there is no canned query available, and we are a rodsadmin, then install our own: - if not specific_query and ses.users.get(ses.username).type == 'rodsadmin': - specific_query = UserAccessTypeQuery(ses, args=specific_query_args) + if isinstance(specific_query, SpecificQuery): + _specific_query = specific_query + elif isinstance(specific_query, str): + _specific_query = SpecificQuery(ses, alias=specific_query) # Dispatch to the available/chosen query method. - if specific_query: - feedback_method('specific') - max_permission = self._get_maximum_permission_for_path_via_SpecificQuery(specific_query) + if _specific_query is not None: + def denullify(s): + return s if s else "x" + _specific_query.reset_for_new_query( + args=[ + user.name, + getattr(user,'zone',ses.zone), + denullify(coll.path if isinstance(coll, iRODSCollection) else coll), + denullify(data.path if isinstance(data, iRODSDataObject) else data), + ] + ) + max_permission = self._get_maximum_permission_for_path_via_SpecificQuery(_specific_query) res = [max_permission] else: - feedback_method('general') res = None groups = {g[Group.id] for g in ses.query(Group, User).filter(Group.id != user.id, User.id == user.id)} groups |= {user.id} @@ -181,7 +148,7 @@ def feedback_method(string): raise QueryParameterError(msg) retval_transform = ( - (lambda _: iRODSAccess.strings[_]) if return_type == irods.access.return_type.STRING else (lambda _:_) + (lambda _: iRODSAccess.strings[_]) if return_type is str else (lambda _:_) ) res = res[0] if res else None return retval_transform(res) if isinstance(res, int) else None diff --git a/irods/query.py b/irods/query.py index 121761cbf..7e0b764be 100644 --- a/irods/query.py +++ b/irods/query.py @@ -292,6 +292,11 @@ def __init__(self, sess, sql=None, alias=None, columns=None, args=None): self._columns = columns self._args = args or [] + def reset_for_new_query(self, args=None): + self._continue_index = 0 + if args is not None: + self._args = args + def register(self): if not self._sql: raise ValueError("Empty query") @@ -368,45 +373,6 @@ def get_results(self): yield result -class AutoAliasedSpecificQuery(SpecificQuery): - # Subclasses should set this to a nonzero length string in valid SQL. - SQL = None - - def __init__(self, sess, **kw): - self.__alias = None - - if 'sql' not in kw: - kw['sql'] = getattr(self,'SQL') or 'select 1' - - if kw.get('alias') is None: - kw['alias'] = self.__alias = str(uuid.uuid1()) - - super().__init__(sess,**kw) - - if self.__alias: - self.register() - - def _clone(self): - q = copy.copy(self) - - # So that the _clone's destruction does not remove the query from the catalog: - q.__alias = None - return q - - def __del__(self): - if self.__alias is not None: - SpecificQuery(self.session, alias=self.__alias).remove() - - def execute(self, args=(), **kw): - # Support independent iterators. - q = self._clone() - if not q._args: - q._args = list(args) - results = super().execute(instance=q, **kw) - return results - - @property - def auto_alias(self): return self.__alias # Record a copy of the original value of IRODS_QUERY_LIMIT for possible future access. diff --git a/irods/test/access_test.py b/irods/test/access_test.py index 9b918fc3f..86bd649b0 100644 --- a/irods/test/access_test.py +++ b/irods/test/access_test.py @@ -1,7 +1,8 @@ #! /usr/bin/env python - + import os import sys +import textwrap import unittest import irods.access @@ -15,6 +16,50 @@ from irods.test import helpers from irods.user import iRODSUser +from irods.manager.access_manager import alias_for_get_maximum_permission_for_path + +SQL = textwrap.dedent(''' + WITH target_user AS ( + SELECT user_id + FROM r_user_main + WHERE user_name = ? + AND zone_name = ? + ), + resolved_object AS ( + -- The path names a collection + SELECT coll_id AS object_id + FROM r_coll_main + WHERE coll_name = ? + + UNION + + -- The path names a data object: collection path + '/' + data name + SELECT d.data_id AS object_id + FROM r_data_main AS d + JOIN r_coll_main AS c ON c.coll_id = d.coll_id + WHERE CONCAT(c.coll_name, '/', d.data_name) = ? + ), + grants AS ( + -- Direct grants + SELECT a.access_type_id + FROM r_objt_access AS a + JOIN target_user AS u ON u.user_id = a.user_id + JOIN resolved_object AS o ON o.object_id = a.object_id + + UNION ALL + + -- Grants to groups the user belongs to + SELECT a.access_type_id + FROM r_user_group AS ug + JOIN target_user AS u ON u.user_id = ug.user_id + JOIN r_objt_access AS a ON a.user_id = ug.group_user_id + JOIN resolved_object AS o ON o.object_id = a.object_id + ) + SELECT access_type_id + FROM grants + ORDER BY access_type_id DESC + LIMIT 1;''' +) class TestAccess(unittest.TestCase): def setUp(self): @@ -551,6 +596,19 @@ def test_get_maximum_permission_for_path__issue_561(self): ses.acls.set(iRODSAccess(USER_PERM, obj.path, user.name)) ses.acls.set(iRODSAccess(GROUP_PERM, obj.path, group.name)) + options = {} + try: + SpecificQuery(ses, alias='findQueryByAlias', args=(alias_for_get_maximum_permission_for_path,)).execute() + except CAT_NO_ROWS_FOUND: + options = dict( + query_alias = str(uuid.uuid1()), + sql = SQL + ) + + specific_query = SpecificQuery(ses, alias=query_alias, **options) + +# def test_get_maximum_permission_for_path__issue_561(self): + for is_member in (False, True): if is_member: group.addmember(user.name,user.zone) @@ -563,15 +621,14 @@ def test_get_maximum_permission_for_path__issue_561(self): ("data", data), ("data", data.path), ): - for query_type in ('general', 'specific'): - with self.subTest(f"{is_member=}, testing {usr} {obj} {query_type}"): + for s_query in (None, specific_query): + with self.subTest(f"{is_member=} testing {usr=} {obj=} {s_query=}"): force_genquery_option = {'general': True, 'specific': False}[query_type] feedback = [] permission = ses.acls.get_maximum_permission_for_path( usr, - feedback=feedback, - force_genquery=force_genquery_option, return_type=irods.access.return_type.STRING, + specific_query=s_query, **{obj_type: obj}, ) self.assertEqual(permission, expected_permission) @@ -584,6 +641,8 @@ def test_get_maximum_permission_for_path__issue_561(self): user.remove() if group: group.remove() + if query_alias: + specific_query.remove() if __name__ == "__main__": From 0cc80dfd016fbf9241f16babe70817dc13d5b255 Mon Sep 17 00:00:00 2001 From: d-w-moore Date: Thu, 29 Oct 2026 13:45:22 -0400 Subject: [PATCH 7/7] reflect return type for typing --- irods/manager/access_manager.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/irods/manager/access_manager.py b/irods/manager/access_manager.py index d596ba809..12d9422f1 100644 --- a/irods/manager/access_manager.py +++ b/irods/manager/access_manager.py @@ -52,7 +52,7 @@ def get_maximum_permission_for_path(self, data : [str|iRODSDataObject|None]=None, specific_query : [SpecificQuery|str|None]='', return_type : type[int]|type[str]=int, - ): + ) -> int|str: """ Return the highest permission the named user/group has with respect to the given data object or collection. In calculating this result, all ACLs and