path: root/python_moonclient/python_moonclient/core/check_tools.py

from python_moonclient.core.cli_exceptions import MoonCliException


def check_optionnal_result(result):
    if type(result) is not dict:
        raise MoonCliException("Unexpected request result. It should be a dictionnary")
    if "result" in result:
        check_result(result)


def check_result(result):
    if type(result) is not dict or "result" not in result:
        raise MoonCliException(
            "Unexpected request result. It should be a dictionnary with a 'result' entry")
    if result["result"] is None:
        raise MoonCliException("Unexpected request result. The 'result' entry shall not be null")


def _check_generic_in_result(field, result, check_not_null=False):
    if type(field) is not str or type(result) is not dict or field not in result:
        raise MoonCliException(
            "Unexpected request result. It should be a dictionnary with a '{}' entry".format(field))
    if check_not_null is True and result[field] is None:
        raise MoonCliException(
            "Unexpected request result. The '{}' entry shall not be null".format(field))


def check_slaves_in_result(result):
    _check_generic_in_result("slaves", result)


def check_pdp_in_result(result):
    _check_generic_in_result("pdps", result)


def check_model_in_result(result, check_not_null=False):
    _check_generic_in_result("models", result)
    if check_not_null is True and result["models"] is None:
        raise MoonCliException("Unexpected request result. The 'models' entry shall not be null")


def check_meta_rule_in_result(result):
    _check_generic_in_result("meta_rules", result)


def check_rule_in_result(result):
    _check_generic_in_result("rules", result)


def check_subject_in_result(result):
    _check_generic_in_result("subjects", result)


def check_subject_category_in_result(result):
    _check_generic_in_result("subject_categories", result)


def check_object_category_in_result(result):
    _check_generic_in_result("object_categories", result)


def check_action_category_in_result(result):
    _check_generic_in_result("action_categories", result)


def check_policy_in_result(result):
    _check_generic_in_result("policies", result)


def check_object_in_result(result):
    _check_generic_in_result("objects", result)


def check_action_in_result(result):
    _check_generic_in_result("actions", result)


def check_subject_assignment_in_result(result):
    _check_generic_in_result("subject_assignments", result, True)


def check_object_assignment_in_result(result):
    _check_generic_in_result("object_assignments", result, True)


def check_action_assignment_in_result(result):
    _check_generic_in_result("action_assignments", result, True)


def check_pdp_id(pdp_id, result):
    check_pdp_in_result(result)
    if pdp_id not in result['pdps']:
        raise MoonCliException("Unexpected request result. Unknown pdp id")


def _check_generic_name(field, name, field_elt_id, result, do_check_name=True):
    if type(field) is str:
        if result[field] is None:
            raise MoonCliException(
                "Unexpected request result : {} shall not be empty".format(field))
        if field_elt_id not in result[field]:
            raise MoonCliException("Unexpected request result. Unknown {} id".format(field))
        if "name" not in result[field][field_elt_id]:
            raise MoonCliException(
                "Unexpected request result : {} with id {} has no name".format(field, field_elt_id))
        if do_check_name and name != result[field][field_elt_id]["name"]:
            raise MoonCliException(
                "Unexpected request result : {} with id {} has a bad name. Expected {}".format(
                    field, field_elt_id, name))


def check_model_name(name, model_id, result, do_check_name):
    _check_generic_name("models", name, model_id, result, do_check_name)


def check_pdp_name(name, pdp_id, result):
    _check_generic_name("pdps", name, pdp_id, result)


def check_subject_categories_name(name, category_id, result):
    _check_generic_name("subject_categories", name, category_id, result)


def check_object_categories_name(name, category_id, result):
    _check_generic_name("object_categories", name, category_id, result)


def check_action_categories_name(name, category_id, result):
    _check_generic_name("action_categories", name, category_id, result)


def check_meta_rules_name(name, meta_rule_id, result):
    _check_generic_name("meta_rules", name, meta_rule_id, result, False)


