Skip to content

Commit

Permalink
Linting (only)
Browse files Browse the repository at this point in the history
  • Loading branch information
@claravox
claravox committed Oct 13, 2023
1 parent bfb15bf commit 7eb4922
Show file tree
Hide file tree
Showing 2 changed files with 213 additions and 183 deletions.
268 changes: 132 additions & 136 deletions yclienttools/common_rules.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
from yclienttools import common_queries
from yclienttools.exceptions import SizeNotSupportedException


class RuleInterface:

def __init__(self, session, yoda_version):
Expand All @@ -22,12 +23,12 @@ def __init__(self, session, yoda_version):
"""
self.session = session
self.set_re = False if yoda_version == "1.7" else True
self.uuGroupAdd_version = "1.7" if yoda_version in ["1.7", "1.8"] else "1.9"
self.uuGroupAdd_version = "1.7" if yoda_version in [
"1.7", "1.8"] else "1.9"
self.default_rule_engine = 'irods_rule_engine_plugin-irods_rule_language-instance'


def call_rule(self, rulename, params, number_outputs,
rule_engine = None):
rule_engine=None):
"""Run a rule
:param rulename: name of the rule
Expand All @@ -40,23 +41,25 @@ def call_rule(self, rulename, params, number_outputs,
for input_var in params.keys():
body += "*{},".format(input_var)

outparams = list(map(lambda n : '*outparam{}'.format(str(n+1)), range(number_outputs)))
outparams = list(
map(lambda n: '*outparam{}'.format(str(n + 1)), range(number_outputs)))
body += '{}); writeLine("stdout","{}")}}'.format(
",".join(outparams),
"\n".join(outparams))

input_params = { "*{}".format(k) : '"{}"'.format(v) for (k,v) in params.items() }
input_params = {"*{}".format(k): '"{}"'.format(v)
for (k, v) in params.items()}
output_params = 'ruleExecOut'

if self.set_re:
re_config = { 'instance_name': self.default_rule_engine if rule_engine is None
else rule_engine }
re_config = {'instance_name': self.default_rule_engine if rule_engine is None
else rule_engine}
else:
re_config = {}

myrule = Rule(
self.session,
rule_file = StringIO(body),
rule_file=StringIO(body),
params=input_params,
output=output_params,
**re_config)
Expand All @@ -67,163 +70,156 @@ def call_rule(self, rulename, params, number_outputs,

return buf[:number_outputs]


def _string_list_to_list(self, s):
if s.startswith("[") and s.endswith("]"):
return s[1:-1].split(",")
else:
raise ValueError("Unable to convert string representation of list to list")

raise ValueError(
"Unable to convert string representation of list to list")

def call_uuGroupGetMembers(self, groupname):
"""Returns list of group members"""
parms = OrderedDict([
( 'groupname', groupname)] )
[out] = self.call_rule('uuGroupGetMembers', parms, 1)
if len(out) >= 1023 and not out.endswith("]"):
raise SizeNotSupportedException("Group member list exceeds 1023 bytes")
return self._string_list_to_list(out)

"""Returns list of group members"""
parms = OrderedDict([
('groupname', groupname)])
[out] = self.call_rule('uuGroupGetMembers', parms, 1)
if len(out) >= 1023 and not out.endswith("]"):
raise SizeNotSupportedException(
"Group member list exceeds 1023 bytes")
return self._string_list_to_list(out)

def call_uuGroupUserRemove(self, groupname, user):
"""Removes a user from a group"""
parms = OrderedDict([
( 'groupname', groupname),
( 'user', user) ])
return self.call_rule('uuGroupUserRemove', parms, 2)

"""Removes a user from a group"""
parms = OrderedDict([
('groupname', groupname),
('user', user)])
return self.call_rule('uuGroupUserRemove', parms, 2)

def call_uuGroupGetMemberType(self, groupname, user):
""":returns: member type of a group member"""
parms = OrderedDict([
( 'groupname', groupname),
( 'user', user) ])
return self.call_rule('uuGroupGetMemberType', parms, 1)[0]

def call_uuGroupUserAddByOtherCreator(self, groupname, username, creator_user, creator_zone):
"""Adds user to group on the behalf of a creator user.
:param: groupname
:param: username
:param: creator_user
:param: creator_zone
:returns: (status, message) ; status !=0 is error
"""
parms = OrderedDict([
('groupname', groupname),
('username', username),
('creatorUser', creator_user),
('creatorZone', creator_zone)])
return self.call_rule('uuGroupUserAdd', parms, 2)
""":returns: member type of a group member"""
parms = OrderedDict([
('groupname', groupname),
('user', user)])
return self.call_rule('uuGroupGetMemberType', parms, 1)[0]

def call_uuGroupUserAddByOtherCreator(
self, groupname, username, creator_user, creator_zone):
"""Adds user to group on the behalf of a creator user.
:param: groupname
:param: username
:param: creator_user
:param: creator_zone
:returns: (status, message) ; status !=0 is error
"""
parms = OrderedDict([
('groupname', groupname),
('username', username),
('creatorUser', creator_user),
('creatorZone', creator_zone)])
return self.call_rule('uuGroupUserAdd', parms, 2)

def call_uuGroupUserAdd(self, groupname, username):
"""Adds user to group.
:param: groupname
:param: username
:returns: (status, message) ; status !=0 is error
"""
parms = OrderedDict([
('groupname', groupname),
('username', username)])
return self.call_rule('uuGroupUserAdd', parms, 2)
"""Adds user to group.
:param: groupname
:param: username
:returns: (status, message) ; status !=0 is error
"""
parms = OrderedDict([
('groupname', groupname),
('username', username)])
return self.call_rule('uuGroupUserAdd', parms, 2)

def call_uuGroupUserChangeRole(self, groupname, username, newrole):
"""Change role of user in group
:param groupname: name of group
:param username: name of user
:param newrole: new role (can be "manager", "reader", "normal")
:returns: (status, message) ; status != 0 is error
"""
parms = OrderedDict([
('groupname', groupname),
('username', username),
('newrole', newrole)])
return self.call_rule('uuGroupUserChangeRole', parms, 2)
"""Change role of user in group
:param groupname: name of group
:param username: name of user
:param newrole: new role (can be "manager", "reader", "normal")
:returns: (status, message) ; status != 0 is error
"""
parms = OrderedDict([
('groupname', groupname),
('username', username),
('newrole', newrole)])
return self.call_rule('uuGroupUserChangeRole', parms, 2)

def call_uuGroupExists(self, groupname):
"""Check whether group name exists on Yoda
:param groupname: name of group
:returns: false/true
"""
parms = OrderedDict([('groupname', groupname)])
[out] = self.call_rule('uuGroupExists', parms, 1)
return out == 'true'
"""Check whether group name exists on Yoda
:param groupname: name of group
:returns: false/true
"""
parms = OrderedDict([('groupname', groupname)])
[out] = self.call_rule('uuGroupExists', parms, 1)
return out == 'true'

def call_uuUserExists(self, username):
"""Check whether user name exists on Yoda
:param username: name of user
:returns: false/true
"""
parms = OrderedDict([('username', username)])
[out] = self.call_rule('uuUserExists', parms, 1)
return out == 'true'
"""Check whether user name exists on Yoda
:param username: name of user
:returns: false/true
"""
parms = OrderedDict([('username', username)])
[out] = self.call_rule('uuUserExists', parms, 1)
return out == 'true'

def call_uuGroupAdd(self, groupname, category,
subcategory, description, classification, schema_id='default-2', expiration_date=''):
"""Adds a group
:param groupname: name of group
:param category: category / community
:param subcategory: subcategory
:param description: description
:param classification: security classification
:param schema_id: schema id
:param expiration_date: expiration date
:returns: (status, message). Status not 0 means error,
-1089000 means group name already exists
"""
if self.uuGroupAdd_version == "1.7":
parms = OrderedDict([
('groupname', groupname),
('category', category),
('subcategory', subcategory),
('description', description),
('classification', classification)])
elif self.uuGroupAdd_version == "1.9":
parms = OrderedDict([
('groupname', groupname),
('category', category),
('subcategory', subcategory),
('schema_id', schema_id if schema_id not in ("", ".") else "default-2"),
('expiration_date', expiration_date),
('description', description),
('dataClassification', classification),
('co_identifier', '')
])

return self.call_rule('uuGroupAdd', parms, 2)

subcategory, description, classification, schema_id='default-2', expiration_date=''):
"""Adds a group
:param groupname: name of group
:param category: category / community
:param subcategory: subcategory
:param description: description
:param classification: security classification
:param schema_id: schema id
:param expiration_date: expiration date
:returns: (status, message). Status not 0 means error,
-1089000 means group name already exists
"""
if self.uuGroupAdd_version == "1.7":
parms = OrderedDict([
('groupname', groupname),
('category', category),
('subcategory', subcategory),
('description', description),
('classification', classification)])
elif self.uuGroupAdd_version == "1.9":
parms = OrderedDict([
('groupname', groupname),
('category', category),
('subcategory', subcategory),
('schema_id', schema_id if schema_id not in ("", ".") else "default-2"),
('expiration_date', expiration_date),
('description', description),
('dataClassification', classification),
('co_identifier', '')
])

return self.call_rule('uuGroupAdd', parms, 2)

def call_uuGroupModify(self, groupname, property, value):
"""Modifies one property of a group
:param groupname: name of group
:param property: property to change
:param value: value to change the property to
"""Modifies one property of a group
:returns: (status, message). Status not 0 means error.
"""
parms = OrderedDict([('groupname', groupname),
('property', property),
('value', value)])
return self.call_rule('uuGroupModify', parms, 2)
:param groupname: name of group
:param property: property to change
:param value: value to change the property to
:returns: (status, message). Status not 0 means error.
"""
parms = OrderedDict([('groupname', groupname),
('property', property),
('value', value)])
return self.call_rule('uuGroupModify', parms, 2)

def call_uuGroupRemove(self, groupname):
"""Removes an empty group
"""Removes an empty group
:param groupname: name of group
:param groupname: name of group
:returns: (status, message). Status not 0 means error.
"""
parms = OrderedDict([('groupname', groupname)])
return self.call_rule('uuGroupRemove', parms, 2)
:returns: (status, message). Status not 0 means error.
"""
parms = OrderedDict([('groupname', groupname)])
return self.call_rule('uuGroupRemove', parms, 2)
Loading

0 comments on commit 7eb4922

Please sign in to comment.