import logging
from python_moonutilities.security_functions import check_auth
from python_moondb.core import PolicyManager
+from python_moonutilities.security_functions import validate_input
__version__ = "4.3.2"
"/policies/<string:uuid>/subject_assignments/<string:perimeter_id>/<string:category_id>/<string:data_id>",
)
+ @validate_input("get", kwargs_state=[True, False, False,False,False])
@check_auth
def get(self, uuid=None, perimeter_id=None, category_id=None,
data_id=None, user_id=None):
"error": str(e)}, 500
return {"subject_assignments": data}
+ @validate_input("post", kwargs_state=[True, False, False, False, False], body_state=[True, True, True])
@check_auth
def post(self, uuid=None, perimeter_id=None, category_id=None,
data_id=None, user_id=None):
"error": str(e)}, 500
return {"subject_assignments": data}
+ @validate_input("delete", kwargs_state=[True, True, True, True, False])
@check_auth
def delete(self, uuid=None, perimeter_id=None, category_id=None,
data_id=None, user_id=None):
"/policies/<string:uuid>/object_assignments/<string:perimeter_id>/<string:category_id>/<string:data_id>",
)
+ @validate_input("get", kwargs_state=[True, False, False,False,False])
@check_auth
def get(self, uuid=None, perimeter_id=None, category_id=None,
data_id=None, user_id=None):
"error": str(e)}, 500
return {"object_assignments": data}
+ @validate_input("post", kwargs_state=[True, False, False, False, False], body_state=[True, True, True])
@check_auth
def post(self, uuid=None, perimeter_id=None, category_id=None,
data_id=None, user_id=None):
"error": str(e)}, 500
return {"object_assignments": data}
+ @validate_input("delete", kwargs_state=[True, True, True, True, False])
@check_auth
def delete(self, uuid=None, perimeter_id=None, category_id=None,
data_id=None, user_id=None):
"/policies/<string:uuid>/action_assignments/<string:perimeter_id>/<string:category_id>/<string:data_id>",
)
+ @validate_input("get", kwargs_state=[True, False, False,False,False])
@check_auth
def get(self, uuid=None, perimeter_id=None, category_id=None,
data_id=None, user_id=None):
"error": str(e)}, 500
return {"action_assignments": data}
+ @validate_input("post", kwargs_state=[True, False, False, False, False], body_state=[True, True, True])
@check_auth
def post(self, uuid=None, perimeter_id=None, category_id=None,
data_id=None, user_id=None):
"error": str(e)}, 500
return {"action_assignments": data}
+ @validate_input("delete", kwargs_state=[True, True, True, True, False])
@check_auth
def delete(self, uuid=None, perimeter_id=None, category_id=None,
data_id=None, user_id=None):
import logging
from python_moonutilities.security_functions import check_auth
from python_moondb.core import PolicyManager
+from python_moonutilities.security_functions import validate_input
__version__ = "4.3.2"
"<string:data_id>",
)
+ @validate_input("get", kwargs_state=[True, False, False, False])
@check_auth
def get(self, uuid=None, category_id=None, data_id=None, user_id=None):
"""Retrieve all subject categories or a specific one if sid is given
"error": str(e)}, 500
return {"subject_data": data}
+ @validate_input("post", kwargs_state=[True, True, False, False], body_state=[True, False])
@check_auth
def post(self, uuid=None, category_id=None, data_id=None, user_id=None):
"""Create or update a subject.
"error": str(e)}, 500
return {"subject_data": data}
+ @validate_input("delete", kwargs_state=[True, False, False, False])
@check_auth
def delete(self, uuid=None, category_id=None, data_id=None, user_id=None):
"""Delete a subject for a given policy
"<string:data_id>",
)
+ @validate_input("get", kwargs_state=[True, False, False, False])
@check_auth
def get(self, uuid=None, category_id=None, data_id=None, user_id=None):
"""Retrieve all object categories or a specific one if sid is given
"error": str(e)}, 500
return {"object_data": data}
+ @validate_input("post", kwargs_state=[True, True, False, False], body_state=[True, False])
@check_auth
def post(self, uuid=None, category_id=None, data_id=None, user_id=None):
"""Create or update a object.
"error": str(e)}, 500
return {"object_data": data}
+ @validate_input("delete", kwargs_state=[True, False, False, False])
@check_auth
def delete(self, uuid=None, category_id=None, data_id=None, user_id=None):
"""Delete a object for a given policy
"<string:data_id>",
)
+ @validate_input("get", kwargs_state=[True, False, False, False])
@check_auth
def get(self, uuid=None, category_id=None, data_id=None, user_id=None):
"""Retrieve all action categories or a specific one if sid is given
"error": str(e)}, 500
return {"action_data": data}
+ @validate_input("post", kwargs_state=[True, True, False, False], body_state=[True, False])
@check_auth
def post(self, uuid=None, category_id=None, data_id=None, user_id=None):
"""Create or update a action.
"error": str(e)}, 500
return {"action_data": data}
+ @validate_input("delete", kwargs_state=[True, False, False, False])
@check_auth
def delete(self, uuid=None, category_id=None, data_id=None, user_id=None):
"""Delete a action for a given policy
import logging
from python_moonutilities.security_functions import check_auth
from python_moondb.core import ModelManager
+from python_moonutilities.security_functions import validate_input
__version__ = "4.3.2"
"/subject_categories/<string:category_id>",
)
+ @validate_input("get",kwargs_state=[False,False])
@check_auth
def get(self, category_id=None, user_id=None):
"""Retrieve all subject categories or a specific one
"error": str(e)}, 500
return {"subject_categories": data}
+ @validate_input("post",body_state=[True,False])
@check_auth
def post(self, category_id=None, user_id=None):
"""Create or update a subject category.
"error": str(e)}, 500
return {"subject_categories": data}
+ @validate_input("delete",kwargs_state=[True,False])
@check_auth
def delete(self, category_id=None, user_id=None):
"""Delete a subject category
"/object_categories/<string:category_id>",
)
+ @validate_input("get",kwargs_state=[False,False])
@check_auth
def get(self, category_id=None, user_id=None):
"""Retrieve all object categories or a specific one
"error": str(e)}, 500
return {"object_categories": data}
+ @validate_input("post", body_state=[True, False])
@check_auth
def post(self, category_id=None, user_id=None):
"""Create or update a object category.
"error": str(e)}, 500
return {"object_categories": data}
+ @validate_input("delete", kwargs_state=[True, False])
@check_auth
def delete(self, category_id=None, user_id=None):
"""Delete an object category
"/action_categories/<string:category_id>",
)
+ @validate_input("get", kwargs_state=[False, False])
@check_auth
def get(self, category_id=None, user_id=None):
"""Retrieve all action categories or a specific one
"error": str(e)}, 500
return {"action_categories": data}
+ @validate_input("post", body_state=[True, False])
@check_auth
def post(self, category_id=None, user_id=None):
"""Create or update an action category.
"error": str(e)}, 500
return {"action_categories": data}
+ @validate_input("delete", kwargs_state=[True, False])
@check_auth
def delete(self, category_id=None, user_id=None):
"""Delete an action
import logging
from python_moonutilities.security_functions import check_auth
from python_moondb.core import ModelManager
+from python_moonutilities.security_functions import validate_input
__version__ = "4.3.2"
"/meta_rules/<string:meta_rule_id>/"
)
+ @validate_input("get", kwargs_state=[False, False])
@check_auth
def get(self, meta_rule_id=None, user_id=None):
"""Retrieve all sub meta rules
"error": str(e)}, 500
return {"meta_rules": data}
+ @validate_input("post", body_state=[True, True, True, True])
@check_auth
def post(self, meta_rule_id=None, user_id=None):
"""Add a meta rule
"error": str(e)}, 500
return {"meta_rules": data}
+ @validate_input("patch", kwargs_state=[True, False], body_state=[True, True, True, True])
@check_auth
def patch(self, meta_rule_id=None, user_id=None):
"""Update a meta rule
"error": str(e)}, 500
return {"meta_rules": data}
+ @validate_input("delete", kwargs_state=[True, False])
@check_auth
def delete(self, meta_rule_id=None, user_id=None):
"""Delete a meta rule
import logging
from python_moonutilities.security_functions import check_auth
from python_moondb.core import ModelManager
+from python_moonutilities.security_functions import validate_input
__version__ = "4.3.2"
"/models/<string:uuid>/",
)
+ @validate_input("get", kwargs_state=[False, False])
@check_auth
def get(self, uuid=None, user_id=None):
"""Retrieve all models
"error": str(e)}, 500
return {"models": data}
+ @validate_input("post", body_state=[True, False, True])
@check_auth
def post(self, uuid=None, user_id=None):
"""Create model.
"error": str(e)}, 500
return {"models": data}
+ @validate_input("delete", kwargs_state=[True, False])
@check_auth
def delete(self, uuid=None, user_id=None):
"""Delete a model
"error": str(e)}, 500
return {"result": True}
+ @validate_input("patch", kwargs_state=[True, False], body_state=[True, False, True])
@check_auth
def patch(self, uuid=None, user_id=None):
"""Update a model
from python_moondb.core import PolicyManager
from python_moondb.core import ModelManager
from python_moonutilities import configuration, exceptions
+from python_moonutilities.security_functions import validate_input
__version__ = "4.3.2"
"/pdp/<string:uuid>/",
)
+ @validate_input("get", kwargs_state=[False, False])
@check_auth
def get(self, uuid=None, user_id=None):
"""Retrieve all pdp
"error": str(e)}, 500
return {"pdps": data}
+ @validate_input("post", body_state=[True, True, True, False])
@check_auth
def post(self, uuid=None, user_id=None):
"""Create pdp.
"error": str(e)}, 500
return {"pdps": data}
+ @validate_input("delete", kwargs_state=[True, False])
@check_auth
def delete(self, uuid=None, user_id=None):
"""Delete a pdp
"error": str(e)}, 500
return {"result": True}
+ @validate_input("patch", kwargs_state=[True, False], body_state=[True, True, True, False])
@check_auth
def patch(self, uuid=None, user_id=None):
"""Update a pdp
import logging
from python_moonutilities.security_functions import check_auth
from python_moondb.core import PolicyManager
+from python_moonutilities.security_functions import validate_input
+
__version__ = "4.3.2"
"/policies/<string:uuid>/subjects/<string:perimeter_id>",
)
+ @validate_input("get", kwargs_state=[False, False, False])
@check_auth
def get(self, uuid=None, perimeter_id=None, user_id=None):
"""Retrieve all subjects or a specific one if perimeter_id is
"error": str(e)}, 500
return {"subjects": data}
+ @validate_input("post", body_state=[True, False, False, False])
@check_auth
def post(self, uuid=None, perimeter_id=None, user_id=None):
"""Create or update a subject.
"error": str(e)}, 500
return {"subjects": data}
+ @validate_input("patch", kwargs_state=[False, True, False], body_state=[True, False, False, False])
@check_auth
def patch(self, uuid=None, perimeter_id=None, user_id=None):
"""Create or update a subject.
"error": str(e)}, 500
return {"subjects": data}
+ @validate_input("delete", kwargs_state=[False, True, False])
@check_auth
def delete(self, uuid=None, perimeter_id=None, user_id=None):
"""Delete a subject for a given policy
"/policies/<string:uuid>/objects/<string:perimeter_id>",
)
+ @validate_input("get", kwargs_state=[False, False, False])
@check_auth
def get(self, uuid=None, perimeter_id=None, user_id=None):
"""Retrieve all objects or a specific one if perimeter_id is
"error": str(e)}, 500
return {"objects": data}
+ @validate_input("post", body_state=[True, False, False, False])
@check_auth
def post(self, uuid=None, perimeter_id=None, user_id=None):
"""Create or update a object.
"error": str(e)}, 500
return {"objects": data}
+ @validate_input("patch", kwargs_state=[False, True, False], body_state=[True, False, False, False])
@check_auth
def patch(self, uuid=None, perimeter_id=None, user_id=None):
"""Create or update a object.
"error": str(e)}, 500
return {"objects": data}
+ @validate_input("delete", kwargs_state=[False, True, False])
@check_auth
def delete(self, uuid=None, perimeter_id=None, user_id=None):
"""Delete a object for a given policy
"/policies/<string:uuid>/actions/<string:perimeter_id>",
)
+ @validate_input("get", kwargs_state=[False, False, False])
@check_auth
def get(self, uuid=None, perimeter_id=None, user_id=None):
"""Retrieve all actions or a specific one if perimeter_id
"error": str(e)}, 500
return {"actions": data}
+ @validate_input("post", body_state=[True, False, False, False])
@check_auth
def post(self, uuid=None, perimeter_id=None, user_id=None):
"""Create or update a action.
"error": str(e)}, 500
return {"actions": data}
+ @validate_input("patch", kwargs_state=[False, True, False], body_state=[True, False, False, False])
@check_auth
def patch(self, uuid=None, perimeter_id=None, user_id=None):
"""Create or update a action.
"error": str(e)}, 500
return {"actions": data}
+ @validate_input("delete", kwargs_state=[False, True, False])
@check_auth
def delete(self, uuid=None, perimeter_id=None, user_id=None):
"""Delete a action for a given policy
import logging
from python_moonutilities.security_functions import check_auth
from python_moondb.core import PolicyManager
+from python_moonutilities.security_functions import validate_input
__version__ = "4.3.2"
"/policies/<string:uuid>/rules/<string:rule_id>/",
)
+ @validate_input("get", kwargs_state=[False, False, False])
@check_auth
def get(self, uuid=None, rule_id=None, user_id=None):
"""Retrieve all rules or a specific one
"error": str(e)}, 500
return {"rules": data}
+ @validate_input("post", kwargs_state=[True, False, False], body_state=[True, False, False, False])
@check_auth
def post(self, uuid=None, rule_id=None, user_id=None):
"""Add a rule to a meta rule
"error": str(e)}, 500
return {"rules": data}
+ @validate_input("delete", kwargs_state=[True, True, False])
@check_auth
def delete(self, uuid=None, rule_id=None, user_id=None):
"""Delete one rule linked to a specific sub meta rule
def clean_subjects(client):
subjects = test_perimeter.get_subjects(client)
logger.info("subjects {}".format(subjects))
- for key in subjects["subjects"]:
- subject = subjects["subjects"][key]
+ for key in subjects[1]["subjects"]:
+ subject = subjects[1]["subjects"][key]
policy_keys = subject["policy_list"]
logger.info("subjects policy_keys {}".format(policy_keys))
for policy_key in policy_keys:
def clean_objects(client):
objects = test_perimeter.get_objects(client)
logger.info("objects {}".format(objects))
- for key in objects["objects"]:
- object_ = objects["objects"][key]
+ for key in objects[1]["objects"]:
+ object_ = objects[1]["objects"][key]
policy_keys = object_["policy_list"]
logger.info("objects policy_keys {}".format(policy_keys))
for policy_key in policy_keys:
def clean_actions(client):
actions = test_perimeter.get_actions(client)
logger.info("objects {}".format(actions))
- for key in actions["actions"]:
- action = actions["actions"][key]
+ for key in actions[1]["actions"]:
+ action = actions[1]["actions"][key]
policy_keys = action["policy_list"]
logger.info("action policy_keys {}".format(policy_keys))
for policy_key in policy_keys:
assert value['description'] == "description of {}".format("testuser")
+def test_add_subject_categories_with_empty_user():
+ client = utilities.register_client()
+ req, subject_categories = add_subject_categories(client, "")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "Empty String"
+
+
+def test_add_subject_categories_with_user_contain_space():
+ client = utilities.register_client()
+ req, subject_categories = add_subject_categories(client, "test user")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "String contains space"
+
+
def test_delete_subject_categories():
client = utilities.register_client()
req = delete_subject_categories(client, "testuser")
assert value['description'] == "description of {}".format("testuser")
+def test_add_object_categories_with_empty_user():
+ client = utilities.register_client()
+ req, object_categories = add_object_categories(client, "")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "Empty String"
+
+
+def test_add_object_categories_with_user_contain_space():
+ client = utilities.register_client()
+ req, object_categories = add_object_categories(client, "test user")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "String contains space"
+
+
def test_delete_object_categories():
client = utilities.register_client()
req = delete_object_categories(client, "testuser")
assert value['description'] == "description of {}".format("testuser")
+def test_add_action_categories_with_empty_user():
+ client = utilities.register_client()
+ req, action_categories = add_action_categories(client, "")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "Empty String"
+
+
+def test_add_action_categories_with_user_contain_space():
+ client = utilities.register_client()
+ req, action_categories = add_action_categories(client, "test user")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "String contains space"
+
+
def test_delete_action_categories():
client = utilities.register_client()
req = delete_action_categories(client, "testuser")
return req, meta_rules
+def add_meta_rules_without_subject_category_ids(client, name):
+ data = {
+ "name": name,
+ "subject_categories": [],
+ "object_categories": ["object_category_id1"],
+ "action_categories": ["action_category_id1"]
+ }
+ req = client.post("/meta_rules", data=json.dumps(data),
+ headers={'Content-Type': 'application/json'})
+ meta_rules = utilities.get_json(req.data)
+ return req, meta_rules
+
+
+def update_meta_rules(client, name, metaRuleId):
+ data = {
+ "name": name,
+ "subject_categories": ["subject_category_id1_update",
+ "subject_category_id2_update"],
+ "object_categories": ["object_category_id1_update"],
+ "action_categories": ["action_category_id1_update"]
+ }
+ req = client.patch("/meta_rules/{}".format(metaRuleId), data=json.dumps(data),
+ headers={'Content-Type': 'application/json'})
+ meta_rules = utilities.get_json(req.data)
+ return req, meta_rules
+
+
+def update_meta_rules_without_subject_category_ids(client, name):
+ data = {
+ "name": name,
+ "subject_categories": [],
+ "object_categories": ["object_category_id1"],
+ "action_categories": ["action_category_id1"]
+ }
+ req = client.post("/meta_rules", data=json.dumps(data),
+ headers={'Content-Type': 'application/json'})
+ meta_rules = utilities.get_json(req.data)
+ return req, meta_rules
+
+
def delete_meta_rules(client, name):
request, meta_rules = get_meta_rules(client)
for key, value in meta_rules['meta_rules'].items():
assert value["action_categories"][0] == "action_category_id1"
+def test_add_meta_rules_with_empty_user():
+ client = utilities.register_client()
+ req, meta_rules = add_meta_rules(client, "")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "Empty String"
+
+
+def test_add_meta_rules_with_user_contain_space():
+ client = utilities.register_client()
+ req, meta_rules = add_meta_rules(client, "test user")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "String contains space"
+
+
+def test_add_meta_rules_without_subject_categories():
+ client = utilities.register_client()
+ req, meta_rules = add_meta_rules_without_subject_category_ids(client, "testuser")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == 'Empty Container'
+
+
def test_delete_meta_rules():
client = utilities.register_client()
req = delete_meta_rules(client, "testuser")
client = utilities.register_client()
req = delete_meta_rules_without_id(client)
assert req.status_code == 500
+
+
+def test_update_meta_rules():
+ client = utilities.register_client()
+ req = add_meta_rules(client, "testuser")
+ meta_rule_id = list(req[1]['meta_rules'])[0]
+ req_update = update_meta_rules(client, "testuser", meta_rule_id)
+ assert req_update[0].status_code == 200
+ value = list(req_update[1]["meta_rules"].values())[0]
+ assert value["subject_categories"][0] == "subject_category_id1_update"
+ delete_meta_rules(client, "testuser")
+ get_meta_rules(client)
+
+
+def test_update_meta_rules_without_id():
+ client = utilities.register_client()
+ req_update = update_meta_rules(client, "testuser", "")
+ assert req_update[0].status_code == 500
+
+
+def test_update_meta_rules_without_user():
+ client = utilities.register_client()
+ req_update = update_meta_rules(client, "", "")
+ assert req_update[0].status_code == 500
+ assert json.loads(req_update[0].data)["message"] == "Empty String"
+
+
+def test_update_meta_rules_without_subject_categories():
+ client = utilities.register_client()
+ req_update = update_meta_rules_without_subject_category_ids(client, "testuser")
+ assert req_update[0].status_code == 500
+ assert json.loads(req_update[0].data)["message"] == "Empty Container"
"category_id": category_id,
"data_id": "data_id1"
}
- req = client.post("/policies/{}/subject_assignments/{}".format(policy_id, category_id), data=json.dumps(data),
+ req = client.post("/policies/{}/subject_assignments".format(policy_id), data=json.dumps(data),
headers={'Content-Type': 'application/json'})
subject_assignment = utilities.get_json(req.data)
return req, subject_assignment
-def delete_subject_assignment(client, policy_id):
- req = client.delete("/policies/{}/subject_assignments".format(policy_id))
+def delete_subject_assignment(client, policy_id, sub_id, cat_id,data_id):
+ req = client.delete("/policies/{}/subject_assignments/{}/{}/{}".format(policy_id, sub_id, cat_id,data_id))
return req
assert value[id]['subject_id'] == "id1"
+def test_add_subject_assignment_without_cat_id():
+ policy_id = utilities.get_policy_id()
+ client = utilities.register_client()
+ req, subject_assignment = add_subject_assignment(client, policy_id, "")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == 'Empty String'
+
+
def test_get_subject_assignment():
policy_id = utilities.get_policy_id()
client = utilities.register_client()
def test_delete_subject_assignment():
client = utilities.register_client()
policy_id = utilities.get_policy_id()
- success_req = delete_subject_assignment(client, policy_id)
+ req, subject_assignment = get_subject_assignment(client, policy_id)
+ value = subject_assignment["subject_assignments"]
+ id = list(value.keys())[0]
+ success_req = delete_subject_assignment(client, policy_id, value[id]['subject_id'], value[id]['category_id'],value[id]['assignments'][0])
assert success_req.status_code == 200
+
+def test_delete_subject_assignment_without_policy_id():
+ client = utilities.register_client()
+ success_req = delete_subject_assignment(client, "", "id1", "111" ,"data_id1")
+ assert success_req.status_code == 500
+
+
# ---------------------------------------------------------------------------
# object_categories_test
return req, object_assignment
-def delete_object_assignment(client, policy_id):
- req = client.delete("/policies/{}/object_assignments".format(policy_id))
+def delete_object_assignment(client, policy_id, obj_id, cat_id, data_id):
+ req = client.delete("/policies/{}/object_assignments/{}/{}/{}".format(policy_id, obj_id, cat_id, data_id))
return req
assert value[id]['object_id'] == "id1"
+def test_add_object_assignment_without_cat_id():
+ policy_id = utilities.get_policy_id()
+ client = utilities.register_client()
+ req, object_assignment = add_object_assignment(client, policy_id, "")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == 'Empty String'
+
+
def test_delete_object_assignment():
client = utilities.register_client()
policy_id = utilities.get_policy_id()
- success_req = delete_object_assignment(client, policy_id)
+ req, object_assignment = get_object_assignment(client, policy_id)
+ value = object_assignment["object_assignments"]
+ id = list(value.keys())[0]
+ success_req = delete_object_assignment(client, policy_id, value[id]['object_id'], value[id]['category_id'],value[id]['assignments'][0])
assert success_req.status_code == 200
+
+def test_delete_object_assignment_without_policy_id():
+ client = utilities.register_client()
+ success_req = delete_object_assignment(client, "", "id1", "111" ,"data_id1")
+ assert success_req.status_code == 500
+
+
# ---------------------------------------------------------------------------
# action_categories_test
return req, action_assignment
-def delete_action_assignment(client, policy_id):
- req = client.delete("/policies/{}/action_assignments".format(policy_id))
+def delete_action_assignment(client, policy_id, action_id, cat_id, data_id):
+ req = client.delete("/policies/{}/action_assignments/{}/{}/{}".format(policy_id, action_id, cat_id, data_id))
return req
assert value[id]['action_id'] == "id1"
+def test_add_action_assignment_without_cat_id():
+ policy_id = utilities.get_policy_id()
+ client = utilities.register_client()
+ req, action_assignment = add_action_assignment(client, policy_id, "")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == 'Empty String'
+
+
def test_delete_action_assignment():
client = utilities.register_client()
policy_id = utilities.get_policy_id()
- success_req = delete_action_assignment(client, policy_id)
+ req, action_assignment = get_action_assignment(client, policy_id)
+ value = action_assignment["action_assignments"]
+ id = list(value.keys())[0]
+ success_req = delete_action_assignment(client, policy_id, value[id]['action_id'], value[id]['category_id'],value[id]['assignments'][0])
assert success_req.status_code == 200
-# ---------------------------------------------------------------------------
\ No newline at end of file
+
+def test_delete_action_assignment_without_policy_id():
+ client = utilities.register_client()
+ success_req = delete_action_assignment(client, "", "id1", "111" ,"data_id1")
+ assert success_req.status_code == 500
+
+# ---------------------------------------------------------------------------
success_req = delete_subject_data(client, policy_id)
assert success_req.status_code == 200
+
+def test_add_subject_data_with_empty_user():
+ policy_id = utilities.get_policy_id()
+ client = utilities.register_client()
+ req, subject_data = add_subject_data(client, "", policy_id, "111")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "Empty String"
+
+
+def test_add_subject_data_with_user_contain_space():
+ policy_id = utilities.get_policy_id()
+ client = utilities.register_client()
+ req, subject_data = add_subject_data(client, "test user", policy_id, "111")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "String contains space"
+
+
+def test_delete_subject_data_without_policy_id():
+ client = utilities.register_client()
+ success_req = delete_subject_data(client, "")
+ assert success_req.status_code == 500
+
# ---------------------------------------------------------------------------
# object_categories_test
success_req = delete_object_data(client, policy_id)
assert success_req.status_code == 200
+
+def test_add_object_data_with_empty_user():
+ policy_id = utilities.get_policy_id()
+ client = utilities.register_client()
+ req, subject_data = add_subject_data(client, "", policy_id, "111")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "Empty String"
+
+
+def test_add_object_data_with_user_contain_space():
+ policy_id = utilities.get_policy_id()
+ client = utilities.register_client()
+ req, object_data = add_object_data(client, "test user", policy_id, "111")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "String contains space"
+
+
+def test_delete_object_data_without_policy_id():
+ client = utilities.register_client()
+ success_req = delete_object_data(client, "")
+ assert success_req.status_code == 500
# ---------------------------------------------------------------------------
# action_categories_test
success_req = delete_action_data(client, policy_id)
assert success_req.status_code == 200
+
+def test_add_action_data_with_empty_user():
+ policy_id = utilities.get_policy_id()
+ client = utilities.register_client()
+ req, action_data = add_action_data(client, "", policy_id, "111")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "Empty String"
+
+
+def test_add_action_data_with_user_contain_space():
+ policy_id = utilities.get_policy_id()
+ client = utilities.register_client()
+ req, action_data = add_action_data(client, "test user", policy_id, "111")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "String contains space"
+
+
+def test_delete_action_data_without_policy_id():
+ client = utilities.register_client()
+ success_req = delete_action_data(client, "")
+ assert success_req.status_code == 500
# ---------------------------------------------------------------------------
{"rules": [{"meta_rule": {"name": "good meta rule"}, "policy": {"name": "test policy"}, "instructions": {"decision": "grant"}, "enabled": True, "rule": {"subject_data": [{"name": "subject data"}], "object_data": [{"name": "object data"}], "action_data": [{"name": "action data"}]}}]}]
-
-
def test_import_models_without_new_meta_rules():
client = utilities.register_client()
import_export_utilities.clean_all(client)
assert data == "Import ok !"
get_elements = get_method(client)
- get_elements = get_elements[type_element + "s"]
+ get_elements = get_elements[1][type_element + "s"]
assert len(list(get_elements.keys())) == 1
values = list(get_elements.values())
return req, pdp
+def update_pdp(client, data, pdp_id):
+ req = client.patch("/pdp/{}".format(pdp_id), data=json.dumps(data),
+ headers={'Content-Type': 'application/json'})
+ pdp = utilities.get_json(req.data)
+ return req, pdp
+
+
def delete_pdp(client, key):
req = client.delete("/pdp/{}".format(key))
return req
success_req = delete_pdp(client, key)
break
assert success_req.status_code == 200
+
+
+def test_add_pdp_with_empty_user():
+ data = {
+ "name": "",
+ "security_pipeline": ["policy_id_1", "policy_id_2"],
+ "keystone_project_id": "keystone_project_id",
+ "description": "description of testuser"
+ }
+ client = utilities.register_client()
+ req, models = add_pdp(client, data)
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "Empty String"
+
+
+def test_add_pdp_with_user_contain_space():
+ data = {
+ "name": "test user",
+ "security_pipeline": ["policy_id_1", "policy_id_2"],
+ "keystone_project_id": "keystone_project_id",
+ "description": "description of testuser"
+ }
+ client = utilities.register_client()
+ req, models = add_pdp(client, data)
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "String contains space"
+
+
+def test_add_pdp_without_security_pipeline():
+ data = {
+ "name": "testuser",
+ "security_pipeline": [],
+ "keystone_project_id": "keystone_project_id",
+ "description": "description of testuser"
+ }
+ client = utilities.register_client()
+ req, meta_rules = add_pdp(client, data)
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == 'Empty Container'
+
+
+def test_add_pdp_without_keystone():
+ data = {
+ "name": "testuser",
+ "security_pipeline": ["policy_id_1", "policy_id_2"],
+ "keystone_project_id": "",
+ "description": "description of testuser"
+ }
+ client = utilities.register_client()
+ req, meta_rules = add_pdp(client, data)
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == 'Empty String'
+
+
+def test_update_pdp():
+ data_add = {
+ "name": "testuser",
+ "security_pipeline": ["policy_id_1", "policy_id_2"],
+ "keystone_project_id": "keystone_project_id",
+ "description": "description of testuser"
+ }
+ data_update = {
+ "name": "testuser",
+ "security_pipeline": ["policy_id_1_update", "policy_id_2_update"],
+ "keystone_project_id": "keystone_project_id_update",
+ "description": "description of testuser"
+ }
+ client = utilities.register_client()
+ req = add_pdp(client, data_add)
+ pdp_id = list(req[1]['pdps'])[0]
+ req_update = update_pdp(client, data_update, pdp_id)
+ assert req_update[0].status_code == 200
+ value = list(req_update[1]["pdps"].values())[0]
+ assert value["keystone_project_id"] == "keystone_project_id_update"
+ request, pdp = get_pdp(client)
+ for key, value in pdp['pdps'].items():
+ if value['name'] == "testuser":
+ delete_pdp(client, key)
+ break
+
+
+def test_update_pdp_without_id():
+ client = utilities.register_client()
+ req_update = update_pdp(client, "testuser", "")
+ assert req_update[0].status_code == 500
+
+
+def test_update_pdp_without_user():
+ data = {
+ "name": "",
+ "security_pipeline": ["policy_id_1", "policy_id_2"],
+ "keystone_project_id": "keystone_project_id",
+ "description": "description of testuser"
+ }
+ client = utilities.register_client()
+ req_update = update_pdp(client, data, "")
+ assert req_update[0].status_code == 500
+ assert json.loads(req_update[0].data)["message"] == "Empty String"
+
+
+def test_update_pdp_without_security_pipeline():
+ data = {
+ "name": "testuser",
+ "security_pipeline": [],
+ "keystone_project_id": "keystone_project_id",
+ "description": "description of testuser"
+ }
+ client = utilities.register_client()
+ req_update = update_pdp(client, data, "")
+ assert req_update[0].status_code == 500
+ assert json.loads(req_update[0].data)["message"] == "Empty Container"
\ No newline at end of file
def get_subjects(client):
req = client.get("/subjects")
- assert req.status_code == 200
subjects = utilities.get_json(req.data)
- assert isinstance(subjects, dict)
- assert "subjects" in subjects
- return subjects
+ return req, subjects
def add_subjects(client, name):
}
req = client.post("/subjects", data=json.dumps(data),
headers={'Content-Type': 'application/json'})
- assert req.status_code == 200
subjects = utilities.get_json(req.data)
+ return req, subjects
+
+
+def delete_subject(client):
+ subjects = get_subjects(client)
+ for key, value in subjects[1]['subjects'].items():
+ if value['name'] == "testuser":
+ req = client.delete("/subjects/{}".format(key))
+ break
+ return req
+
+
+def delete_subjects_without_perimeter_id(client):
+ req = client.delete("/subjects/{}".format(""))
+ return req
+
+
+def test_perimeter_get_subject():
+ client = utilities.register_client()
+ req, subjects = get_subjects(client)
+ assert req.status_code == 200
assert isinstance(subjects, dict)
- key = list(subjects["subjects"].keys())[0]
+ assert "subjects" in subjects
+
+
+def test_perimeter_add_subject():
+ client = utilities.register_client()
+ req, subjects = add_subjects(client, "testuser")
+ assert req.status_code == 200
value = list(subjects["subjects"].values())[0]
assert "subjects" in subjects
- assert key == "1111111111111"
- assert value['id'] == "1111111111111"
- assert value['name'] == name
- assert value["description"] == "description of {}".format(name)
- assert value["email"] == "{}@moon".format(name)
- return subjects
+ assert value['name'] == "testuser"
+ assert value["email"] == "{}@moon".format("testuser")
-def add_subjects_without_name(client, name):
- data = {
- "name": name,
- "description": "description of {}".format(name),
- "password": "password for {}".format(name),
- "email": "{}@moon".format(name)
- }
- req = client.post("/subjects", data=json.dumps(data),
- headers={'Content-Type': 'application/json'})
+def test_perimeter_add_subject_without_name():
+ client = utilities.register_client()
+ req, subjects = add_subjects(client, "")
assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "Empty String"
-def delete_subject(client, name):
- subjects = get_subjects(client)
- for key, value in subjects['subjects'].items():
- if value['name'] == name:
- req = client.delete("/subjects/{}".format(key))
- assert req.status_code == 200
- break
- subjects = get_subjects(client)
- assert name not in [x['name'] for x in subjects["subjects"].values()]
+def test_perimeter_add_subject_with_name_contain_spaces():
+ client = utilities.register_client()
+ req, subjects = add_subjects(client, "test user")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "String contains space"
-def test_subject():
+def test_perimeter_delete_subject():
client = utilities.register_client()
- get_subjects(client)
- add_subjects(client, "testuser")
- add_subjects_without_name(client, "")
- delete_subject(client, "testuser")
+ req = delete_subject(client)
+ assert req.status_code == 200
+
+
+def test_perimeter_delete_subjects_without_perimeter_id():
+ client = utilities.register_client()
+ req = delete_subjects_without_perimeter_id(client)
+ assert req.status_code == 500
def get_objects(client):
req = client.get("/objects")
- assert req.status_code == 200
objects = utilities.get_json(req.data)
- assert isinstance(objects, dict)
- assert "objects" in objects
- return objects
+ return req, objects
def add_objects(client, name):
}
req = client.post("/objects", data=json.dumps(data),
headers={'Content-Type': 'application/json'})
- assert req.status_code == 200
objects = utilities.get_json(req.data)
- assert isinstance(objects, dict)
- key = list(objects["objects"].keys())[0]
- value = list(objects["objects"].values())[0]
- assert "objects" in objects
- assert value['name'] == name
- assert value["description"] == "description of {}".format(name)
- return objects
+ return req, objects
-def delete_objects(client, name):
+def delete_object(client):
objects = get_objects(client)
- for key, value in objects['objects'].items():
- if value['name'] == name:
+ for key, value in objects[1]['objects'].items():
+ if value['name'] == "testuser":
req = client.delete("/objects/{}".format(key))
- assert req.status_code == 200
break
- objects = get_objects(client)
- assert name not in [x['name'] for x in objects["objects"].values()]
+ return req
+
+def delete_objects_without_perimeter_id(client):
+ req = client.delete("/objects/{}".format(""))
+ return req
-def test_objects():
+
+def test_perimeter_get_object():
client = utilities.register_client()
- get_objects(client)
- add_objects(client, "testuser")
- delete_objects(client, "testuser")
+ req, objects = get_objects(client)
+ assert req.status_code == 200
+ assert isinstance(objects, dict)
+ assert "objects" in objects
+
+
+def test_perimeter_add_object():
+ client = utilities.register_client()
+ req, objects = add_objects(client, "testuser")
+ assert req.status_code == 200
+ value = list(objects["objects"].values())[0]
+ assert "objects" in objects
+ assert value['name'] == "testuser"
+
+
+def test_perimeter_add_object_without_name():
+ client = utilities.register_client()
+ req, objects = add_objects(client, "")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "Empty String"
+
+
+def test_perimeter_add_object_with_name_contain_spaces():
+ client = utilities.register_client()
+ req, objects = add_objects(client, "test user")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "String contains space"
+
+
+def test_perimeter_delete_object():
+ client = utilities.register_client()
+ req = delete_object(client)
+ assert req.status_code == 200
+
+
+def test_perimeter_delete_objects_without_perimeter_id():
+ client = utilities.register_client()
+ req = delete_objects_without_perimeter_id(client)
+ assert req.status_code == 500
def get_actions(client):
req = client.get("/actions")
- assert req.status_code == 200
actions = utilities.get_json(req.data)
- assert isinstance(actions, dict)
- assert "actions" in actions
- return actions
+ return req, actions
def add_actions(client, name):
}
req = client.post("/actions", data=json.dumps(data),
headers={'Content-Type': 'application/json'})
- assert req.status_code == 200
actions = utilities.get_json(req.data)
- assert isinstance(actions, dict)
- key = list(actions["actions"].keys())[0]
- value = list(actions["actions"].values())[0]
- assert "actions" in actions
- assert value['name'] == name
- assert value["description"] == "description of {}".format(name)
- return actions
+ return req, actions
-def delete_actions(client, name):
+def delete_actions(client):
actions = get_actions(client)
- for key, value in actions['actions'].items():
- if value['name'] == name:
+ for key, value in actions[1]['actions'].items():
+ if value['name'] == "testuser":
req = client.delete("/actions/{}".format(key))
- assert req.status_code == 200
break
- actions = get_actions(client)
- assert name not in [x['name'] for x in actions["actions"].values()]
+ return req
+
+def delete_actions_without_perimeter_id(client):
+ req = client.delete("/actions/{}".format(""))
+ return req
-def test_actions():
+
+def test_perimeter_get_actions():
+ client = utilities.register_client()
+ req, actions = get_actions(client)
+ assert req.status_code == 200
+ assert isinstance(actions, dict)
+ assert "actions" in actions
+
+
+def test_perimeter_add_actions():
client = utilities.register_client()
- get_actions(client)
- add_actions(client, "testuser")
- delete_actions(client, "testuser")
+ req, actions = add_actions(client, "testuser")
+ assert req.status_code == 200
+ value = list(actions["actions"].values())[0]
+ assert "actions" in actions
+ assert value['name'] == "testuser"
+
+
+def test_perimeter_add_actions_without_name():
+ client = utilities.register_client()
+ req, actions = add_actions(client, "")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "Empty String"
+
+
+def test_perimeter_add_actions_with_name_contain_spaces():
+ client = utilities.register_client()
+ req, actions = add_actions(client, "test user")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "String contains space"
+
+
+def test_perimeter_delete_actions():
+ client = utilities.register_client()
+ req = delete_actions(client)
+ assert req.status_code == 200
+
+
+def test_perimeter_delete_actions_without_perimeter_id():
+ client = utilities.register_client()
+ req = delete_actions_without_perimeter_id(client)
+ assert req.status_code == 500
+
return req, rules
+def add_rules_without_meta_rule_id(client, policy_id):
+ data = {
+ "meta_rule_id": "",
+ "rule": ["subject_data_id2", "object_data_id2", "action_data_id2"],
+ "instructions": (
+ {"decision": "grant"},
+ ),
+ "enabled": True
+ }
+ req = client.post("/policies/{}/rules".format(policy_id), data=json.dumps(data),
+ headers={'Content-Type': 'application/json'})
+ rules = utilities.get_json(req.data)
+ return req, rules
+
+
+def add_rules_without_rule(client, policy_id):
+ data = {
+ "meta_rule_id": "meta_rule_id1",
+ "instructions": (
+ {"decision": "grant"},
+ ),
+ "enabled": True
+ }
+ req = client.post("/policies/{}/rules".format(policy_id), data=json.dumps(data),
+ headers={'Content-Type': 'application/json'})
+ rules = utilities.get_json(req.data)
+ return req, rules
+
+
def delete_rules(client, policy_id, meta_rule_id):
req = client.delete("/policies/{}/rules/{}".format(policy_id, meta_rule_id))
return req
assert value[id]["meta_rule_id"] == "meta_rule_id1"
+def test_add_rules_without_policy_id():
+ client = utilities.register_client()
+ req, rules = add_rules(client, None)
+ assert req.status_code == 500
+
+
+def test_add_rules_without_meta_rule_id():
+ policy_id = utilities.get_policy_id()
+ client = utilities.register_client()
+ req, rules = add_rules_without_meta_rule_id(client, policy_id)
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == 'Empty String'
+
+
+def test_add_rules_without_rule():
+ policy_id = utilities.get_policy_id()
+ client = utilities.register_client()
+ req, rules = add_rules_without_rule(client, policy_id)
+ assert req.status_code == 500
+
+
def test_delete_rules():
+ client = utilities.register_client()
+ rules = delete_rules(client, "", "")
+ assert rules.status_code == 500
+
+
+def test_delete_rules_without_policy_id():
client = utilities.register_client()
policy_id = utilities.get_policy_id()
req, added_rules = get_rules(client, policy_id)
id = added_rules["rules"]['rules'][0]['id']
- rules = delete_rules(client, policy_id, id)
+ rules = delete_rules(client, None, id)
assert rules.status_code == 200
return req, models
+def update_model(client, name, model_id):
+ data = {
+ "name": name,
+ "description": "description of {}".format(name),
+ "meta_rules": ["meta_rule_id1_update", "meta_rule_id2_update"]
+ }
+ req = client.patch("/models/{}".format(model_id), data=json.dumps(data),
+ headers={'Content-Type': 'application/json'})
+ models = utilities.get_json(req.data)
+ return req, models
+
+
+def add_model_without_meta_rules_ids(client, name):
+ data = {
+ "name": name,
+ "description": "description of {}".format(name),
+ "meta_rules": []
+ }
+ req = client.post("/models", data=json.dumps(data),
+ headers={'Content-Type': 'application/json'})
+ models = utilities.get_json(req.data)
+ return req, models
+
+
+def update_model_without_meta_rules_ids(client, name):
+ data = {
+ "name": name,
+ "description": "description of {}".format(name),
+ "meta_rules": []
+ }
+ req = client.patch("/models", data=json.dumps(data),
+ headers={'Content-Type': 'application/json'})
+ models = utilities.get_json(req.data)
+ return req, models
+
+
def delete_models(client, name):
request, models = get_models(client)
for key, value in models['models'].items():
print(value)
client.delete("/models/{}".format(key))
+
def test_get_models():
client = utilities.register_client()
req, models= get_models(client)
req = delete_models_without_id(client)
assert req.status_code == 500
+
+def test_add_model_with_empty_user():
+ clean_models()
+ client = utilities.register_client()
+ req, models = add_models(client, "")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "Empty String"
+
+
+def test_add_model_with_user_contain_space():
+ clean_models()
+ client = utilities.register_client()
+ req, models = add_models(client, "test user")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == "String contains space"
+
+
+def test_add_model_without_meta_rules():
+ clean_models()
+ client = utilities.register_client()
+ req, meta_rules = add_model_without_meta_rules_ids(client, "testuser")
+ assert req.status_code == 500
+ assert json.loads(req.data)["message"] == 'Empty Container'
+
+
+def test_update_model():
+ clean_models()
+ client = utilities.register_client()
+ req = add_models(client, "testuser")
+ model_id = list(req[1]['models'])[0]
+ req_update = update_model(client, "testuser", model_id)
+ assert req_update[0].status_code == 200
+ value = list(req_update[1]["models"].values())[0]
+ assert value["meta_rules"][0] == "meta_rule_id1_update"
+ delete_models(client, "testuser")
+
+
+def test_update_meta_rules_without_id():
+ clean_models()
+ client = utilities.register_client()
+ req_update = update_model(client, "testuser", "")
+ assert req_update[0].status_code == 500
+
+
+def test_update_meta_rules_without_user():
+ client = utilities.register_client()
+ req_update = update_model(client, "", "")
+ assert req_update[0].status_code == 500
+ assert json.loads(req_update[0].data)["message"] == "Empty String"
+
+
+def test_update_meta_rules_without_meta_rules():
+ client = utilities.register_client()
+ req_update = update_model_without_meta_rules_ids(client, "testuser")
+ assert req_update[0].status_code == 500
+ assert json.loads(req_update[0].data)["message"] == "Empty Container"
+
def filter_input(func_or_str):
-
def __filter(string):
if string and type(string) is str:
return "".join(re.findall("[\w\- +]*", string))
return None
+"""
+To do should check value of Dictionary but it's dependent on from where it's coming
+"""
+
+
+def validate_data(data):
+ def __validate_string(string):
+ if not string:
+ raise ValueError('Empty String')
+ '''
+ is it valid to contains space inbetween
+
+ '''
+
+ if " " in string:
+ raise ValueError('String contains space')
+
+ def __validate_list_or_tuple(container):
+ if not container:
+ raise ValueError('Empty Container')
+ for i in container:
+ validate_data(i)
+
+ def __validate_dict(dictionary):
+ if not dictionary:
+ raise ValueError('Empty Dictionary')
+ for key in dictionary:
+ validate_data(dictionary[key])
+
+ if isinstance(data, str):
+ __validate_string(data)
+ elif isinstance(data, list) or isinstance(data, tuple):
+ __validate_list_or_tuple(data)
+ elif isinstance(data, dict):
+ __validate_dict(data)
+ else:
+ raise ValueError('Value is Not String or Container or Dictionary')
+
+
+def validate_input(type, args_state=[], kwargs_state=[], body_state=[]):
+ """
+ this fucntion works only on List or tuple or dictionary of Strings ,and String direct
+ Check if input of function is Valid or not, Valid if not has spaces and values is not None or empty.
+
+ :param type: type of request if function is used as decorator
+ :param args_state: list of Booleans for args,
+ values must be order as target values of arguments,
+ True if None is not Allowed and False if is allowed
+ :param kwargs_state: list of Booleans for kwargs as order of input kwargs,
+ values must be order as target values of arguments,
+ True if None is not Allowed and False if is allowed
+ :param body_state: list of Booleans for arguments in body of request if request is post,
+ values must be order as target values of arguments,
+ True if None is not Allowed and False if is allowed
+ :return:
+ """
+
+ def validate_input_decorator(func):
+ def wrapped(*args, **kwargs):
+
+ temp_args = []
+ """
+ this loop made to filter args from object class,
+ when put this function as decorator in function control
+ then there is copy of this class add to front of args
+ """
+ for arg in args:
+ if isinstance(arg, str) == True or \
+ isinstance(arg, list) == True or \
+ isinstance(arg, dict) == True:
+ temp_args.append(arg)
+
+ while len(args_state) < len(temp_args):
+ args_state.append(True)
+
+ for i in range(0, len(temp_args)):
+ if args_state[i]:
+ validate_data(temp_args[i])
+
+ while len(kwargs_state) < len(kwargs):
+ kwargs_state.append(True)
+ counter = 0
+ for i in kwargs:
+ if kwargs_state[counter]:
+ validate_data({i: kwargs[i]})
+
+ counter = counter + 1
+
+ if type == "post" or type == "patch":
+ body = request.json
+ while len(body_state) < len(body):
+ body_state.append(True)
+ counter = 0
+ for i in body:
+ if body_state[counter]:
+ validate_data({i: body[i]})
+
+ counter = counter + 1
+
+ return func(*args, **kwargs)
+
+ return wrapped
+
+ return validate_input_decorator
+
+
def enforce(action_names, object_name, **extra):
"""Fake version of the enforce decorator"""
+
def wrapper_func(func):
def wrapper_args(*args, **kwargs):
# LOG.info("kwargs={}".format(kwargs))
# kwargs['user_id'] = kwargs.pop('user_id', "admin")
# LOG.info("Calling enforce on {} with args={} kwargs={}".format(func.__name__, args, kwargs))
return func(*args, **kwargs)
+
return wrapper_args
+
return wrapper_func
user_id = kwargs.pop("user_id", token)
result = function(*args, **kwargs, user_id=user_id)
return result
+
return wrapper
--- /dev/null
+import pytest
+
+
+def test_valid_string():
+ from python_moonutilities.security_functions import validate_data
+ validate_data("CorrectString")
+
+def test_unvalid_string():
+ from python_moonutilities.security_functions import validate_data
+ with pytest.raises(Exception) as exception_info:
+ validate_data("Notcorrect String")
+
+ assert str(exception_info.value) == 'String contains space'
+
+def test_empty_string():
+ from python_moonutilities.security_functions import validate_data
+ with pytest.raises(Exception) as exception_info:
+ validate_data("")
+
+ assert str(exception_info.value) == 'Empty String'
+
+
+def test_none_value():
+ from python_moonutilities.security_functions import validate_data
+ with pytest.raises(Exception) as exception_info:
+ validate_data(None)
+
+ assert str(exception_info.value) == 'Value is Not String or Container or Dictionary'
+
+
+def test_int_value():
+ from python_moonutilities.security_functions import validate_data
+ with pytest.raises(Exception) as exception_info:
+ validate_data(1)
+
+ assert str(exception_info.value) == 'Value is Not String or Container or Dictionary'
+
+
+def test_float_value():
+ from python_moonutilities.security_functions import validate_data
+ with pytest.raises(Exception) as exception_info:
+ validate_data(1.23)
+
+ assert str(exception_info.value) == 'Value is Not String or Container or Dictionary'
+
+
+def test_correct_list():
+ from python_moonutilities.security_functions import validate_data
+ validate_data(["skjdnfa","dao","daosdjpw"])
+
+
+def test_correct_list():
+ from python_moonutilities.security_functions import validate_data
+ validate_data(["skjdnfa"])
+
+
+def test_correct_instead_list():
+ from python_moonutilities.security_functions import validate_data
+ validate_data([["skjdnfa","daswi"],[["daskdlw"],["daklwo"]],["dawl","afioa"],["dawno"]])
+
+
+def test_empty_list():
+ from python_moonutilities.security_functions import validate_data
+ with pytest.raises(Exception) as exception_info:
+ validate_data([])
+
+ assert str(exception_info.value) == 'Empty Container'
+
+
+def test_empty_list_inside_other_list():
+ from python_moonutilities.security_functions import validate_data
+ with pytest.raises(Exception) as exception_info:
+ validate_data(["dajiwdj",[]])
+
+ assert str(exception_info.value) == 'Empty Container'
+
+
+def test_incorrect_string_inside_list():
+ from python_moonutilities.security_functions import validate_data
+ with pytest.raises(Exception) as exception_info:
+ validate_data(["dajiwdj",["dakwe","daow awoepa"]])
+
+ assert str(exception_info.value) == 'String contains space'
+
+
+def test_empty_string_inside_list():
+ from python_moonutilities.security_functions import validate_data
+ with pytest.raises(Exception) as exception_info:
+ validate_data(["dajiwdj", ["dakwe", ""]])
+
+ assert str(exception_info.value) == 'Empty String'
+
+
+def test_correct_tuples():
+ from python_moonutilities.security_functions import validate_data
+ validate_data(("dasdw","dawdwa"))
+
+
+def test_empty_tuples():
+ from python_moonutilities.security_functions import validate_data
+ with pytest.raises(Exception) as exception_info:
+ validate_data(())
+
+ assert str(exception_info.value) == 'Empty Container'
+
+def test_correct_tuple_of_tuple():
+ from python_moonutilities.security_functions import validate_data
+ validate_data(("gjosjefa",("diwajdi","oejfoea"),(("jwdi","fjia"),("nfioa","ifao"))))
+
+
+def test_incorrect_tuple():
+ from python_moonutilities.security_functions import validate_data
+ with pytest.raises(Exception) as exception_info:
+ validate_data(("djawo","dowa afw"))
+
+ assert str(exception_info.value) == 'String contains space'
+
+
+def test_correct_dictionary():
+ from python_moonutilities.security_functions import validate_data
+ validate_data({"daiwdw":"dwioajd"})
+
+
+def test_incorrect_dictionary():
+ from python_moonutilities.security_functions import validate_data
+ with pytest.raises(Exception) as exception_info:
+ validate_data({"daiwdw":"dwioa jd"})
+
+ assert str(exception_info.value) == 'String contains space'
+
+def test_empty_dictionary():
+ from python_moonutilities.security_functions import validate_data
+ with pytest.raises(Exception) as exception_info:
+ validate_data({})
+
+ assert str(exception_info.value) == 'Empty Dictionary'
+
+
+def test_correct_function_pass():
+ from python_moonutilities.security_functions import validate_input
+
+ @validate_input()
+ def temp_function(string,list,tuple):
+ if string!="teststring" :
+ raise ValueError("values which passed incorrect")
+
+ temp_function("teststring",["teststring",["teststring"]],("teststring",("teststring")))
+
+def test_incorrect_function_pass1():
+ from python_moonutilities.security_functions import validate_input
+
+ @validate_input()
+ def temp_function(string, list, tuple):
+ if string != "teststring":
+ raise ValueError("values which passed incorrect")
+
+ with pytest.raises(Exception) as exception_info:
+ temp_function("teststring",list=["teststring", ["testst ring"]],tuple=("teststring", ("teststri ng")))
+
+ assert str(exception_info.value) == 'String contains space'
+
+
+def test_incorrect_function_pass2():
+ from python_moonutilities.security_functions import validate_input
+
+ @validate_input()
+ def temp_function(string, list, dictionary):
+ if string != "teststring":
+ raise ValueError("values which passed incorrect")
+
+ with pytest.raises(Exception) as exception_info:
+ temp_function("teststring", ["teststring", ["teststri ng"]], {"teststring": ("teststring")})
+
+ assert str(exception_info.value) == 'String contains space'
+
+
+def test_incorrect_function_pass3():
+ from python_moonutilities.security_functions import validate_input
+
+ class x:
+ @validate_input()
+ def temp_function(string, list, dictionary):
+ if string != "teststring":
+ raise ValueError("values which passed incorrect")
+
+ e=x;
+
+ with pytest.raises(Exception) as exception_info:
+ e.temp_function("teststring", ["teststring", ["teststri ng"]], {"teststring": ("teststring")})
+
+ assert str(exception_info.value) == 'String contains space'
Code Review / moon.git / commitdiff