def check_policy_name(name, policy_id, result):
    _check_generic_name("policies", name, policy_id, result)


def check_subject_name(name, subject_id, result):
    _check_generic_name("subjects", name, subject_id, result)


def check_object_name(name, object_id, result):
    _check_generic_name("objects", name, object_id, result)


def check_action_name(name, action_id, result):
    _check_generic_name("actions", name, action_id, result)


def check_scat_id_in_dict(scat_id, in_dict):
    if scat_id not in in_dict:
        raise MoonCliException("Unexpected request result. Subject category not in result")


def check_ocat_id_in_dict(ocat_id, in_dict):
    if ocat_id not in in_dict:
        raise MoonCliException("Unexpected request result. Object category not in result")


def check_acat_id_in_dict(acat_id, in_dict):
    if acat_id not in in_dict:
        raise MoonCliException("Unexpected request result. Action category not in result")


def check_policy_id_in_pipeline(policy_id, pipeline):
    if policy_id not in pipeline:
        raise MoonCliException(
            "Unexpected request result. The policy id {} shall be in the pipeline".format(
                policy_id))


def _check_generic_policy_in_dict(field, policy_id, in_dict):
    if type(field) is str:
        if policy_id is not None:
            if "policy_list" not in in_dict:
                raise MoonCliException(
                    "Unexpected request result. The policy list of the {} shall not be empty".format(
                        field))
            if policy_id not in in_dict["policy_list"]:
                raise MoonCliException(
                    "Unexpected request result. The policy with id {} shall be in the {}".format(
                        policy_id, field))


def check_subject_policy(policy_id, in_dict):
    _check_generic_policy_in_dict("subject", policy_id, in_dict)


def check_object_policy(policy_id, in_dict):
    _check_generic_policy_in_dict("object", policy_id, in_dict)


def check_action_policy(policy_id, in_dict):
    _check_generic_policy_in_dict("action", policy_id, in_dict)


def _check_generic_elt_id(field1, field1_id, field2, field2_id, result):
    if type(field1) is str and type(field2) is str:
        if result[field1] is None:
            raise MoonCliException(
                "Unexpected request result: {} shall not be empty".format(field1))
        if field1_id not in result[field1]:
            raise MoonCliException("Unexpected request result. Unknown {} with id".format(field1))
        if field2 not in result[field1][field1_id]:
            raise MoonCliException(
                "Unexpected request result. {} element with id {} has no {} field".format(field1,
                                                                                          field1_id,
                                                                                          field2))
        if field2_id != result[field1][field1_id][field2]:
            raise MoonCliException(
                "Unexpected request result. {} element with id {} has a bad {} id. Expected {}".format(
                    field1, field1_id, field2, field2_id))


def check_policy_model_id(model_id, policy_id, result):
    _check_generic_elt_id("policies", policy_id, "model_id", model_id, result)


def check_pdp_project_id(project_id, pdp_id, result):
    _check_generic_elt_id("pdps", pdp_id, "keystone_project_id", project_id, result)


def check_subject_description(description, in_dict):
    if description is not None:
        if "description" not in in_dict:
            raise MoonCliException(
                "Unexpected request result. The description of the subject shall not be empty")
        if description not in in_dict["description"]:
            raise MoonCliException(
                "Unexpected request result. The description {} shall be in the subject".format(
                    description))


def check_meta_rules_list_in_model(meta_rule_list, model_id, result):
    if result["models"] is None:
        raise MoonCliException("Unexpected request result. results shall not be empty")
    if model_id not in result['models']:
        raise MoonCliException("Unexpected request result. Unknown Model id")
    if "meta_rules" not in result['models'][model_id]:
        raise MoonCliException(
            "Unexpected request result. Meta rules related to model with id {} are empty".format(
                model_id))
    if meta_rule_list != result['models'][model_id]["meta_rules"]:
        raise MoonCliException(
            "Unexpected request result. Meta rule of model with id {} are different from those expected".format(
                model_id))


