From 7bb53c64da2dcf88894bfd31503accdd81498f3d Mon Sep 17 00:00:00 2001 From: Thomas Duval Date: Wed, 3 Jun 2020 10:06:52 +0200 Subject: Update to new version 5.4 Signed-off-by: Thomas Duval Change-Id: Idcd868133d75928a1ffd74d749ce98503e0555ea --- .../tests/unit_python/api/test_policies.py | 444 ++++++++++++--------- 1 file changed, 263 insertions(+), 181 deletions(-) (limited to 'moon_manager/tests/unit_python/api/test_policies.py') diff --git a/moon_manager/tests/unit_python/api/test_policies.py b/moon_manager/tests/unit_python/api/test_policies.py index 76161d53..a07ba725 100644 --- a/moon_manager/tests/unit_python/api/test_policies.py +++ b/moon_manager/tests/unit_python/api/test_policies.py @@ -1,23 +1,32 @@ -# Copyright 2018 Open Platform for NFV Project, Inc. and its contributors -# This software is distributed under the terms and conditions of the 'Apache-2.0' -# license which can be found in the file 'LICENSE' in this package distribution -# or at 'http://www.apache.org/licenses/LICENSE-2.0'. +# Software Name: MOON -import json +# 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. + + +from falcon import HTTP_200, HTTP_400, HTTP_405, HTTP_409 +import hug from uuid import uuid4 -import api.utilities as utilities +import pytest +from moon_utilities import exceptions from helpers import model_helper from helpers import policy_helper -from helpers import data_builder -def get_policies(client): - req = client.get("/policies") - policies = utilities.get_json(req.data) - return req, policies +def get_policies(auth_headers): + from moon_manager.api import policy + req = hug.test.get(policy, "policies", headers=auth_headers) + return req -def add_policies(client, name): +def add_policies(name, auth_headers): + from moon_manager.api import policy req = model_helper.add_model(model_id="mls_model_id" + uuid4().hex) model_id = list(req.keys())[0] data = { @@ -26,106 +35,153 @@ def add_policies(client, name): "model_id": model_id, "genre": "genre" } - req = client.post("/policies", data=json.dumps(data), - headers={'Content-Type': 'application/json'}) - policies = utilities.get_json(req.data) - return req, policies + req = hug.test.post(policy, "policies", data, headers=auth_headers) + return req -def delete_policies_without_id(client): - req = client.delete("/policies/{}".format("")) +def delete_policies_without_id(auth_headers): + from moon_manager.api import policy + req = hug.test.delete(policy, "policies/{}".format(""), headers=auth_headers) return req def test_get_policies(): - client = utilities.register_client() - req, policies = get_policies(client) - assert req.status_code == 200 - assert isinstance(policies, dict) - assert "policies" in policies + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} + req = get_policies(auth_headers=auth_headers) + assert req.status == HTTP_200 + assert isinstance(req.data, dict) + assert "policies" in req.data def test_add_policies(): + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} policy_name = "testuser" + uuid4().hex - client = utilities.register_client() - req, policies = add_policies(client, policy_name) - assert req.status_code == 200 - assert isinstance(policies, dict) - value = list(policies["policies"].values())[0] - assert "policies" in policies + req = add_policies(policy_name, auth_headers=auth_headers) + assert req.status == HTTP_200 + assert isinstance(req.data, dict) + value = list(req.data["policies"].values())[0] + assert "policies" in req.data assert value['name'] == policy_name assert value["description"] == "description of {}".format(policy_name) def test_add_policies_without_model(): + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} + from moon_manager.api import policy policy_name = "testuser" + uuid4().hex - client = utilities.register_client() data = { "name": policy_name, "description": "description of {}".format(policy_name), "model_id": "", "genre": "genre" } - req = client.post("/policies/", data=json.dumps(data), - headers={'Content-Type': 'application/json'}) + req = hug.test.post(policy, "policies/", data, headers=auth_headers) - assert req.status_code == 200 + assert req.status == HTTP_200 def test_add_policies_with_same_name(): + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} name = uuid4().hex policy_name = name - client = utilities.register_client() - req, policies = add_policies(client, policy_name) - assert req.status_code == 200 - assert isinstance(policies, dict) - value = list(policies["policies"].values())[0] - assert "policies" in policies + req = add_policies(policy_name, auth_headers=auth_headers) + assert req.status == HTTP_200 + assert isinstance(req.data, dict) + value = list(req.data["policies"].values())[0] + assert "policies" in req.data assert value['name'] == policy_name assert value["description"] == "description of {}".format(policy_name) - client = utilities.register_client() - req, policies = add_policies(client, policy_name) - assert req.status_code == 409 - assert json.loads(req.data)["message"] == '409: Policy Already Exists' + with pytest.raises(exceptions.PolicyExisting) as exception_info: + req = add_policies(policy_name, auth_headers=auth_headers) + assert "409: Policy Already Exists" == str(exception_info.value) + # assert req.status == HTTP_409 + # assert req.data["message"] == '409: Policy Already Exists' def test_add_policy_with_empty_name(): + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} policy_name = "" - client = utilities.register_client() - req, policies = add_policies(client, policy_name) - assert req.status_code == 400 - assert json.loads(req.data)["message"] == '400: Policy Content Error' + with pytest.raises(exceptions.PolicyContentError) as exception_info: + req = add_policies(policy_name, auth_headers=auth_headers) + assert "400: Policy Content Error" == str(exception_info.value) + # assert req.status == HTTP_400 + # assert req.data["message"] == '400: Policy Content Error' -def test_update_policies_with_model(): +def test_add_policy_with_model_has_no_meta_rule(): + from moon_utilities.auth_functions import get_api_key_for_user + from moon_manager.api import policy + + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} policy_name = "testuser" + uuid4().hex - client = utilities.register_client() + req = model_helper.add_model_without_meta_rule() + model_id = list(req.keys())[0] data = { "name": policy_name, "description": "description of {}".format(policy_name), - "model_id": "", + "model_id": model_id, "genre": "genre" } - req = client.post("/policies/", data=json.dumps(data), - headers={'Content-Type': 'application/json'}) - policy_id = next(iter(utilities.get_json(req.data)['policies'])) - req = model_helper.add_model(model_id="mls_model_id" + uuid4().hex) + with pytest.raises(exceptions.MetaRuleUnknown) as exception_info: + hug.test.post(policy, "policies/", data, headers=auth_headers) + assert "400: Meta Rule Unknown" == str(exception_info.value) + + +def test_add_policy_with_model_has_blank_subject_meta_rule(): + from moon_utilities.auth_functions import get_api_key_for_user + from moon_manager.api import policy + + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} + policy_name = "testuser" + uuid4().hex + req = model_helper.add_model_with_blank_subject_meta_rule() model_id = list(req.keys())[0] data = { - "name": policy_name + "-2", + "name": policy_name, "description": "description of {}".format(policy_name), "model_id": model_id, "genre": "genre" } - req = client.patch("/policies/{}".format(policy_id), data=json.dumps(data), - headers={'Content-Type': 'application/json'}) - assert req.status_code == 200 - assert json.loads(req.data)['policies'][policy_id]['name'] == policy_name + '-2' + with pytest.raises(exceptions.MetaRuleContentError) as exception_info: + hug.test.post(policy, "policies/", data, headers=auth_headers) + assert "400: Meta Rule Error" == str(exception_info.value) + + + +# FIXME: uncomment when model API is re-inserted +# def test_update_policies_with_model(): +# from moon_manager.api import policy +# policy_name = "testuser" + uuid4().hex +# data = { +# "name": policy_name, +# "description": "description of {}".format(policy_name), +# "model_id": "", +# "genre": "genre" +# } +# req = hug.test.post(policy, "policies/", data) +# policy_id = next(iter(req.data['policies'])) +# req = model_helper.add_model(model_id="mls_model_id" + uuid4().hex) +# model_id = list(req.data.keys())[0] +# data = { +# "name": policy_name + "-2", +# "description": "description of {}".format(policy_name), +# "model_id": model_id, +# "genre": "genre" +# } +# req = hug.test.patch("policies/{}".format(policy_id), data) +# assert req.status == HTTP_200 +# assert req.data['policies'][policy_id]['name'] == policy_name + '-2' def test_update_policies_name_success(): + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} + from moon_manager.api import policy policy_name = "testuser" + uuid4().hex - client = utilities.register_client() req = model_helper.add_model(model_id="mls_model_id" + uuid4().hex) model_id = list(req.keys())[0] data = { @@ -134,9 +190,8 @@ def test_update_policies_name_success(): "model_id": model_id, "genre": "genre" } - req = client.post("/policies/", data=json.dumps(data), - headers={'Content-Type': 'application/json'}) - policy_id = next(iter(utilities.get_json(req.data)['policies'])) + req = hug.test.post(policy, "policies/", data, headers=auth_headers) + policy_id = next(iter(req.data['policies'])) data = { "name": policy_name + "-2", @@ -144,26 +199,24 @@ def test_update_policies_name_success(): "model_id": model_id, "genre": "genre" } - req = client.patch("/policies/{}".format(policy_id), data=json.dumps(data), - headers={'Content-Type': 'application/json'}) - assert req.status_code == 200 - assert json.loads(req.data)['policies'][policy_id]['name'] == policy_name + '-2' + req = hug.test.patch(policy, "policies/{}".format(policy_id), data, headers=auth_headers) + assert req.status == HTTP_200 + assert req.data['policies'][policy_id]['name'] == policy_name + '-2' -def test_update_policies_model_unused(): +def test_update_blank_policies_with_model(): + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} + from moon_manager.api import policy policy_name = uuid4().hex - client = utilities.register_client() - req = model_helper.add_model(model_id="mls_model_id" + uuid4().hex) - model_id = list(req.keys())[0] data = { "name": policy_name, "description": "description of {}".format(policy_name), - "model_id": model_id, + "model_id": "", "genre": "genre" } - req = client.post("/policies/", data=json.dumps(data), - headers={'Content-Type': 'application/json'}) - policy_id = next(iter(utilities.get_json(req.data)['policies'])) + req = hug.test.post(policy, "policies/", data, headers=auth_headers) + policy_id = next(iter(req.data['policies'])) req = model_helper.add_model(model_id="mls_model_id" + uuid4().hex) model_id = list(req.keys())[0] data = { @@ -172,55 +225,82 @@ def test_update_policies_model_unused(): "model_id": model_id, "genre": "genre" } - req = client.patch("/policies/{}".format(policy_id), data=json.dumps(data), - headers={'Content-Type': 'application/json'}) - assert req.status_code == 200 + req = hug.test.patch(policy, "policies/{}".format(policy_id), data, headers=auth_headers) + assert req.status == HTTP_200 -def test_update_policy_name_with_existed_one(): - policy_name1 = "testuser" + uuid4().hex - client = utilities.register_client() +def test_update_policies_model_unused(): + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} + from moon_manager.api import policy + policy_name = uuid4().hex req = model_helper.add_model(model_id="mls_model_id" + uuid4().hex) model_id = list(req.keys())[0] data = { - "name": policy_name1, - "description": "description of {}".format(policy_name1), + "name": policy_name, + "description": "description of {}".format(policy_name), "model_id": model_id, "genre": "genre" } - req = client.post("/policies/", data=json.dumps(data), - headers={'Content-Type': 'application/json'}) - policy_id1 = next(iter(utilities.get_json(req.data)['policies'])) - - policy_name2 = "testuser" + uuid4().hex - client = utilities.register_client() + req = hug.test.post(policy, "policies/", data, headers=auth_headers) + policy_id = next(iter(req.data['policies'])) req = model_helper.add_model(model_id="mls_model_id" + uuid4().hex) model_id = list(req.keys())[0] data = { - "name": policy_name2, - "description": "description of {}".format(policy_name2), + "name": policy_name, + "description": "description of {}".format(policy_name), "model_id": model_id, "genre": "genre" } - req = client.post("/policies/", data=json.dumps(data), - headers={'Content-Type': 'application/json'}) - policy_id2 = next(iter(utilities.get_json(req.data)['policies'])) - data = { - "name": policy_name1, - "description": "description of {}".format(policy_name1), - "model_id": model_id, - "genre": "genre" - } - req = client.patch("/policies/{}".format(policy_id2), data=json.dumps(data), - headers={'Content-Type': 'application/json'}) - assert req.status_code == 409 - assert json.loads(req.data)["message"] == '409: Policy Already Exists' + with pytest.raises(exceptions.PolicyUpdateError) as exception_info: + req = hug.test.patch(policy, "policies/{}".format(policy_id), data, headers=auth_headers) + assert "400: Policy update error" == str(exception_info.value) + + +# FIXME: uncomment when model API is re-inserted +# def test_update_policy_name_with_existed_one(): +# from moon_manager.api import policy +# policy_name1 = "testuser" + uuid4().hex +# req = model_helper.add_model(model_id="mls_model_id" + uuid4().hex) +# model_id = list(req.keys())[0] +# data = { +# "name": policy_name1, +# "description": "description of {}".format(policy_name1), +# "model_id": model_id, +# "genre": "genre" +# } +# req = hug.test.post(policy, "policies/", data) +# policy_id1 = next(iter(req.data['policies'])) +# +# policy_name2 = "testuser" + uuid4().hex +# eq = model_helper.add_model(model_id="mls_model_id" + uuid4().hex) +# model_id = list(req.data.keys())[0] +# data = { +# "name": policy_name2, +# "description": "description of {}".format(policy_name2), +# "model_id": model_id, +# "genre": "genre" +# } +# req = hug.test.post(policy, "policies/", data) +# policy_id2 = next(iter(req.data['policies'])) +# +# data = { +# "name": policy_name1, +# "description": "description of {}".format(policy_name1), +# "model_id": model_id, +# "genre": "genre" +# } +# req = hug.test.patch(policy, "policies/{}".format(policy_id2), data) +# assert req.status == HTTP_409 +# assert req.data["message"] == '409: Policy Already Exists' def test_update_policies_with_empty_name(): + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} + from moon_manager.api import policy policy_name = "testuser" + uuid4().hex - client = utilities.register_client() req = model_helper.add_model(model_id="mls_model_id" + uuid4().hex) model_id = list(req.keys())[0] data = { @@ -229,9 +309,8 @@ def test_update_policies_with_empty_name(): "model_id": model_id, "genre": "genre" } - req = client.post("/policies/", data=json.dumps(data), - headers={'Content-Type': 'application/json'}) - policy_id = next(iter(utilities.get_json(req.data)['policies'])) + req = hug.test.post(policy, "policies/", data, headers=auth_headers) + policy_id = next(iter(req.data['policies'])) data = { "name": "", @@ -239,15 +318,18 @@ def test_update_policies_with_empty_name(): "model_id": model_id, "genre": "genre" } - req = client.patch("/policies/{}".format(policy_id), data=json.dumps(data), - headers={'Content-Type': 'application/json'}) - assert req.status_code == 400 - assert json.loads(req.data)["message"] == '400: Policy Content Error' + with pytest.raises(exceptions.PolicyContentError) as exception_info: + req = hug.test.patch(policy, "policies/{}".format(policy_id), data, headers=auth_headers) + assert "400: Policy Content Error" == str(exception_info.value) + # assert req.status == HTTP_400 + # assert req.data["message"] == '400: Policy Content Error' def test_update_policies_with_blank_model(): + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} + from moon_manager.api import policy policy_name = "testuser" + uuid4().hex - client = utilities.register_client() req = model_helper.add_model(model_id="mls_model_id" + uuid4().hex) model_id = list(req.keys())[0] data = { @@ -256,9 +338,8 @@ def test_update_policies_with_blank_model(): "model_id": model_id, "genre": "genre" } - req = client.post("/policies/", data=json.dumps(data), - headers={'Content-Type': 'application/json'}) - policy_id = next(iter(utilities.get_json(req.data)['policies'])) + req = hug.test.post(policy, "policies/", data, headers=auth_headers) + policy_id = next(iter(req.data['policies'])) data = { "name": policy_name, @@ -267,76 +348,77 @@ def test_update_policies_with_blank_model(): "genre": "genre" } - req = client.patch("/policies/{}".format(policy_id), data=json.dumps(data), - headers={'Content-Type': 'application/json'}) - assert req.status_code == 200 + with pytest.raises(exceptions.PolicyUpdateError) as exception_info: + req = hug.test.patch(policy, "policies/{}".format(policy_id), data, headers=auth_headers) + assert "400: Policy update error" == str(exception_info.value) -def test_update_policies_connected_to_rules_with_blank_model(): - client = utilities.register_client() - req, rules, policy_id = data_builder.add_rules(client) - req = client.get("/policies") - data = utilities.get_json(req.data) - for policy_obj_id in data['policies']: - if policy_obj_id == policy_id: - policy = data['policies'][policy_obj_id] - policy['model_id'] = '' - req = client.patch("/policies/{}".format(policy_id), data=json.dumps(policy), - headers={'Content-Type': 'application/json'}) - assert req.status_code == 400 - assert json.loads(req.data)["message"] == '400: Policy update error' +# FIXME: uncomment when model API is re-inserted +# def test_update_policies_connected_to_rules_with_blank_model(): +# from moon_manager.api import policy +# req, rules, policy_id = data_builder.add_rules() +# req = hug.test.get(policy, "policies") +# for policy_obj_id in req.data['policies']: +# if policy_obj_id == policy_id: +# policy = req.data['policies'][policy_obj_id] +# policy['model_id'] = '' +# req = hug.test.patch("/policies/{}".format(policy_id), req.data) +# assert req.status == HTTP_400 +# assert req.data["message"] == '400: Policy update error' def test_delete_policies(): - client = utilities.register_client() - - policy = policy_helper.add_policies() - policy_id = list(policy.keys())[0] - - req = client.delete("/policies/{}".format(policy_id)) - assert req.status_code == 200 - - -def test_delete_policy_with_dependencies_rule(): - client = utilities.register_client() - req, rules, policy_id = data_builder.add_rules(client) - req = client.delete("/policies/{}".format(policy_id)) - assert req.status_code == 400 - assert json.loads(req.data)["message"] == '400: Policy With Rule Error' - - -def test_delete_policy_with_dependencies_subject_data(): - client = utilities.register_client() - req, rules, policy_id = data_builder.add_rules(client) - req = client.delete("/policies/{}/rules/{}".format(policy_id, next(iter(rules['rules'])))) - assert req.status_code == 200 - req = client.delete("/policies/{}".format(policy_id)) - assert req.status_code == 400 - assert json.loads(req.data)["message"] == '400: Policy With Data Error' - - -def test_delete_policy_with_dependencies_perimeter(): - client = utilities.register_client() - policy = policy_helper.add_policies() - policy_id = next(iter(policy)) - - data = { - "name": 'testuser'+uuid4().hex, - "description": "description of {}".format(uuid4().hex), - "password": "password for {}".format(uuid4().hex), - "email": "{}@moon".format(uuid4().hex) - } - req = client.post("/policies/{}/subjects".format(policy_id), data=json.dumps(data), - headers={'Content-Type': 'application/json'}) - - assert req.status_code == 200 - req = client.delete("/policies/{}".format(policy_id)) - assert req.status_code == 400 - assert json.loads(req.data)["message"] == '400: Policy With Perimeter Error' + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} + from moon_manager.api import policy + _policy = policy_helper.add_policies() + policy_id = list(_policy.keys())[0] + req = hug.test.delete(policy, "policies/{}".format(policy_id), headers=auth_headers) + assert req.status == HTTP_200 + + +# FIXME: uncomment when rule API is re-inserted +# def test_delete_policy_with_dependencies_rule(): +# from moon_manager.api import policy +# req, rules, policy_id = data_builder.add_rules() +# req = hug.test.delete(policy, "policies/{}".format(policy_id)) +# assert req.status == HTTP_400 +# assert req.data["message"] == '400: Policy With Rule Error' + + +# FIXME: uncomment when perimeter API is re-inserted +# def test_delete_policy_with_dependencies_subject_data(): +# from moon_manager.api import policy +# req, rules, policy_id = data_builder.add_rules() +# req = hug.test.delete(policy, "policies/{}/rules/{}".format(policy_id, next(iter(rules['rules'])))) +# assert req.status == HTTP_200 +# req = hug.test.delete(policy, "policies/{}".format(policy_id)) +# assert req.status == HTTP_400 +# assert req.data["message"] == '400: Policy With Data Error' + + +# FIXME: uncomment when perimeter API is re-inserted +# def test_delete_policy_with_dependencies_perimeter(): +# from moon_manager.api import policy +# _policy = policy_helper.add_policies() +# policy_id = next(iter(_policy)) +# +# data = { +# "name": 'testuser'+uuid4().hex, +# "description": "description of {}".format(uuid4().hex), +# "password": "password for {}".format(uuid4().hex), +# "email": "{}@moon".format(uuid4().hex) +# } +# req = hug.test.post(policy, "policies/{}/subjects".format(policy_id), data) +# +# assert req.status == HTTP_200 +# req = hug.test.delete(policy, "policies/{}".format(policy_id)) +# assert req.status == HTTP_400 +# assert req.data["message"] == '400: Policy With Perimeter Error' def test_delete_policies_without_id(): - client = utilities.register_client() - req = delete_policies_without_id(client) - assert req.status_code == 400 - assert json.loads(req.data)["message"] == '400: Policy Unknown' + from moon_utilities.auth_functions import get_api_key_for_user + auth_headers = {"X-Api-Key": get_api_key_for_user("admin")} + req = delete_policies_without_id(auth_headers=auth_headers) + assert req.status == HTTP_405 -- cgit 1.2.3-korg