# Getter and Setter for sub_meta_rule
+ def get_aggregation_algorithm_dict(self, intra_extension_id):
+ with sql.transaction() as session:
+ query = session.query(AggregationAlgorithm)
+ query = query.filter_by(intra_extension_id=intra_extension_id)
+ ref = query.first()
+ return {ref.id: ref.aggregation_algorithm}
+
def set_aggregation_algorithm_dict(self, intra_extension_id, aggregation_algorithm_id, aggregation_algorithm_dict):
with sql.transaction() as session:
query = session.query(AggregationAlgorithm)
session.flush()
return self.get_aggregation_algorithm_dict(intra_extension_id)
- def get_aggregation_algorithm_dict(self, intra_extension_id):
- with sql.transaction() as session:
- query = session.query(AggregationAlgorithm)
- query = query.filter_by(intra_extension_id=intra_extension_id)
- ref = query.first()
- return {ref.id: ref.aggregation_algorithm}
-
- def del_aggregation_algorithm(self, intra_extension_id, aggregation_algorithm_id):
- with sql.transaction() as session:
- query = session.query(AggregationAlgorithm)
- query = query.filter_by(intra_extension_id=intra_extension_id, id=aggregation_algorithm_id)
- ref = query.first()
- session.delete(ref)
+ # def del_aggregation_algorithm(self, intra_extension_id, aggregation_algorithm_id):
+ # with sql.transaction() as session:
+ # query = session.query(AggregationAlgorithm)
+ # query = query.filter_by(intra_extension_id=intra_extension_id, id=aggregation_algorithm_id)
+ # ref = query.first()
+ # session.delete(ref)
# Getter and Setter for sub_meta_rule
subject_category_dict = dict()
subject_category_dict['name'] = kw.get('subject_category_name', None)
subject_category_dict['description'] = kw.get('subject_category_description', None)
- return self.admin_api.add_subject_category(user_id, intra_extension_id, subject_category_dict)
+ return self.admin_api.add_subject_category_dict(user_id, intra_extension_id, subject_category_dict)
@controller.protected()
def get_subject_category(self, context, **kw):
subject_category_dict = dict()
subject_category_dict['name'] = kw.get('subject_category_name', None)
subject_category_dict['description'] = kw.get('subject_category_description', None)
- return self.admin_api.set_subject_category(user_id, intra_extension_id, subject_category_id, subject_category_dict)
+ return self.admin_api.set_subject_category_dict(user_id, intra_extension_id, subject_category_id, subject_category_dict)
@controller.protected()
def get_object_categories(self, context, **kw):
object_category_dict = dict()
object_category_dict['name'] = kw.get('object_category_name', None)
object_category_dict['description'] = kw.get('object_category_description', None)
- return self.admin_api.add_object_category(user_id, intra_extension_id, object_category_dict)
+ return self.admin_api.add_object_category_dict(user_id, intra_extension_id, object_category_dict)
@controller.protected()
def get_object_category(self, context, **kw):
object_category_dict = dict()
object_category_dict['name'] = kw.get('object_category_name', None)
object_category_dict['description'] = kw.get('object_category_description', None)
- return self.admin_api.set_object_category(user_id, intra_extension_id, object_category_id, object_category_dict)
+ return self.admin_api.set_object_category_dict(user_id, intra_extension_id, object_category_id, object_category_dict)
@controller.protected()
def get_action_categories(self, context, **kw):
action_category_dict = dict()
action_category_dict['name'] = kw.get('action_category_name', None)
action_category_dict['description'] = kw.get('action_category_description', None)
- return self.admin_api.add_action_category(user_id, intra_extension_id, action_category_dict)
+ return self.admin_api.add_action_category_dict(user_id, intra_extension_id, action_category_dict)
@controller.protected()
def get_action_category(self, context, **kw):
action_category_dict = dict()
action_category_dict['name'] = kw.get('action_category_name', None)
action_category_dict['description'] = kw.get('action_category_description', None)
- return self.admin_api.set_action_category(user_id, intra_extension_id, action_category_id, action_category_dict)
+ return self.admin_api.set_action_category_dict(user_id, intra_extension_id, action_category_id, action_category_dict)
# Perimeter functions
@controller.protected()
intra_extension_id = kw.get('intra_extension_id', None)
subject_dict = dict()
subject_dict['name'] = kw.get('subject_name', None)
- subject_dict['description'] = kw.get('description', None)
+ subject_dict['description'] = kw.get('subject_description', None)
return self.admin_api.add_subject_dict(user_id, intra_extension_id, subject_dict)
@controller.protected()
user_id = self._get_user_id_from_token(context.get('token_id'))
intra_extension_id = kw.get('intra_extension_id', None)
object_id = kw.get('object_id', None)
- return self.admin_api.get_objects_dict(user_id, intra_extension_id, object_id)
+ return self.admin_api.get_object_dict(user_id, intra_extension_id, object_id)
@controller.protected()
def del_object(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
intra_extension_id = kw.get('intra_extension_id', None)
action_id = kw.get('action_id', None)
- return self.admin_api.get_actions_dict(user_id, intra_extension_id, action_id)
+ return self.admin_api.get_action_dict(user_id, intra_extension_id, action_id)
@controller.protected()
def del_action(self, context, **kw):
intra_extension_id = kw.get('intra_extension_id', None)
object_category_id = kw.get('object_category_id', None)
object_scope_id = kw.get('object_scope_id', None)
- return self.admin_api.get_object_scopes_dict(user_id, intra_extension_id, object_category_id, object_scope_id)
+ return self.admin_api.get_object_scope_dict(user_id, intra_extension_id, object_category_id, object_scope_id)
@controller.protected()
def del_object_scope(self, context, **kw):
intra_extension_id = kw.get('intra_extension_id', None)
action_category_id = kw.get('action_category_id', None)
action_scope_id = kw.get('action_scope_id', None)
- return self.admin_api.get_action_scopes_dict(user_id, intra_extension_id, action_category_id, action_scope_id)
+ return self.admin_api.get_action_scope_dict(user_id, intra_extension_id, action_category_id, action_scope_id)
@controller.protected()
def del_action_scope(self, context, **kw):
object_id = kw.get('object_id', None)
object_category_id = kw.get('object_category_id', None)
object_scope_id = kw.get('object_scope_id', None)
- return self.admin_api.add_objecty_assignment_list(user_id, intra_extension_id, object_id, object_category_id, object_scope_id)
+ return self.admin_api.add_object_assignment_list(user_id, intra_extension_id, object_id, object_category_id, object_scope_id)
@controller.protected()
def get_object_assignment(self, context, **kw):
self.admin_api.del_action_assignment(user_id, intra_extension_id, action_id, action_category_id, action_scope_id)
# Metarule functions
+
@controller.protected()
- def set_aggregation_algorithm(self, context, **kw):
+ def get_aggregation_algorithm(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
intra_extension_id = kw.get('intra_extension_id', None)
- aggregation_algorithm_id = kw.get('aggregation_algorithm_id', None)
- aggregation_algorithm_dict = kw.get('aggregation_algorithm_dict', None)
- return self.admin_api.set_aggregation_algorithm_dict(user_id, intra_extension_id, aggregation_algorithm_id, aggregation_algorithm_dict)
+ return self.admin_api.get_aggregation_algorithm_dict(user_id, intra_extension_id)
@controller.protected()
- def get_aggregation_algorithm(self, context, **kw):
+ def set_aggregation_algorithm(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
intra_extension_id = kw.get('intra_extension_id', None)
- return self.admin_api.get_aggregation_algorithms_dict(user_id, intra_extension_id)
+ aggregation_algorithm_id = kw.get('aggregation_algorithm_id', None)
+ aggregation_algorithm_dict = dict()
+ aggregation_algorithm_dict['name'] = kw.get('aggregation_algorithm_name', None)
+ aggregation_algorithm_dict['description'] = kw.get('aggregation_algorithm_description', None)
+ return self.admin_api.set_aggregation_algorithm_dict(user_id, intra_extension_id, aggregation_algorithm_id, aggregation_algorithm_dict)
@controller.protected()
def get_sub_meta_rules(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
intra_extension_id = kw.get('intra_extension_id', None)
sub_meta_rule_id = kw.get('sub_meta_rule_id', None)
- return self.admin_api.get_sub_meta_rules_dict(user_id, intra_extension_id, sub_meta_rule_id)
+ return self.admin_api.get_sub_meta_rule_dict(user_id, intra_extension_id, sub_meta_rule_id)
@controller.protected()
def del_sub_meta_rule(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
- intra_extension_id = kw.get('intra_extension_id')
+ intra_extension_id = kw.get('intra_extension_id', None)
sub_meta_rule_id = kw.get('sub_meta_rule_id', None)
self.admin_api.del_sub_meta_rule(user_id, intra_extension_id, sub_meta_rule_id)
object_category_list = kw.get('object_categories', [])
action_category_list = kw.get('action_categories', [])
rule_list = subject_category_list + action_category_list + object_category_list
- return self.admin_api.add_rule_list(user_id, intra_extension_id, sub_meta_rule_id, rule_list)
+ return self.admin_api.add_rule_dict(user_id, intra_extension_id, sub_meta_rule_id, rule_list)
@controller.protected()
def get_rule(self, context, **kw):
intra_extension_id = kw.get('intra_extension_id', None)
sub_meta_rule_id = kw.get('sub_meta_rule_id', None)
rule_id = kw.get('rule_id', None)
- return self.admin_api.get_rules_dict(user_id, intra_extension_id, sub_meta_rule_id, rule_id)
+ return self.admin_api.get_rule_dict(user_id, intra_extension_id, sub_meta_rule_id, rule_id)
@controller.protected()
def del_rule(self, context, **kw):
object_category_list = kw.get('object_categories', [])
action_category_list = kw.get('action_categories', [])
rule_list = subject_category_list + action_category_list + object_category_list
- return self.admin_api.set_rule_list(user_id, intra_extension_id, sub_meta_rule_id, rule_id, rule_list)
+ return self.admin_api.set_rule_dict(user_id, intra_extension_id, sub_meta_rule_id, rule_id, rule_list)
@dependency.requires('authz_api')
@filter_input
@enforce(("read", "write"), "subject_categories")
- def add_subject_category(self, user_id, intra_extension_id, subject_category_dict):
+ def add_subject_category_dict(self, user_id, intra_extension_id, subject_category_dict):
subject_categories_dict = self.driver.get_subject_categories_dict(intra_extension_id)
for subject_category_id in subject_categories_dict:
if subject_categories_dict[subject_category_id]['name'] == subject_category_dict['name']:
@filter_input
@enforce("read", "subject_categories")
- def get_subject_category(self, user_id, intra_extension_id, subject_category_id):
- subject_category_dict = self.driver.get_subject_categories_dict(intra_extension_id)
- if subject_category_id not in subject_category_dict:
+ def get_subject_category_dict(self, user_id, intra_extension_id, subject_category_id):
+ subject_categories_dict = self.driver.get_subject_categories_dict(intra_extension_id)
+ if subject_category_id not in subject_categories_dict:
raise SubjectCategoryUnknown()
- return subject_category_dict[subject_category_id]
+ return subject_categories_dict[subject_category_id]
@filter_input
@enforce(("read", "write"), "subject_categories")
@filter_input
@enforce(("read", "write"), "subject_categories")
- def set_subject_category(self, user_id, intra_extension_id, subject_category_id, subject_category_dict):
+ def set_subject_category_dict(self, user_id, intra_extension_id, subject_category_id, subject_category_dict):
if subject_category_id not in self.driver.get_subject_categories_dict(intra_extension_id):
raise SubjectCategoryUnknown()
return self.driver.set_subject_category_dict(intra_extension_id, subject_category_id, subject_category_dict)
@filter_input
@enforce("read", "object_categories")
- def get_object_category_dict(self, user_id, intra_extension_id):
- """
- :param user_id:
- :param intra_extension_id:
- :return:
- """
+ def get_object_categories_dict(self, user_id, intra_extension_id):
return self.driver.get_object_categories_dict(intra_extension_id)
@filter_input
@enforce(("read", "write"), "object_categories")
@enforce(("read", "write"), "object_scopes")
- def add_object_category(self, user_id, intra_extension_id, object_category_dict):
+ def add_object_category_dict(self, user_id, intra_extension_id, object_category_dict):
object_categories_dict = self.driver.get_object_categories_dict(intra_extension_id)
for object_category_id in object_categories_dict:
if object_categories_dict[object_category_id]["name"] == object_category_dict['name']:
@filter_input
@enforce("read", "object_categories")
- def get_object_category(self, user_id, intra_extension_id, object_category_id):
- object_category_dict = self.driver.get_object_categories_dict(intra_extension_id)
- if object_category_id not in object_category_dict:
+ def get_object_category_dict(self, user_id, intra_extension_id, object_category_id):
+ object_categories_dict = self.driver.get_object_categories_dict(intra_extension_id)
+ if object_category_id not in object_categories_dict:
raise ObjectCategoryUnknown()
- return object_category_dict[object_category_id]
+ return object_categories_dict[object_category_id]
@filter_input
@enforce(("read", "write"), "object_categories")
self.driver.del_object_assignment(intra_extension_id, object_id, object_category_id, assignment_id)
self.driver.del_object_category(intra_extension_id, object_category_id)
+ @filter_input
+ @enforce(("read", "write"), "object_categories")
+ def set_object_category_dict(self, user_id, intra_extension_id, object_category_id, object_category_dict):
+ if object_category_id not in self.driver.get_object_categories_dict(intra_extension_id):
+ raise ObjectCategoryUnknown()
+ return self.driver.set_object_category_dict(intra_extension_id, object_category_id, object_category_dict)
+
@filter_input
@enforce("read", "action_categories")
- def get_action_category_dict(self, user_id, intra_extension_id):
- """
- :param user_id:
- :param intra_extension_id:
- :return:
- """
+ def get_action_categories_dict(self, user_id, intra_extension_id):
return self.driver.get_action_categories_dict(intra_extension_id)
@filter_input
@enforce(("read", "write"), "action_categories")
@enforce(("read", "write"), "action_scopes")
- def add_action_category(self, user_id, intra_extension_id, action_category_dict):
+ def add_action_category_dict(self, user_id, intra_extension_id, action_category_dict):
action_categories_dict = self.driver.get_action_categories_dict(intra_extension_id)
for action_category_id in action_categories_dict:
if action_categories_dict[action_category_id]['name'] == action_category_dict['name']:
@filter_input
@enforce("read", "action_categories")
- def get_action_category(self, user_id, intra_extension_id, action_category_id):
- action_category_dict = self.driver.get_action_categories_dict(intra_extension_id)
- if action_category_id not in action_category_dict:
+ def get_action_category_dict(self, user_id, intra_extension_id, action_category_id):
+ action_categories_dict = self.driver.get_action_categories_dict(intra_extension_id)
+ if action_category_id not in action_categories_dict:
raise ActionCategoryUnknown()
- return self.driver.get_action_categories_dict(intra_extension_id)[action_category_id]
+ return action_categories_dict[action_category_id]
@filter_input
@enforce(("read", "write"), "action_categories")
self.driver.del_action_assignment(intra_extension_id, action_id, action_category_id, assignment_id)
self.driver.del_action_category(intra_extension_id, action_category_id)
+ @filter_input
+ @enforce(("read", "write"), "action_categories")
+ def set_action_category_dict(self, user_id, intra_extension_id, action_category_id, action_category_dict):
+ if action_category_id not in self.driver.get_action_categories_dict(intra_extension_id):
+ raise ActionCategoryUnknown()
+ return self.driver.set_action_category_dict(intra_extension_id, action_category_id, action_category_dict)
+
# Perimeter functions
@filter_input
@filter_input
@enforce("read", "subjects")
def get_subject_dict(self, user_id, intra_extension_id, subject_id):
- subject_dict = self.driver.get_subjects_dict(intra_extension_id)
- if subject_id in subject_dict:
+ subjects_dict = self.driver.get_subjects_dict(intra_extension_id)
+ if subject_id not in subjects_dict:
raise SubjectUnknown()
- return subject_dict[subject_id]
+ return subjects_dict[subject_id]
@filter_input
@enforce(("read", "write"), "subjects")
@filter_input
@enforce(("read", "write"), "objects")
- def add_object_dict(self, user_id, intra_extension_id, object_name):
- object_dict = self.driver.get_objects_dict(intra_extension_id)
- for object_id in object_dict:
- if object_dict[object_id]["name"] == object_name:
- raise ObjectNameExisting()
- return self.driver.set_object_dict(intra_extension_id, uuid4().hex, object_name)
-
- @filter_input
- @enforce(("read", "write"), "objects")
- def set_object_dict(self, user_id, intra_extension_id, object_id, object_dict):
+ def add_object_dict(self, user_id, intra_extension_id, object_dict):
objects_dict = self.driver.get_objects_dict(intra_extension_id)
for object_id in objects_dict:
if objects_dict[object_id]["name"] == object_dict['name']:
raise ObjectNameExisting()
- return self.driver.set_object_dict(intra_extension_id, object_id, object_dict)
+ return self.driver.set_object_dict(intra_extension_id, uuid4().hex, object_dict)
@filter_input
@enforce("read", "objects")
def get_object_dict(self, user_id, intra_extension_id, object_id):
- object_dict = self.driver.get_objects_dict(intra_extension_id)
- if object_id in object_dict:
+ objects_dict = self.driver.get_objects_dict(intra_extension_id)
+ if object_id not in objects_dict:
raise ObjectUnknown()
- return object_dict[object_id]
+ return objects_dict[object_id]
@filter_input
@enforce(("read", "write"), "objects")
def del_object(self, user_id, intra_extension_id, object_id):
- if object_id in self.driver.get_objects_dict(intra_extension_id):
+ if object_id not in self.driver.get_objects_dict(intra_extension_id):
raise ObjectUnknown()
# Destroy assignments related to this category
for object_category_id in self.driver.get_object_categories_dict(intra_extension_id):
self.driver.del_object_assignment(intra_extension_id, _object_id, object_category_id, assignment_id)
self.driver.del_object(intra_extension_id, object_id)
+ @filter_input
+ @enforce(("read", "write"), "objects")
+ def set_object_dict(self, user_id, intra_extension_id, object_id, object_dict):
+ objects_dict = self.driver.get_objects_dict(intra_extension_id)
+ for object_id in objects_dict:
+ if objects_dict[object_id]["name"] == object_dict['name']:
+ raise ObjectNameExisting()
+ return self.driver.set_object_dict(intra_extension_id, object_id, object_dict)
+
@filter_input
@enforce("read", "actions")
def get_actions_dict(self, user_id, intra_extension_id):
@filter_input
@enforce(("read", "write"), "actions")
- def add_action_dict(self, user_id, intra_extension_id, action_name):
- action_dict = self.driver.get_actions_dict(intra_extension_id)
- for action_id in action_dict:
- if action_dict[action_id]["name"] == action_name:
- raise ActionNameExisting()
- return self.driver.set_action_dict(intra_extension_id, uuid4().hex, action_name)
-
- @filter_input
- @enforce(("read", "write"), "actions")
- def set_action_dict(self, user_id, intra_extension_id, action_id, action_dict):
+ def add_action_dict(self, user_id, intra_extension_id, action_dict):
actions_dict = self.driver.get_actions_dict(intra_extension_id)
for action_id in actions_dict:
if actions_dict[action_id]["name"] == action_dict['name']:
raise ActionNameExisting()
- return self.driver.set_action_dict(intra_extension_id, action_id, action_dict)
+ return self.driver.set_action_dict(intra_extension_id, uuid4().hex, action_dict)
@filter_input
@enforce("read", "actions")
def get_action_dict(self, user_id, intra_extension_id, action_id):
- action_dict = self.driver.get_actions_dict(intra_extension_id)
- if action_id in action_dict:
+ actions_dict = self.driver.get_actions_dict(intra_extension_id)
+ if action_id not in actions_dict:
raise ActionUnknown()
- return action_dict[action_id]
+ return actions_dict[action_id]
@filter_input
@enforce(("read", "write"), "actions")
def del_action(self, user_id, intra_extension_id, action_id):
- if action_id in self.driver.get_actions_dict(intra_extension_id):
+ if action_id not in self.driver.get_actions_dict(intra_extension_id):
raise ActionUnknown()
# Destroy assignments related to this category
for action_category_id in self.driver.get_action_categories_dict(intra_extension_id):
self.driver.del_action_assignment(intra_extension_id, _action_id, action_category_id, assignment_id)
return self.driver.del_action(intra_extension_id, action_id)
+ @filter_input
+ @enforce(("read", "write"), "actions")
+ def set_action_dict(self, user_id, intra_extension_id, action_id, action_dict):
+ actions_dict = self.driver.get_actions_dict(intra_extension_id)
+ for action_id in actions_dict:
+ if actions_dict[action_id]["name"] == action_dict['name']:
+ raise ActionNameExisting()
+ return self.driver.set_action_dict(intra_extension_id, action_id, action_dict)
+
# Scope functions
@filter_input
for _subject_scope_id in subject_scopes_dict:
if subject_scope_dict['name'] == subject_scopes_dict[_subject_scope_id]['name']:
raise SubjectScopeNameExisting()
- subject_scope_id = uuid4().hex
- return self.driver.set_subject_scope_dict(intra_extension_id, subject_category_id, subject_scope_id, subject_scope_dict)
+ return self.driver.set_subject_scope_dict(intra_extension_id, subject_category_id, uuid4().hex, subject_scope_dict)
@filter_input
@enforce("read", "subject_scopes")
for rule_id in rules_dict:
if subject_scope_id in rules_dict[rule_id]:
self.driver.del_rule(intra_extension_id, sub_meta_rule_id, rule_id)
- return self.driver.del_subject_scope(intra_extension_id, subject_category_id, subject_scope_id)
+ self.driver.del_subject_scope(intra_extension_id, subject_category_id, subject_scope_id)
@filter_input
@enforce(("read", "write"), "subject_scopes")
@enforce("read", "subject_categories")
- def set_subject_scope_dict(self, user_id, intra_extension_id, subject_category_id, subject_scope_id, subject_scope_name):
+ def set_subject_scope_dict(self, user_id, intra_extension_id, subject_category_id, subject_scope_id, subject_scope_dict):
if subject_category_id not in self.driver.get_subject_categories_dict(intra_extension_id):
raise SubjectCategoryUnknown()
- subject_scope_dict = self.driver.get_subject_scopes_dict(intra_extension_id, subject_category_id)
- for _subject_scope_id in subject_scope_dict:
- if subject_scope_name == subject_scope_dict[_subject_scope_id]['name']:
+ subject_scopes_dict = self.driver.get_subject_scopes_dict(intra_extension_id, subject_category_id)
+ for _subject_scope_id in subject_scopes_dict:
+ if subject_scopes_dict[_subject_scope_id]['name'] == subject_scope_dict['name']:
raise SubjectScopeNameExisting()
return self.driver.set_subject_scope_dict(intra_extension_id, subject_category_id, subject_scope_id, subject_scope_dict)
@filter_input
@enforce(("read", "write"), "object_scopes")
@enforce("read", "object_categories")
- def add_object_scope_dict(self, user_id, intra_extension_id, object_category_id, object_scope_name):
+ def add_object_scope_dict(self, user_id, intra_extension_id, object_category_id, object_scope_dict):
if object_category_id not in self.driver.get_object_categories_dict(intra_extension_id):
raise ObjectCategoryUnknown()
- object_scope_dict = self.driver.get_object_scopes_dict(intra_extension_id, object_category_id)
- for _object_scope_id in object_scope_dict:
- if object_scope_name == object_scope_dict[_object_scope_id]['name']:
+ object_scopes_dict = self.driver.get_object_scopes_dict(intra_extension_id, object_category_id)
+ for _object_scope_id in object_scopes_dict:
+ if object_scopes_dict[_object_scope_id]['name'] == object_scope_dict['name']:
raise ObjectScopeNameExisting()
- object_scope_id = uuid4().hex
- return self.driver.set_object_scope_dict(
- intra_extension_id,
- object_category_id,
- object_scope_id,
- object_scope_name)
+ return self.driver.set_object_scope_dict(intra_extension_id, object_category_id, uuid4().hex, object_scope_dict)
@filter_input
@enforce("read", "object_scopes")
def get_object_scope_dict(self, user_id, intra_extension_id, object_category_id, object_scope_id):
if object_category_id not in self.driver.get_object_categories_dict(intra_extension_id):
raise ObjectCategoryUnknown()
- object_scopte_dict = self.driver.get_object_scopes_dict(intra_extension_id, object_category_id)
- if object_scope_id not in object_scopte_dict:
+ object_scopes_dict = self.driver.get_object_scopes_dict(intra_extension_id, object_category_id)
+ if object_scope_id not in object_scopes_dict:
raise ObjectScopeUnknown()
- return object_scopte_dict[object_scope_id]
+ return object_scopes_dict[object_scope_id]
@filter_input
@enforce(("read", "write"), "object_scopes")
for rule_id in rules_dict:
if object_scope_id in rules_dict[rule_id]:
self.driver.del_rule(intra_extension_id, sub_meta_rule_id, rule_id)
- return self.driver.del_object_scope(intra_extension_id, object_category_id, object_scope_id)
+ self.driver.del_object_scope(intra_extension_id, object_category_id, object_scope_id)
@filter_input
@enforce(("read", "write"), "object_scopes")
@enforce("read", "object_categories")
- def set_object_scope_dict(self, user_id, intra_extension_id, object_category_id, object_scope_id, object_scope_name):
+ def set_object_scope_dict(self, user_id, intra_extension_id, object_category_id, object_scope_id, object_scope_dict):
if object_category_id not in self.driver.get_object_categories_dict(intra_extension_id):
raise ObjectCategoryUnknown()
- object_scope_dict = self.driver.get_object_scopes_dict(intra_extension_id, object_category_id)
- for _object_scope_id in object_scope_dict:
- if object_scope_name == object_scope_dict[_object_scope_id]['name']:
+ object_scopes_dict = self.driver.get_object_scopes_dict(intra_extension_id, object_category_id)
+ for _object_scope_id in object_scopes_dict:
+ if object_scopes_dict[_object_scope_id]['name'] == object_scope_dict['name']:
raise ObjectScopeNameExisting()
return self.driver.set_object_scope_dict(intra_extension_id, object_category_id, object_scope_id, object_scope_dict)
@filter_input
@enforce(("read", "write"), "action_scopes")
@enforce("read", "action_categories")
- def add_action_scope_dict(self, user_id, intra_extension_id, action_category_id, action_scope_name):
+ def add_action_scope_dict(self, user_id, intra_extension_id, action_category_id, action_scope_dict):
if action_category_id not in self.driver.get_action_categories_dict(intra_extension_id):
raise ActionCategoryUnknown()
- action_scope_dict = self.driver.get_action_scopes_dict(intra_extension_id, action_category_id)
- for _action_scope_id in action_scope_dict:
- if action_scope_name == action_scope_dict[_action_scope_id]['name']:
+ action_scopes_dict = self.driver.get_action_scopes_dict(intra_extension_id, action_category_id)
+ for _action_scope_id in action_scopes_dict:
+ if action_scopes_dict[_action_scope_id]['name'] == action_scope_dict['name']:
raise ActionScopeNameExisting()
- action_scope_id = uuid4().hex
- return self.driver.set_action_scope_dict(
- intra_extension_id,
- action_category_id,
- action_scope_id,
- action_scope_name)
+ return self.driver.set_action_scope_dict(intra_extension_id, action_category_id, uuid4().hex, action_scope_dict)
@filter_input
@enforce("read", "action_scopes")
def get_action_scope_dict(self, user_id, intra_extension_id, action_category_id, action_scope_id):
if action_category_id not in self.driver.get_action_categories_dict(intra_extension_id):
raise ActionCategoryUnknown()
- action_scopte_dict = self.driver.get_action_scopes_dict(intra_extension_id, action_category_id)
- if action_scope_id not in action_scopte_dict:
+ action_scopes_dict = self.driver.get_action_scopes_dict(intra_extension_id, action_category_id)
+ if action_scope_id not in action_scopes_dict:
raise ActionScopeUnknown()
- return action_scopte_dict[action_scope_id]
+ return action_scopes_dict[action_scope_id]
@filter_input
@enforce(("read", "write"), "action_scopes")
for rule_id in rules_dict:
if action_scope_id in rules_dict[rule_id]:
self.driver.del_rule(intra_extension_id, sub_meta_rule_id, rule_id)
- return self.driver.del_action_scope(intra_extension_id, action_category_id, action_scope_id)
+ self.driver.del_action_scope(intra_extension_id, action_category_id, action_scope_id)
@filter_input
@enforce(("read", "write"), "action_scopes")
@enforce("read", "action_categories")
- def set_action_scope_dict(self, user_id, intra_extension_id, action_category_id, action_scope_id, action_scope_name):
+ def set_action_scope_dict(self, user_id, intra_extension_id, action_category_id, action_scope_id, action_scope_dict):
if action_category_id not in self.driver.get_action_categories_dict(intra_extension_id):
raise ActionCategoryUnknown()
- action_scope_dict = self.driver.get_action_scopes_dict(intra_extension_id, action_category_id)
- for _action_scope_id in action_scope_dict:
- if action_scope_name == action_scope_dict[_action_scope_id]['name']:
+ action_scopes_dict = self.driver.get_action_scopes_dict(intra_extension_id, action_category_id)
+ for _action_scope_id in action_scopes_dict:
+ if action_scopes_dict[_action_scope_id]['name'] == action_scope_dict['name']:
raise ActionScopeNameExisting()
- return self.driver.set_action_scope_dict(
- intra_extension_id,
- action_category_id,
- action_scope_id,
- action_scope_name)
+ return self.driver.set_action_scope_dict(intra_extension_id, action_category_id, action_scope_id, action_scope_dict)
# Assignment functions
@enforce("read", "subjects")
@enforce("read", "subject_categories")
def get_subject_assignment_list(self, user_id, intra_extension_id, subject_id, subject_category_id):
+ """
+ :param user_id:
+ :param intra_extension_id:
+ :param subject_id:
+ :param subject_category_id:
+ :return: [
+ subject_scope_id1, ..., subject_scope_idn
+ ]
+ """
if subject_id not in self.driver.get_subjects_dict(intra_extension_id):
raise SubjectUnknown()
- elif subject_category_id not in self.driver.get_subject_categories_dict(intra_extension_id):
+ if subject_category_id not in self.driver.get_subject_categories_dict(intra_extension_id):
raise SubjectCategoryUnknown()
return self.driver.get_subject_assignment_list(intra_extension_id, subject_id, subject_category_id)
@enforce("read", "subject_categories")
@enforce("read", "subject_scopes")
def add_subject_assignment_list(self, user_id, intra_extension_id, subject_id, subject_category_id, subject_scope_id):
- print(subject_id, self.driver.get_subjects_dict(intra_extension_id))
if subject_id not in self.driver.get_subjects_dict(intra_extension_id):
raise SubjectUnknown()
if subject_category_id not in self.driver.get_subject_categories_dict(intra_extension_id):
raise SubjectCategoryUnknown()
if subject_scope_id not in self.driver.get_subject_scopes_dict(intra_extension_id, subject_category_id):
raise SubjectScopeUnknown()
- if subject_scope_id in self.driver.get_subject_assignment_list(intra_extension_id, subject_id, subject_category_id):
+ elif subject_scope_id in self.driver.get_subject_assignment_list(intra_extension_id, subject_id, subject_category_id):
raise SubjectAssignmentExisting()
return self.driver.add_subject_assignment_list(intra_extension_id, subject_id, subject_category_id, subject_scope_id)
def del_subject_assignment(self, user_id, intra_extension_id, subject_id, subject_category_id, subject_scope_id):
if subject_id not in self.driver.get_subjects_dict(intra_extension_id):
raise SubjectUnknown()
- elif subject_category_id not in self.driver.get_subject_categories_dict(intra_extension_id):
+ if subject_category_id not in self.driver.get_subject_categories_dict(intra_extension_id):
raise SubjectCategoryUnknown()
- elif subject_scope_id not in self.driver.get_subject_scopes_dict(intra_extension_id, subject_category_id):
+ if subject_scope_id not in self.driver.get_subject_scopes_dict(intra_extension_id, subject_category_id):
raise SubjectScopeUnknown()
elif subject_scope_id not in self.driver.get_subject_assignment_list(intra_extension_id, subject_id, subject_category_id):
raise SubjectAssignmentUnknown()
def get_object_assignment_list(self, user_id, intra_extension_id, object_id, object_category_id):
if object_id not in self.driver.get_objects_dict(intra_extension_id):
raise ObjectUnknown()
- elif object_category_id not in self.driver.get_object_categories_dict(intra_extension_id):
+ if object_category_id not in self.driver.get_object_categories_dict(intra_extension_id):
raise ObjectCategoryUnknown()
return self.driver.get_object_assignment_list(intra_extension_id, object_id, object_category_id)
def add_object_assignment_list(self, user_id, intra_extension_id, object_id, object_category_id, object_scope_id):
if object_id not in self.driver.get_objects_dict(intra_extension_id):
raise ObjectUnknown()
- elif object_category_id not in self.driver.get_object_categories_dict(intra_extension_id):
+ if object_category_id not in self.driver.get_object_categories_dict(intra_extension_id):
raise ObjectCategoryUnknown()
- elif object_scope_id not in self.driver.get_object_scopes_dict(intra_extension_id, object_category_id):
+ if object_scope_id not in self.driver.get_object_scopes_dict(intra_extension_id, object_category_id):
raise ObjectScopeUnknown()
elif object_scope_id in self.driver.get_object_assignment_list(intra_extension_id, object_id, object_category_id):
raise ObjectAssignmentExisting()
def del_object_assignment(self, user_id, intra_extension_id, object_id, object_category_id, object_scope_id):
if object_id not in self.driver.get_objects_dict(intra_extension_id):
raise ObjectUnknown()
- elif object_category_id not in self.driver.get_object_categories_dict(intra_extension_id):
+ if object_category_id not in self.driver.get_object_categories_dict(intra_extension_id):
raise ObjectCategoryUnknown()
- elif object_scope_id not in self.driver.get_object_scopes_dict(intra_extension_id, object_category_id):
+ if object_scope_id not in self.driver.get_object_scopes_dict(intra_extension_id, object_category_id):
raise ObjectScopeUnknown()
elif object_scope_id not in self.driver.get_object_assignment_list(intra_extension_id, object_id, object_category_id):
raise ObjectAssignmentUnknown()
def get_action_assignment_list(self, user_id, intra_extension_id, action_id, action_category_id):
if action_id not in self.driver.get_actions_dict(intra_extension_id):
raise ActionUnknown()
- elif action_category_id not in self.driver.get_action_categories_dict(intra_extension_id):
+ if action_category_id not in self.driver.get_action_categories_dict(intra_extension_id):
raise ActionCategoryUnknown()
return self.driver.get_action_assignment_list(intra_extension_id, action_id, action_category_id)
def add_action_assignment_list(self, user_id, intra_extension_id, action_id, action_category_id, action_scope_id):
if action_id not in self.driver.get_actions_dict(intra_extension_id):
raise ActionUnknown()
- elif action_category_id not in self.driver.get_action_categories_dict(intra_extension_id):
+ if action_category_id not in self.driver.get_action_categories_dict(intra_extension_id):
raise ActionCategoryUnknown()
- elif action_scope_id not in self.driver.get_action_scopes_dict(intra_extension_id, action_category_id):
+ if action_scope_id not in self.driver.get_action_scopes_dict(intra_extension_id, action_category_id):
raise ActionScopeUnknown()
elif action_scope_id in self.driver.get_action_assignment_list(intra_extension_id, action_id, action_category_id):
raise ObjectAssignmentExisting()
def del_action_assignment(self, user_id, intra_extension_id, action_id, action_category_id, action_scope_id):
if action_id not in self.driver.get_actions_dict(intra_extension_id):
raise ActionUnknown()
- elif action_category_id not in self.driver.get_action_categories_dict(intra_extension_id):
+ if action_category_id not in self.driver.get_action_categories_dict(intra_extension_id):
raise ActionCategoryUnknown()
- elif action_scope_id not in self.driver.get_action_scopes_dict(intra_extension_id, action_category_id):
+ if action_scope_id not in self.driver.get_action_scopes_dict(intra_extension_id, action_category_id):
raise ActionScopeUnknown()
elif action_scope_id not in self.driver.get_action_assignment_list(intra_extension_id, action_id, action_category_id):
raise ActionAssignmentUnknown()
- return self.driver.del_action_assignment(intra_extension_id, action_id, action_category_id, action_scope_id)
+ self.driver.del_action_assignment(intra_extension_id, action_id, action_category_id, action_scope_id)
# Metarule functions
- @filter_input
- @enforce(("read", "write"), "aggregation_algorithm")
- def set_aggregation_algorithm_dict(self, user_id, intra_extension_id, aggregation_algorithm_id, aggregation_algorithm_dict):
- if aggregation_algorithm_id:
- if aggregation_algorithm_id not in self.configuration_api.get_aggregation_algorithms_dict(ADMIN_ID):
- raise AggregationAlgorithmUnknown()
- else:
- aggregation_algorithm_id = uuid4().hex
- return self.driver.set_aggregation_algorithm_dict(intra_extension_id, aggregation_algorithm_id, aggregation_algorithm_dict)
-
@filter_input
@enforce("read", "aggregation_algorithm")
def get_aggregation_algorithm_dict(self, user_id, intra_extension_id):
raise AggregationAlgorithmNotExisting()
return aggregation_algorithm_dict
+ @filter_input
+ @enforce(("read", "write"), "aggregation_algorithm")
+ def set_aggregation_algorithm_dict(self, user_id, intra_extension_id, aggregation_algorithm_id, aggregation_algorithm_dict):
+ if aggregation_algorithm_id:
+ if aggregation_algorithm_id not in self.configuration_api.get_aggregation_algorithms_dict(ADMIN_ID):
+ raise AggregationAlgorithmUnknown()
+ else:
+ aggregation_algorithm_id = uuid4().hex
+ return self.driver.set_aggregation_algorithm_dict(intra_extension_id, aggregation_algorithm_id, aggregation_algorithm_dict)
+
@filter_input
@enforce("read", "sub_meta_rules")
def get_sub_meta_rules_dict(self, user_id, intra_extension_id):
for _sub_meta_rule_id in sub_meta_rules_dict:
if sub_meta_rule_dict['name'] == sub_meta_rules_dict[_sub_meta_rule_id]["name"]:
raise SubMetaRuleNameExisting()
- elif sub_meta_rule_dict['subject_categories'] == sub_meta_rules_dict[_sub_meta_rule_id]["subject_categories"] and \
+ if sub_meta_rule_dict['subject_categories'] == sub_meta_rules_dict[_sub_meta_rule_id]["subject_categories"] and \
sub_meta_rule_dict['object_categories'] == sub_meta_rules_dict[_sub_meta_rule_id]["object_categories"] and \
sub_meta_rule_dict['action_categories'] == sub_meta_rules_dict[_sub_meta_rule_id]["action_categories"] and \
sub_meta_rule_dict['algorithm'] == sub_meta_rules_dict[_sub_meta_rule_id]["algorithm"]:
raise SubMetaRuleExisting()
- algorithms = self.configuration_api.get_sub_meta_rule_algorithms_dict(user_id)
- if sub_meta_rule_dict['algorithm'] not in algorithms.keys():
+ if sub_meta_rule_dict['algorithm'] not in self.configuration_api.get_sub_meta_rule_algorithms_dict(user_id):
raise SubMetaRuleAlgorithmNotExisting()
sub_meta_rule_id = uuid4().hex
- # TODO (dthom): add new sub-meta-rule to rule
+ # TODO (dthom): add new sub-meta-rule to rule dict
# self.driver.add_rule(intra_extension_id, sub_meta_rule_id, [])
return self.driver.set_sub_meta_rule_dict(intra_extension_id, sub_meta_rule_id, sub_meta_rule_dict)
def set_sub_meta_rule_dict(self, user_id, intra_extension_id, sub_meta_rule_id, sub_meta_rule_dict):
if sub_meta_rule_id not in self.driver.get_sub_meta_rules_dict(intra_extension_id):
raise SubMetaRuleUnknown()
- # TODO (dthom): add new sub-meta-rule to rule
- # self.driver.add_rule(intra_extension_id, sub_meta_rule_id, [])
return self.driver.set_sub_meta_rule_dict(intra_extension_id, sub_meta_rule_id, sub_meta_rule_dict)
# Rule functions
raise SubMetaRuleUnknown()
if rule_list in self.driver.get_rules_dict(intra_extension_id, sub_meta_rule_id).values():
raise RuleExisting()
- rule_id = uuid4().hex
- return self.driver.set_rule_dict(intra_extension_id, sub_meta_rule_id, rule_id, rule_list)
+ return self.driver.set_rule_dict(intra_extension_id, sub_meta_rule_id, uuid4().hex, rule_list)
@filter_input
@enforce("read", "sub_meta_rules")
def del_rule(self, user_id, intra_extension_id, sub_meta_rule_id, rule_id):
if sub_meta_rule_id not in self.driver.get_sub_meta_rules_dict(intra_extension_id):
raise SubMetaRuleUnknown()
- rules_dict = self.driver.get_rules_dict(intra_extension_id, sub_meta_rule_id)
- if rule_id not in rules_dict:
+ if rule_id not in self.driver.get_rules_dict(intra_extension_id, sub_meta_rule_id):
raise RuleUnknown()
self.driver.del_rule(intra_extension_id, sub_meta_rule_id, rule_id)
def set_rule_dict(self, user_id, intra_extension_id, sub_meta_rule_id, rule_id, rule_list):
if sub_meta_rule_id not in self.driver.get_sub_meta_rules_dict(intra_extension_id):
raise SubMetaRuleUnknown()
- rules_dict = self.driver.get_rules_dict(intra_extension_id, sub_meta_rule_id)
- if rule_id not in rules_dict:
+ if rule_id not in self.driver.get_rules_dict(intra_extension_id, sub_meta_rule_id):
raise RuleUnknown()
return self.driver.set_rule_dict(intra_extension_id, sub_meta_rule_id, rule_id, rule_list)
new_subject_category = {"name": "subject_category_test", "description": "subject_category_test"}
- subject_categories = self.admin_manager.add_subject_category(admin_subject_id, authz_ie_dict["id"], new_subject_category)
+ subject_categories = self.admin_manager.add_subject_category_dict(admin_subject_id, authz_ie_dict["id"], new_subject_category)
_subject_categories = dict(subject_categories)
self.assertEqual(len(_subject_categories.keys()), 1)
new_subject_category["id"] = _subject_categories.keys()[0]
{"name": "demo", "description": "demo"})
demo_subject_id, demo_subject_dict = \
self.tenant_api.get_subject_dict_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next()
- object_categories = self.manager.get_object_category_dict(admin_subject_id, authz_ie_dict["id"])
+ object_categories = self.manager.get_object_categories_dict(admin_subject_id, authz_ie_dict["id"])
self.assertIsInstance(object_categories, dict)
for key, value in object_categories.iteritems():
self.assertIsInstance(value, dict)
new_object_category = {"name": "object_category_test", "description": "object_category_test"}
- object_categories = self.admin_manager.add_object_category(admin_subject_id, authz_ie_dict["id"], new_object_category)
+ object_categories = self.admin_manager.add_object_category_dict(admin_subject_id, authz_ie_dict["id"], new_object_category)
_object_categories = dict(object_categories)
self.assertEqual(len(_object_categories.keys()), 1)
new_object_category["id"] = _object_categories.keys()[0]
# Delete the new object_category
self.admin_manager.del_object_category(admin_subject_id, authz_ie_dict["id"], new_object_category["id"])
- object_categories = self.manager.get_object_category_dict(admin_subject_id, authz_ie_dict["id"])
+ object_categories = self.manager.get_object_categories_dict(admin_subject_id, authz_ie_dict["id"])
for key, value in object_categories.iteritems():
self.assertIsInstance(value, dict)
self.assertIn("name", value)
{"name": "demo", "description": "demo"})
demo_subject_id, demo_subject_dict = \
self.tenant_api.get_subject_dict_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next()
- action_categories = self.manager.get_action_category_dict(admin_subject_id, authz_ie_dict["id"])
+ action_categories = self.manager.get_action_categories_dict(admin_subject_id, authz_ie_dict["id"])
self.assertIsInstance(action_categories, dict)
for key, value in action_categories.iteritems():
self.assertIsInstance(value, dict)
new_action_category = {"name": "action_category_test", "description": "action_category_test"}
- action_categories = self.admin_manager.add_action_category(admin_subject_id, authz_ie_dict["id"], new_action_category)
+ action_categories = self.admin_manager.add_action_category_dict(admin_subject_id, authz_ie_dict["id"], new_action_category)
_action_categories = dict(action_categories)
self.assertEqual(len(_action_categories.keys()), 1)
new_action_category["id"] = _action_categories.keys()[0]
# Delete the new action_category
self.admin_manager.del_action_category(admin_subject_id, authz_ie_dict["id"], new_action_category["id"])
- action_categories = self.manager.get_action_category_dict(admin_subject_id, authz_ie_dict["id"])
+ action_categories = self.manager.get_action_categories_dict(admin_subject_id, authz_ie_dict["id"])
for key, value in action_categories.iteritems():
self.assertIsInstance(value, dict)
self.assertIn("name", value)
demo_subject_id, demo_subject_dict = \
self.tenant_api.get_subject_dict_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next()
- subject_categories = self.admin_manager.add_subject_category(
+ subject_categories = self.admin_manager.add_subject_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
demo_subject_id, demo_subject_dict = \
self.tenant_api.get_subject_dict_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next()
- object_categories = self.admin_manager.add_object_category(
+ object_categories = self.admin_manager.add_object_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
demo_subject_id, demo_subject_dict = \
self.tenant_api.get_subject_dict_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next()
- action_categories = self.admin_manager.add_action_category(
+ action_categories = self.admin_manager.add_action_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
subjects_dict = self.manager.get_subjects_dict(admin_subject_id, authz_ie_dict["id"])
- subject_categories = self.admin_manager.add_subject_category(
+ subject_categories = self.admin_manager.add_subject_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
if not object_vm1_id or not object_vm2_id:
raise Exception("Cannot run tests, database is corrupted ? (need upload and list in objects)")
- object_categories = self.admin_manager.add_object_category(
+ object_categories = self.admin_manager.add_object_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
if not action_upload_id or not action_list_id:
raise Exception("Cannot run tests, database is corrupted ? (need upload and list in actions)")
- action_categories = self.admin_manager.add_action_category(
+ action_categories = self.admin_manager.add_action_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
self.assertIsInstance(sub_meta_rules, dict)
categories = {
"subject_categories": self.admin_manager.get_subject_categories_dict(admin_subject_id, authz_ie_dict["id"]),
- "object_categories": self.admin_manager.get_object_category_dict(admin_subject_id, authz_ie_dict["id"]),
- "action_categories": self.admin_manager.get_action_category_dict(admin_subject_id, authz_ie_dict["id"])
+ "object_categories": self.admin_manager.get_object_categories_dict(admin_subject_id, authz_ie_dict["id"]),
+ "action_categories": self.admin_manager.get_action_categories_dict(admin_subject_id, authz_ie_dict["id"])
}
for key, value in sub_meta_rules.iteritems():
self.assertIsInstance(value, dict)
new_subject_category = {"name": "subject_category_test", "description": "subject_category_test"}
self.assertRaises(
AuthzException,
- self.manager.add_subject_category,
+ self.manager.add_subject_category_dict,
demo_subject_id, admin_ie_dict["id"], new_subject_category)
- subject_categories = self.admin_manager.add_subject_category(admin_subject_id, authz_ie_dict["id"], new_subject_category)
+ subject_categories = self.admin_manager.add_subject_category_dict(admin_subject_id, authz_ie_dict["id"], new_subject_category)
_subject_categories = dict(subject_categories)
self.assertEqual(len(_subject_categories.keys()), 1)
new_subject_category["id"] = _subject_categories.keys()[0]
{"name": "demo", "description": "demo"})
demo_subject_id, demo_subject_dict = \
self.tenant_api.get_subject_dict_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next()
- object_categories = self.manager.get_object_category_dict(admin_subject_id, authz_ie_dict["id"])
+ object_categories = self.manager.get_object_categories_dict(admin_subject_id, authz_ie_dict["id"])
self.assertIsInstance(object_categories, dict)
for key, value in object_categories.iteritems():
self.assertIsInstance(value, dict)
new_object_category = {"name": "object_category_test", "description": "object_category_test"}
self.assertRaises(
AuthzException,
- self.manager.add_object_category,
+ self.manager.add_object_category_dict,
demo_subject_id, admin_ie_dict["id"], new_object_category)
- object_categories = self.admin_manager.add_object_category(admin_subject_id, authz_ie_dict["id"], new_object_category)
+ object_categories = self.admin_manager.add_object_category_dict(admin_subject_id, authz_ie_dict["id"], new_object_category)
_object_categories = dict(object_categories)
self.assertEqual(len(_object_categories.keys()), 1)
new_object_category["id"] = _object_categories.keys()[0]
demo_subject_id, authz_ie_dict["id"], new_object_category["id"])
self.admin_manager.del_object_category(admin_subject_id, authz_ie_dict["id"], new_object_category["id"])
- object_categories = self.manager.get_object_category_dict(admin_subject_id, authz_ie_dict["id"])
+ object_categories = self.manager.get_object_categories_dict(admin_subject_id, authz_ie_dict["id"])
for key, value in object_categories.iteritems():
self.assertIsInstance(value, dict)
self.assertIn("name", value)
{"name": "demo", "description": "demo"})
demo_subject_id, demo_subject_dict = \
self.tenant_api.get_subject_dict_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next()
- action_categories = self.manager.get_action_category_dict(admin_subject_id, authz_ie_dict["id"])
+ action_categories = self.manager.get_action_categories_dict(admin_subject_id, authz_ie_dict["id"])
self.assertIsInstance(action_categories, dict)
for key, value in action_categories.iteritems():
self.assertIsInstance(value, dict)
new_action_category = {"name": "action_category_test", "description": "action_category_test"}
self.assertRaises(
AuthzException,
- self.manager.add_action_category,
+ self.manager.add_action_category_dict,
demo_subject_id, admin_ie_dict["id"], new_action_category)
- action_categories = self.admin_manager.add_action_category(admin_subject_id, authz_ie_dict["id"], new_action_category)
+ action_categories = self.admin_manager.add_action_category_dict(admin_subject_id, authz_ie_dict["id"], new_action_category)
_action_categories = dict(action_categories)
self.assertEqual(len(_action_categories.keys()), 1)
new_action_category["id"] = _action_categories.keys()[0]
demo_subject_id, authz_ie_dict["id"], new_action_category["id"])
self.admin_manager.del_action_category(admin_subject_id, authz_ie_dict["id"], new_action_category["id"])
- action_categories = self.manager.get_action_category_dict(admin_subject_id, authz_ie_dict["id"])
+ action_categories = self.manager.get_action_categories_dict(admin_subject_id, authz_ie_dict["id"])
for key, value in action_categories.iteritems():
self.assertIsInstance(value, dict)
self.assertIn("name", value)
demo_subject_id, demo_subject_dict = \
self.tenant_api.get_subject_dict_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next()
- subject_categories = self.admin_manager.add_subject_category(
+ subject_categories = self.admin_manager.add_subject_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
demo_subject_id, demo_subject_dict = \
self.tenant_api.get_subject_dict_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next()
- object_categories = self.admin_manager.add_object_category(
+ object_categories = self.admin_manager.add_object_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
demo_subject_id, demo_subject_dict = \
self.tenant_api.get_subject_dict_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next()
- action_categories = self.admin_manager.add_action_category(
+ action_categories = self.admin_manager.add_action_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
subjects_dict = self.manager.get_subjects_dict(admin_subject_id, authz_ie_dict["id"])
- subject_categories = self.admin_manager.add_subject_category(
+ subject_categories = self.admin_manager.add_subject_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
if not object_vm1_id or not object_vm2_id:
raise Exception("Cannot run tests, database is corrupted ? (need upload and list in objects)")
- object_categories = self.admin_manager.add_object_category(
+ object_categories = self.admin_manager.add_object_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
if not action_upload_id or not action_list_id:
raise Exception("Cannot run tests, database is corrupted ? (need upload and list in actions)")
- action_categories = self.admin_manager.add_action_category(
+ action_categories = self.admin_manager.add_action_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
self.assertIsInstance(sub_meta_rules, dict)
categories = {
"subject_categories": self.admin_manager.get_subject_categories_dict(admin_subject_id, authz_ie_dict["id"]),
- "object_categories": self.admin_manager.get_object_category_dict(admin_subject_id, authz_ie_dict["id"]),
- "action_categories": self.admin_manager.get_action_category_dict(admin_subject_id, authz_ie_dict["id"])
+ "object_categories": self.admin_manager.get_object_categories_dict(admin_subject_id, authz_ie_dict["id"]),
+ "action_categories": self.admin_manager.get_action_categories_dict(admin_subject_id, authz_ie_dict["id"])
}
for key, value in sub_meta_rules.iteritems():
self.assertIsInstance(value, dict)
new_subject_category = {"name": "subject_category_test", "description": "subject_category_test"}
- subject_categories = self.admin_manager.add_subject_category(admin_subject_id, authz_ie_dict["id"], new_subject_category)
+ subject_categories = self.admin_manager.add_subject_category_dict(admin_subject_id, authz_ie_dict["id"], new_subject_category)
_subject_categories = dict(subject_categories)
self.assertEqual(len(_subject_categories.keys()), 1)
new_subject_category["id"] = _subject_categories.keys()[0]
{"name": "demo", "description": "demo"})
demo_subject_id, demo_subject_dict = \
self.tenant_api.get_subject_dict_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next()
- object_categories = self.manager.get_object_category_dict(admin_subject_id, authz_ie_dict["id"])
+ object_categories = self.manager.get_object_categories_dict(admin_subject_id, authz_ie_dict["id"])
self.assertIsInstance(object_categories, dict)
for key, value in object_categories.iteritems():
self.assertIsInstance(value, dict)
new_object_category = {"name": "object_category_test", "description": "object_category_test"}
- object_categories = self.admin_manager.add_object_category(admin_subject_id, authz_ie_dict["id"], new_object_category)
+ object_categories = self.admin_manager.add_object_category_dict(admin_subject_id, authz_ie_dict["id"], new_object_category)
_object_categories = dict(object_categories)
self.assertEqual(len(_object_categories.keys()), 1)
new_object_category["id"] = _object_categories.keys()[0]
# Delete the new object_category
self.admin_manager.del_object_category(admin_subject_id, authz_ie_dict["id"], new_object_category["id"])
- object_categories = self.manager.get_object_category_dict(admin_subject_id, authz_ie_dict["id"])
+ object_categories = self.manager.get_object_categories_dict(admin_subject_id, authz_ie_dict["id"])
for key, value in object_categories.iteritems():
self.assertIsInstance(value, dict)
self.assertIn("name", value)
{"name": "demo", "description": "demo"})
demo_subject_id, demo_subject_dict = \
self.tenant_api.get_subject_dict_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next()
- action_categories = self.manager.get_action_category_dict(admin_subject_id, authz_ie_dict["id"])
+ action_categories = self.manager.get_action_categories_dict(admin_subject_id, authz_ie_dict["id"])
self.assertIsInstance(action_categories, dict)
for key, value in action_categories.iteritems():
self.assertIsInstance(value, dict)
new_action_category = {"name": "action_category_test", "description": "action_category_test"}
- action_categories = self.admin_manager.add_action_category(admin_subject_id, authz_ie_dict["id"], new_action_category)
+ action_categories = self.admin_manager.add_action_category_dict(admin_subject_id, authz_ie_dict["id"], new_action_category)
_action_categories = dict(action_categories)
self.assertEqual(len(_action_categories.keys()), 1)
new_action_category["id"] = _action_categories.keys()[0]
# Delete the new action_category
self.admin_manager.del_action_category(admin_subject_id, authz_ie_dict["id"], new_action_category["id"])
- action_categories = self.manager.get_action_category_dict(admin_subject_id, authz_ie_dict["id"])
+ action_categories = self.manager.get_action_categories_dict(admin_subject_id, authz_ie_dict["id"])
for key, value in action_categories.iteritems():
self.assertIsInstance(value, dict)
self.assertIn("name", value)
demo_subject_id, demo_subject_dict = \
self.tenant_api.get_subject_dict_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next()
- subject_categories = self.admin_manager.add_subject_category(
+ subject_categories = self.admin_manager.add_subject_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
demo_subject_id, demo_subject_dict = \
self.tenant_api.get_subject_dict_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next()
- object_categories = self.admin_manager.add_object_category(
+ object_categories = self.admin_manager.add_object_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
demo_subject_id, demo_subject_dict = \
self.tenant_api.get_subject_dict_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next()
- action_categories = self.admin_manager.add_action_category(
+ action_categories = self.admin_manager.add_action_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
subjects_dict = self.manager.get_subjects_dict(admin_subject_id, authz_ie_dict["id"])
- subject_categories = self.admin_manager.add_subject_category(
+ subject_categories = self.admin_manager.add_subject_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
if not object_vm1_id or not object_vm2_id:
raise Exception("Cannot run tests, database is corrupted ? (need upload and list in objects)")
- object_categories = self.admin_manager.add_object_category(
+ object_categories = self.admin_manager.add_object_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
if not action_upload_id or not action_list_id:
raise Exception("Cannot run tests, database is corrupted ? (need upload and list in actions)")
- action_categories = self.admin_manager.add_action_category(
+ action_categories = self.admin_manager.add_action_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
self.assertIsInstance(sub_meta_rules, dict)
categories = {
"subject_categories": self.admin_manager.get_subject_categories_dict(admin_subject_id, authz_ie_dict["id"]),
- "object_categories": self.admin_manager.get_object_category_dict(admin_subject_id, authz_ie_dict["id"]),
- "action_categories": self.admin_manager.get_action_category_dict(admin_subject_id, authz_ie_dict["id"])
+ "object_categories": self.admin_manager.get_object_categories_dict(admin_subject_id, authz_ie_dict["id"]),
+ "action_categories": self.admin_manager.get_action_categories_dict(admin_subject_id, authz_ie_dict["id"])
}
for key, value in sub_meta_rules.iteritems():
self.assertIsInstance(value, dict)
# Add a subject scope and test ObjectCategoryAssignmentOutOfScope
my_subject_category = {"name": "my_subject_category", "description": "my_subject_category description"}
- _tmp = self.manager.add_subject_category(
+ _tmp = self.manager.add_subject_category_dict(
admin_subject_id,
authz_ie_dict["id"],
my_subject_category
# Add an object scope and test ActionCategoryAssignmentOutOfScope
my_object_category = {"name": "my_object_category", "description": "my_object_category description"}
- _tmp = self.manager.add_object_category(
+ _tmp = self.manager.add_object_category_dict(
admin_subject_id,
authz_ie_dict["id"],
my_object_category
# Add an action scope and test SubjectCategoryAssignmentUnknown
my_action_category = {"name": "my_action_category", "description": "my_action_category description"}
- _tmp = self.manager.add_action_category(
+ _tmp = self.manager.add_action_category_dict(
admin_subject_id,
authz_ie_dict["id"],
my_action_category
new_subject_category = {"name": "subject_category_test", "description": "subject_category_test"}
self.assertRaises(
AuthzException,
- self.manager.add_subject_category,
+ self.manager.add_subject_category_dict,
demo_subject_id, admin_ie_dict["id"], new_subject_category)
- subject_categories = self.admin_manager.add_subject_category(admin_subject_id, authz_ie_dict["id"], new_subject_category)
+ subject_categories = self.admin_manager.add_subject_category_dict(admin_subject_id, authz_ie_dict["id"], new_subject_category)
_subject_categories = dict(subject_categories)
self.assertEqual(len(_subject_categories.keys()), 1)
new_subject_category["id"] = _subject_categories.keys()[0]
{"name": "demo", "description": "demo"})
demo_subject_id, demo_subject_dict = \
self.tenant_api.get_subject_dict_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next()
- object_categories = self.manager.get_object_category_dict(admin_subject_id, authz_ie_dict["id"])
+ object_categories = self.manager.get_object_categories_dict(admin_subject_id, authz_ie_dict["id"])
self.assertIsInstance(object_categories, dict)
for key, value in object_categories.iteritems():
self.assertIsInstance(value, dict)
new_object_category = {"name": "object_category_test", "description": "object_category_test"}
self.assertRaises(
AuthzException,
- self.manager.add_object_category,
+ self.manager.add_object_category_dict,
demo_subject_id, admin_ie_dict["id"], new_object_category)
- object_categories = self.admin_manager.add_object_category(admin_subject_id, authz_ie_dict["id"], new_object_category)
+ object_categories = self.admin_manager.add_object_category_dict(admin_subject_id, authz_ie_dict["id"], new_object_category)
_object_categories = dict(object_categories)
self.assertEqual(len(_object_categories.keys()), 1)
new_object_category["id"] = _object_categories.keys()[0]
demo_subject_id, authz_ie_dict["id"], new_object_category["id"])
self.admin_manager.del_object_category(admin_subject_id, authz_ie_dict["id"], new_object_category["id"])
- object_categories = self.manager.get_object_category_dict(admin_subject_id, authz_ie_dict["id"])
+ object_categories = self.manager.get_object_categories_dict(admin_subject_id, authz_ie_dict["id"])
for key, value in object_categories.iteritems():
self.assertIsInstance(value, dict)
self.assertIn("name", value)
{"name": "demo", "description": "demo"})
demo_subject_id, demo_subject_dict = \
self.tenant_api.get_subject_dict_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next()
- action_categories = self.manager.get_action_category_dict(admin_subject_id, authz_ie_dict["id"])
+ action_categories = self.manager.get_action_categories_dict(admin_subject_id, authz_ie_dict["id"])
self.assertIsInstance(action_categories, dict)
for key, value in action_categories.iteritems():
self.assertIsInstance(value, dict)
new_action_category = {"name": "action_category_test", "description": "action_category_test"}
self.assertRaises(
AuthzException,
- self.manager.add_action_category,
+ self.manager.add_action_category_dict,
demo_subject_id, admin_ie_dict["id"], new_action_category)
- action_categories = self.admin_manager.add_action_category(admin_subject_id, authz_ie_dict["id"], new_action_category)
+ action_categories = self.admin_manager.add_action_category_dict(admin_subject_id, authz_ie_dict["id"], new_action_category)
_action_categories = dict(action_categories)
self.assertEqual(len(_action_categories.keys()), 1)
new_action_category["id"] = _action_categories.keys()[0]
demo_subject_id, authz_ie_dict["id"], new_action_category["id"])
self.admin_manager.del_action_category(admin_subject_id, authz_ie_dict["id"], new_action_category["id"])
- action_categories = self.manager.get_action_category_dict(admin_subject_id, authz_ie_dict["id"])
+ action_categories = self.manager.get_action_categories_dict(admin_subject_id, authz_ie_dict["id"])
for key, value in action_categories.iteritems():
self.assertIsInstance(value, dict)
self.assertIn("name", value)
demo_subject_id, demo_subject_dict = \
self.tenant_api.get_subject_dict_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next()
- subject_categories = self.admin_manager.add_subject_category(
+ subject_categories = self.admin_manager.add_subject_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
demo_subject_id, demo_subject_dict = \
self.tenant_api.get_subject_dict_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next()
- object_categories = self.admin_manager.add_object_category(
+ object_categories = self.admin_manager.add_object_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
demo_subject_id, demo_subject_dict = \
self.tenant_api.get_subject_dict_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next()
- action_categories = self.admin_manager.add_action_category(
+ action_categories = self.admin_manager.add_action_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
subjects_dict = self.manager.get_subjects_dict(admin_subject_id, authz_ie_dict["id"])
- subject_categories = self.admin_manager.add_subject_category(
+ subject_categories = self.admin_manager.add_subject_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
if not object_vm1_id or not object_vm2_id:
raise Exception("Cannot run tests, database is corrupted ? (need upload and list in objects)")
- object_categories = self.admin_manager.add_object_category(
+ object_categories = self.admin_manager.add_object_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
if not action_upload_id or not action_list_id:
raise Exception("Cannot run tests, database is corrupted ? (need upload and list in actions)")
- action_categories = self.admin_manager.add_action_category(
+ action_categories = self.admin_manager.add_action_category_dict(
admin_subject_id,
authz_ie_dict["id"],
{
self.assertIsInstance(sub_meta_rules, dict)
categories = {
"subject_categories": self.admin_manager.get_subject_categories_dict(admin_subject_id, authz_ie_dict["id"]),
- "object_categories": self.admin_manager.get_object_category_dict(admin_subject_id, authz_ie_dict["id"]),
- "action_categories": self.admin_manager.get_action_category_dict(admin_subject_id, authz_ie_dict["id"])
+ "object_categories": self.admin_manager.get_object_categories_dict(admin_subject_id, authz_ie_dict["id"]),
+ "action_categories": self.admin_manager.get_action_categories_dict(admin_subject_id, authz_ie_dict["id"])
}
for key, value in sub_meta_rules.iteritems():
self.assertIsInstance(value, dict)
Code Review / moon.git / commitdiff