def check_name_in_slaves(name, slaves):
    if name is None:
        raise MoonCliException("The slave name must be provided !")
    names = map(lambda x: x['name'], slaves)
    if name not in names:
        raise MoonCliException("The slave '{}' was not found !".format(name))


def _check_generic_data_data(field, result):
    if type(field) is str:
        if field not in result:
            raise MoonCliException(
                "Unexpected request result. The {} field shall be in result".format(field))
        # if "data" not in resulti[field]:
        #    raise MoonCliException("Unexpected request result. The data field shall be in result['{}']".format(field))


def _check_id_in_generic_data_data(field, data_id, result):
    if type(field) is str:
        _check_generic_data_data(field, result)
        for _data in result[field]:
            if data_id not in list(_data['data'].keys()):
                raise MoonCliException(
                    "Unexpected request result. Data id {} not in {}".format(data_id, field))


def _check_id_not_in_generic_data_data(field, data_id, result):
    if type(field) is str:
        _check_generic_data_data(field, result)
        for _data in result[field]:
            if data_id in list(_data['data'].keys()):
                raise MoonCliException(
                    "Unexpected request result. Data id {} shall not be in {}".format(data_id,
                                                                                      field))


def _check_category_in_generic_data_data(field, category_id, result):
    _check_generic_data_data(field, result)
    for _data in result[field]:
        if category_id != _data["category_id"]:
            raise MoonCliException(
                "Unexpected request result. Category id {} not in {} data".format(category_id,
                                                                                  field))


def check_subject_data_data(result):
    _check_generic_data_data("subject_data", result)


def check_id_in_subject_data_data(data_id, result):
    _check_id_in_generic_data_data("subject_data", data_id, result)


def check_id_not_in_subject_data_data(data_id, result):
    _check_id_not_in_generic_data_data("subject_data", data_id, result)


def check_category_id_in_subject_data_data(category_id, result):
    _check_category_in_generic_data_data('subject_data', category_id, result)


def check_object_data_data(result):
    _check_generic_data_data("object_data", result)


def check_id_in_object_data_data(data_id, result):
    _check_id_in_generic_data_data("object_data", data_id, result)


def check_id_not_in_object_data_data(data_id, result):
    _check_id_not_in_generic_data_data("object_data", data_id, result)


def check_category_id_in_object_data_data(category_id, result):
    _check_category_in_generic_data_data('object_data', category_id, result)


def check_action_data_data(result):
    _check_generic_data_data("action_data", result)


def check_id_in_action_data_data(data_id, result):
    _check_id_in_generic_data_data("action_data", data_id, result)


def check_id_not_in_action_data_data(data_id, result):
    _check_id_not_in_generic_data_data("action_data", data_id, result)


def check_category_id_in_action_data_data(category_id, result):
    _check_category_in_generic_data_data('action_data', category_id, result)


def _check_generic_assignments(field, field_id_name, field_id, field_cat_id, field_data_id, result):
    if type(field) is str and type(field_id_name) is str:
        for key in result[field]:
            if field_id_name not in result[field][key]:
                raise MoonCliException(
                    "Unexpected request result. subject_id not in result[{}] data".format(field))
            if "category_id" not in result[field][key]:
                raise MoonCliException(
                    "Unexpected request result. category_id not in result[{}] data".format(field))
            if "assignments" not in result[field][key]:
                raise MoonCliException(
                    "Unexpected request result. assignments not in result[{}] data".format(field))
            if result[field][key][field_id_name] == field_id and \
                    result[field][key]["category_id"] == field_cat_id:
                if field_data_id not in result[field][key]["assignments"]:
                    raise MoonCliException(
                        "Unexpected request result. {} data with id {} not in result[{}][]['assignements'] data".format(
                            field, field_data_id, field))


