diff options
Diffstat (limited to 'moon_manager/tests/unit_python/api/test_assignement.py')
| -rw-r--r-- | moon_manager/tests/unit_python/api/test_assignement.py | 349 |
1 files changed, 269 insertions, 80 deletions
diff --git a/moon_manager/tests/unit_python/api/test_assignement.py b/moon_manager/tests/unit_python/api/test_assignement.py index b56fb420..3a127477 100644 --- a/moon_manager/tests/unit_python/api/test_assignement.py +++ b/moon_manager/tests/unit_python/api/test_assignement.py @@ -1,19 +1,60 @@ +# Software Name: MOON + +# Version: 5.4 + +# SPDX-FileCopyrightText: Copyright (c) 2018-2020 Orange and its contributors +# SPDX-License-Identifier: Apache-2.0 + +# This software is distributed under the 'Apache License 2.0', +# the text of which is available at 'http://www.apache.org/licenses/LICENSE-2.0.txt' +# or see the "LICENSE" file for more details. + + +import hug +from uuid import uuid4 +import pytest import api.utilities as utilities -import json from helpers import data_builder as builder -from uuid import uuid4 +from moon_utilities import exceptions + + +def delete_assignment_based_on_parameters(type, policy_id, pre_id=None, cat_id=None, data_id=None): + if type in ["subject_assignments", "object_assignments", "action_assignments"] and policy_id: + url = "/policies/" + policy_id + "/" + type + if pre_id: + url += "/" + pre_id + if cat_id: + url += "/" + cat_id + if data_id: + url += "/" + data_id + else: + return "" + from moon_manager.api import assignments + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} + + req = hug.test.delete(assignments, url, headers=auth_headers) + return req # subject_categories_test -def get_subject_assignment(client, policy_id): - req = client.get("/policies/{}/subject_assignments".format(policy_id)) +def get_subject_assignment(policy_id): + from moon_manager.api import assignments + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} + + req = hug.test.get(assignments, "/policies/{}/subject_assignments".format(policy_id), headers=auth_headers) subject_assignment = utilities.get_json(req.data) return req, subject_assignment -def add_subject_assignment(client): +def add_subject_assignment(): + from moon_manager.api import assignments + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} + subject_category_id, object_category_id, action_category_id, meta_rule_id, policy_id = builder.create_new_policy( subject_category_name="subject_category1" + uuid4().hex, object_category_name="object_category1" + uuid4().hex, @@ -27,34 +68,41 @@ def add_subject_assignment(client): "category_id": subject_category_id, "data_id": data_id } - req = client.post("/policies/{}/subject_assignments".format(policy_id), data=json.dumps(data), - headers={'Content-Type': 'application/json'}) + req = hug.test.post(assignments, "/policies/{}/subject_assignments/".format(policy_id), + body=data, headers=auth_headers) subject_assignment = utilities.get_json(req.data) return req, subject_assignment -def add_subject_assignment_without_cat_id(client): +def add_subject_assignment_without_cat_id(): + from moon_manager.api import assignments + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} data = { "id": "subject_id", "category_id": "", "data_id": "data_id" } - req = client.post("/policies/{}/subject_assignments".format("1111"), data=json.dumps(data), - headers={'Content-Type': 'application/json'}) + req = hug.test.post(assignments, "/policies/{}/subject_assignments".format("1111"), body=data, + headers=auth_headers) subject_assignment = utilities.get_json(req.data) return req, subject_assignment -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)) +def delete_subject_assignment(policy_id, sub_id, cat_id, data_id): + from moon_manager.api import assignments + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} + + req = hug.test.delete(assignments, "/policies/{}/subject_assignments/{}/{}/{}".format( + policy_id, sub_id, cat_id, data_id), headers=auth_headers) return req def test_add_subject_assignment(): - client = utilities.register_client() - req, subject_assignment = add_subject_assignment(client) - assert req.status_code == 200 + req, subject_assignment = add_subject_assignment() + assert req.status == hug.HTTP_200 assert isinstance(subject_assignment, dict) assert "subject_assignments" in subject_assignment @@ -62,50 +110,97 @@ def test_add_subject_assignment(): # def test_add_subject_assignment_without_cat_id(): # client = utilities.register_client() # req, subject_assignment = add_subject_assignment_without_cat_id(client) -# assert req.status_code == 400 +# assert req.status == hug.HTTP_400 # assert json.loads(req.data)["message"] == "Key: 'category_id', [Empty String]" def test_get_subject_assignment(): - client = utilities.register_client() policy_id = builder.get_policy_id_with_subject_assignment() - req, subject_assignment = get_subject_assignment(client, policy_id) - assert req.status_code == 200 + req, subject_assignment = get_subject_assignment(policy_id) + assert req.status == hug.HTTP_200 assert isinstance(subject_assignment, dict) assert "subject_assignments" in subject_assignment def test_delete_subject_assignment(): - client = utilities.register_client() policy_id = builder.get_policy_id_with_subject_assignment() - req, subject_assignment = get_subject_assignment(client, policy_id) + req, subject_assignment = get_subject_assignment(policy_id) + value = subject_assignment["subject_assignments"] + _id = list(value.keys())[0] + success_req = delete_subject_assignment( + policy_id, + value[_id]['subject_id'], + value[_id]['category_id'], + value[_id]['assignments'][0]) + assert success_req.status == hug.HTTP_200 + + +def test_delete_subject_assignment_using_policy(): + policy_id = builder.get_policy_id_with_subject_assignment() + req, subject_assignment = get_subject_assignment(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 + success_req = delete_assignment_based_on_parameters( + "subject_assignments", + policy_id) + assert success_req.status == hug.HTTP_200 + + +def test_delete_subject_assignment_using_policy_perimeter_id(): + policy_id = builder.get_policy_id_with_subject_assignment() + req, subject_assignment = get_subject_assignment(policy_id) + value = subject_assignment["subject_assignments"] + _id = list(value.keys())[0] + success_req = delete_assignment_based_on_parameters( + "subject_assignments", + policy_id, + value[_id]['subject_id']) + assert success_req.status == hug.HTTP_200 + + +def test_delete_subject_assignment_using_policy_perimeter_id_category_id(): + policy_id = builder.get_policy_id_with_subject_assignment() + req, subject_assignment = get_subject_assignment(policy_id) + value = subject_assignment["subject_assignments"] + _id = list(value.keys())[0] + success_req = delete_assignment_based_on_parameters( + "subject_assignments", + policy_id, + value[_id]['subject_id'], + value[_id]['category_id']) + assert success_req.status == hug.HTTP_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 == 404 + + with pytest.raises(exceptions.PolicyUnknown) as exception_info: + success_req = delete_subject_assignment("", "id1", "111", "data_id1") + + assert '400: Policy Unknown' == str(exception_info.value) + + # assert success_req.status == hug.HTTP_400 + # assert success_req.data["message"] == "400: Policy Unknown" # --------------------------------------------------------------------------- # object_categories_test -def get_object_assignment(client, policy_id): - req = client.get("/policies/{}/object_assignments".format(policy_id)) +def get_object_assignment(policy_id): + from moon_manager.api import assignments + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} + + req = hug.test.get(assignments, "/policies/{}/object_assignments".format(policy_id), headers=auth_headers) object_assignment = utilities.get_json(req.data) return req, object_assignment -def add_object_assignment(client): +def add_object_assignment(): + from moon_manager.api import assignments + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} + subject_category_id, object_category_id, action_category_id, meta_rule_id, policy_id = builder.create_new_policy( subject_category_name="subject_category1" + uuid4().hex, object_category_name="object_category1" + uuid4().hex, @@ -120,84 +215,135 @@ def add_object_assignment(client): "data_id": data_id } - req = client.post("/policies/{}/object_assignments".format(policy_id), data=json.dumps(data), - headers={'Content-Type': 'application/json'}) + req = hug.test.post(assignments, "/policies/{}/object_assignments".format(policy_id), + body=data, headers=auth_headers) object_assignment = utilities.get_json(req.data) return req, object_assignment -def add_object_assignment_without_cat_id(client): +def add_object_assignment_without_cat_id(): + from moon_manager.api import assignments + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} data = { "id": "object_id", "category_id": "", "data_id": "data_id" } - req = client.post("/policies/{}/object_assignments".format("1111"), data=json.dumps(data), - headers={'Content-Type': 'application/json'}) + req = hug.test.post(assignments, "/policies/{}/object_assignments".format("1111"), + body=data, headers=auth_headers) object_assignment = utilities.get_json(req.data) return req, object_assignment -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)) +def delete_object_assignment(policy_id, obj_id, cat_id, data_id): + from moon_manager.api import assignments + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} + + req = hug.test.delete(assignments, "/policies/{}/object_assignments/{}/{}/{}".format( + policy_id, obj_id, cat_id, data_id), headers=auth_headers) return req def test_get_object_assignment(): policy_id = builder.get_policy_id_with_object_assignment() - client = utilities.register_client() - req, object_assignment = get_object_assignment(client, policy_id) - assert req.status_code == 200 + + req, object_assignment = get_object_assignment(policy_id) + assert req.status == hug.HTTP_200 assert isinstance(object_assignment, dict) assert "object_assignments" in object_assignment def test_add_object_assignment(): - client = utilities.register_client() - req, object_assignment = add_object_assignment(client) - assert req.status_code == 200 + req, object_assignment = add_object_assignment() + assert req.status == hug.HTTP_200 assert "object_assignments" in object_assignment # def test_add_object_assignment_without_cat_id(): # client = utilities.register_client() # req, object_assignment = add_object_assignment_without_cat_id(client) -# assert req.status_code == 400 +# assert req.status == hug.HTTP_400 # assert json.loads(req.data)["message"] == "Key: 'category_id', [Empty String]" def test_delete_object_assignment(): - client = utilities.register_client() policy_id = builder.get_policy_id_with_object_assignment() - req, object_assignment = get_object_assignment(client, policy_id) + req, object_assignment = get_object_assignment(policy_id) value = object_assignment["object_assignments"] _id = list(value.keys())[0] - success_req = delete_object_assignment(client, - policy_id, + success_req = delete_object_assignment(policy_id, value[_id]['object_id'], value[_id]['category_id'], value[_id]['assignments'][0]) - assert success_req.status_code == 200 + assert success_req.status == hug.HTTP_200 + + +def test_delete_object_assignment_using_policy(): + policy_id = builder.get_policy_id_with_object_assignment() + req, object_assignment = get_object_assignment(policy_id) + value = object_assignment["object_assignments"] + _id = list(value.keys())[0] + success_req = delete_assignment_based_on_parameters( + "object_assignments", + policy_id) + assert success_req.status == hug.HTTP_200 + + +def test_delete_object_assignment_using_policy_perimeter_id(): + policy_id = builder.get_policy_id_with_object_assignment() + req, object_assignment = get_object_assignment(policy_id) + value = object_assignment["object_assignments"] + _id = list(value.keys())[0] + success_req = delete_assignment_based_on_parameters( + "object_assignments", + policy_id, + value[_id]['object_id']) + assert success_req.status == hug.HTTP_200 + + +def test_delete_object_assignment_using_policy_perimeter_id_category_id(): + policy_id = builder.get_policy_id_with_object_assignment() + req, object_assignment = get_object_assignment(policy_id) + value = object_assignment["object_assignments"] + _id = list(value.keys())[0] + success_req = delete_assignment_based_on_parameters( + "object_assignments", + policy_id, + value[_id]['object_id'], + value[_id]['category_id']) + assert success_req.status == hug.HTTP_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 == 404 + with pytest.raises(exceptions.PolicyUnknown) as exception_info: + success_req = delete_object_assignment("", "id1", "111", "data_id1") + # assert success_req.status == hug.HTTP_400 + # assert success_req.data["message"] == "400: Policy Unknown" + assert '400: Policy Unknown' == str(exception_info.value) # --------------------------------------------------------------------------- # action_categories_test -def get_action_assignment(client, policy_id): - req = client.get("/policies/{}/action_assignments".format(policy_id)) +def get_action_assignment(policy_id): + from moon_manager.api import assignments + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} + + req = hug.test.get(assignments, "/policies/{}/action_assignments".format(policy_id), headers=auth_headers) action_assignment = utilities.get_json(req.data) return req, action_assignment -def add_action_assignment(client): +def add_action_assignment(): + from moon_manager.api import assignments + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} + subject_category_id, object_category_id, action_category_id, meta_rule_id, policy_id = builder.create_new_policy( subject_category_name="subject_category1" + uuid4().hex, object_category_name="object_category1" + uuid4().hex, @@ -211,70 +357,113 @@ def add_action_assignment(client): "category_id": action_category_id, "data_id": data_id } - req = client.post("/policies/{}/action_assignments".format(policy_id), data=json.dumps(data), - headers={'Content-Type': 'application/json'}) + req = hug.test.post(assignments, "/policies/{}/action_assignments".format(policy_id), + body=data, + headers=auth_headers) action_assignment = utilities.get_json(req.data) return req, action_assignment -def add_action_assignment_without_cat_id(client): +def add_action_assignment_without_cat_id(): + from moon_manager.api import assignments + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} data = { "id": "action_id", "category_id": "", "data_id": "data_id" } - req = client.post("/policies/{}/action_assignments".format("1111"), data=json.dumps(data), - headers={'Content-Type': 'application/json'}) + req = hug.test.post(assignments, "/policies/{}/action_assignments".format("1111"), + body=data, headers=auth_headers) action_assignment = utilities.get_json(req.data) return req, action_assignment -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)) +def delete_action_assignment(policy_id, action_id, cat_id, data_id): + from moon_manager.api import assignments + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} + + req = hug.test.delete(assignments, "/policies/{}/action_assignments/{}/{}/{}".format( + policy_id, action_id, cat_id, data_id), headers=auth_headers) return req def test_get_action_assignment(): policy_id = builder.get_policy_id_with_action_assignment() - client = utilities.register_client() - req, action_assignment = get_action_assignment(client, policy_id) - assert req.status_code == 200 + req, action_assignment = get_action_assignment(policy_id) + assert req.status == hug.HTTP_200 assert isinstance(action_assignment, dict) assert "action_assignments" in action_assignment def test_add_action_assignment(): - client = utilities.register_client() - req, action_assignment = add_action_assignment(client) - assert req.status_code == 200 + req, action_assignment = add_action_assignment() + assert req.status == hug.HTTP_200 assert "action_assignments" in action_assignment # def test_add_action_assignment_without_cat_id(): # client = utilities.register_client() # req, action_assignment = add_action_assignment_without_cat_id(client) -# assert req.status_code == 400 +# assert req.status == hug.HTTP_400 # assert json.loads(req.data)["message"] == "Key: 'category_id', [Empty String]" def test_delete_action_assignment(): - client = utilities.register_client() policy_id = builder.get_policy_id_with_action_assignment() - req, action_assignment = get_action_assignment(client, policy_id) + req, action_assignment = get_action_assignment(policy_id) value = action_assignment["action_assignments"] id = list(value.keys())[0] - success_req = delete_action_assignment(client, - policy_id, + success_req = delete_action_assignment(policy_id, value[id]['action_id'], value[id]['category_id'], value[id]['assignments'][0]) - assert success_req.status_code == 200 + assert success_req.status == hug.HTTP_200 + + +def test_delete_action_assignment_policy(): + policy_id = builder.get_policy_id_with_action_assignment() + req, action_assignment = get_action_assignment(policy_id) + value = action_assignment["action_assignments"] + id = list(value.keys())[0] + success_req = delete_assignment_based_on_parameters( + "action_assignments", + policy_id) + assert success_req.status == hug.HTTP_200 + + +def test_delete_action_assignment_policy_perimeter_id(): + policy_id = builder.get_policy_id_with_action_assignment() + req, action_assignment = get_action_assignment(policy_id) + value = action_assignment["action_assignments"] + id = list(value.keys())[0] + success_req = delete_assignment_based_on_parameters( + "action_assignments", + policy_id, + value[id]['action_id']) + assert success_req.status == hug.HTTP_200 + + +def test_delete_action_assignment_policy_perimeter_id_category_id(): + policy_id = builder.get_policy_id_with_action_assignment() + req, action_assignment = get_action_assignment(policy_id) + value = action_assignment["action_assignments"] + id = list(value.keys())[0] + success_req = delete_assignment_based_on_parameters( + "action_assignments", + policy_id, + value[id]['action_id'], + value[id]['category_id']) + assert success_req.status == hug.HTTP_200 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 == 404 + with pytest.raises(exceptions.PolicyUnknown) as exception_info: + success_req = delete_action_assignment("", "id1", "111", "data_id1") + # assert success_req.status == hug.HTTP_400 + # assert success_req.data["message"] == "400: Policy Unknown" + assert '400: Policy Unknown' == str(exception_info.value) # --------------------------------------------------------------------------- |