def check_subject_assignements(subject_id, subject_act_id, subject_data_id, result):
    _check_generic_assignments("subject_assignments", "subject_id", subject_id, subject_act_id,
                               subject_data_id, result)


def check_object_assignements(object_id, object_act_id, object_data_id, result):
    _check_generic_assignments("object_assignments", "object_id", object_id, object_act_id,
                               object_data_id, result)


def check_action_assignements(action_id, action_act_id, action_data_id, result):
    _check_generic_assignments("action_assignments", "action_id", action_id, action_act_id,
                               action_data_id, result)


def _check_not_generic_assignments(field, field_id_name, field_id, field_cat_id, field_data_id,
                                   result):
    if type(field) is str and type(field_id_name) is str:
        for key in result[field]:
            if field_id_name not in result[field][key]:
                raise MoonCliException(
                    "Unexpected request result. subject_id not in result[{}] data".format(field))
            if "category_id" not in result[field][key]:
                raise MoonCliException(
                    "Unexpected request result. category_id not in result[{}] data".format(field))
            if "assignments" not in result[field][key]:
                raise MoonCliException(
                    "Unexpected request result. assignments not in result[{}] data".format(field))
            if result[field][key]['subject_id'] == field_id and \
                    result[field][key]["category_id"] == field_cat_id:
                if field_data_id in result[field][key]["assignments"]:
                    raise MoonCliException(
                        "Unexpected request result. {} data with id {} shall not be in result[{}][]['assignements'] data".format(
                            field, field_data_id, field))


def check_not_subject_assignements(subject_id, subject_act_id, subject_data_id, result):
    _check_not_generic_assignments("subject_assignments", "subject_id", subject_id, subject_act_id,
                                   subject_data_id, result)


def check_not_object_assignements(object_id, object_act_id, object_data_id, result):
    _check_not_generic_assignments("object_assignments", "object_id", object_id, object_act_id,
                                   object_data_id, result)


def check_not_action_assignements(action_id, action_act_id, action_data_id, result):
    _check_not_generic_assignments("action_assignments", "action_id", action_id, action_act_id,
                                   action_data_id, result)


def check_policy_id_in_dict(policy_id, in_dict):
    if "policy_id" not in in_dict:
        raise MoonCliException("Unexpected request result. policy_id not in result")
    if policy_id != in_dict["policy_id"]:
        raise MoonCliException(
            "Unexpected request result. Bad policy id in result, expected {}".format(policy_id))


def check_meta_rule_id_in_dict(meta_rule_id, in_dict):
    if "meta_rule_id" not in in_dict:
        raise MoonCliException("Unexpected request result. meta_rule_id not in result")
    if meta_rule_id != in_dict["meta_rule_id"]:
        raise MoonCliException(
            "Unexpected request result. Bad meta rule id in result, expected {}".format(
                meta_rule_id))


def check_rule_in_dict(rule, in_dict):
    if "rule" not in in_dict:
        raise MoonCliException("Unexpected request result. rule not in result")
    if rule != in_dict["rule"]:
        raise MoonCliException(
            "Unexpected request result. Bad rule in result, expected {}".format(rule))


def check_rule_id_in_list(meta_rule_id, rule_id, rule, in_dict):
    for item in in_dict:
        if "meta_rule_id" not in item:
            raise MoonCliException("Unexpected request result. meta_rule_id field not in result")
        if meta_rule_id == item["meta_rule_id"]:
            if rule_id == item["id"]:
                if rule != item["rule"]:
                    raise MoonCliException(
                        "Unexpected request result. Bad rule in result, expected {}".format(rule))


def check_rule_id_not_in_list(rule_id, in_dict):
    found_rule = False
    for item in in_dict:
        if rule_id == item["id"]:
            found_rule = True
    if found_rule is True:
        raise MoonCliException(
            "Unexpected request result. Rule with id {} shall not be in result".format(rule_id))