From 6f18ee4526bd33b54ed8da49dcb62abcf935c905 Mon Sep 17 00:00:00 2001 From: aticig Date: Mon, 15 Aug 2022 23:48:23 +0300 Subject: [PATCH] Refactoring NBI decriptor_topics unit tests, enabling stage-test to fail with tox results Test_VnfdTopic and Test_NsdTopic classes are refactored. Change-Id: Ic8e8310f2c2913cf07ff25755f1a2196d16eb944 Signed-off-by: aticig --- devops-stages/stage-test.sh | 8 +- osm_nbi/tests/test_descriptor_topics.py | 1754 ++++++++++++----------- requirements-test.in | 2 +- requirements-test.txt | 4 +- 4 files changed, 905 insertions(+), 863 deletions(-) diff --git a/devops-stages/stage-test.sh b/devops-stages/stage-test.sh index e5a8e85..a247f37 100755 --- a/devops-stages/stage-test.sh +++ b/devops-stages/stage-test.sh @@ -1,4 +1,4 @@ -#!/bin/sh +#!/bin/bash # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -13,5 +13,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -OUTPUT=$(TOX_PARALLEL_NO_SPINNER=1 tox --parallel=auto) -printf "$OUTPUT" +set -e +echo "Launching tox" +tox --parallel=auto + diff --git a/osm_nbi/tests/test_descriptor_topics.py b/osm_nbi/tests/test_descriptor_topics.py index 147f3e7..27c0bd3 100755 --- a/osm_nbi/tests/test_descriptor_topics.py +++ b/osm_nbi/tests/test_descriptor_topics.py @@ -62,7 +62,7 @@ def compare_desc(tc, d1, d2, k): Lists of different length are not compared :param tc: Test Case wich provides context (in particular the assert* methods) :param d1,d2: Descriptors to be compared - :param key/item being compared + :param k: key/item being compared :return: Nothing """ if isinstance(d1, dict) and isinstance(d2, dict): @@ -105,10 +105,7 @@ class Test_VnfdTopic(TestCase): new_desc = deepcopy(template) return old_desc, new_desc - @patch("osm_nbi.descriptor_topics.shutil") - @patch("osm_nbi.descriptor_topics.os.rename") - def test_new_vnfd(self, mock_rename, mock_shutil): - did = db_vnfd_content["_id"] + def prepare_vnfd_creation(self): self.fs.path = "" self.fs.get_params.return_value = {} self.fs.file_exists.return_value = False @@ -116,503 +113,547 @@ class Test_VnfdTopic(TestCase): "/tmp/" + str(uuid4()), "a+b" ) test_vnfd = deepcopy(db_vnfd_content) + did = db_vnfd_content["_id"] + self.db.create.return_value = did + self.db.get_one.side_effect = [ + {"_id": did, "_admin": deepcopy(db_vnfd_content["_admin"])}, + None, + ] + return did, test_vnfd + + def prepare_test_vnfd(self, test_vnfd): del test_vnfd["_id"] del test_vnfd["_admin"] - with self.subTest(i=1, t="Normal Creation"): - self.db.create.return_value = did - rollback = [] - did2, oid = self.topic.new(rollback, fake_session, {}) - db_args = self.db.create.call_args[0] - msg_args = self.msg.write.call_args[0] - self.assertEqual(len(rollback), 1, "Wrong rollback length") - self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic") - self.assertEqual(msg_args[1], "created", "Wrong message action") - self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content") - self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic") - self.assertEqual(did2, did, "Wrong DB VNFD id") - self.assertIsNotNone(db_args[1]["_admin"]["created"], "Wrong creation time") - self.assertEqual( - db_args[1]["_admin"]["modified"], - db_args[1]["_admin"]["created"], - "Wrong modification time", - ) - self.assertEqual( - db_args[1]["_admin"]["projects_read"], - [test_pid], - "Wrong read-only project list", - ) - self.assertEqual( - db_args[1]["_admin"]["projects_write"], - [test_pid], - "Wrong read-write project list", + del test_vnfd["vdu"][0]["cloud-init-file"] + del test_vnfd["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][ + "day1-2" + ][0]["execution-environment-list"][0]["juju"] + return test_vnfd + + @patch("osm_nbi.descriptor_topics.shutil") + @patch("osm_nbi.descriptor_topics.os.rename") + def test_new_vnfd_normal_creation(self, mock_rename, mock_shutil): + did, test_vnfd = self.prepare_vnfd_creation() + test_vnfd = self.prepare_test_vnfd(test_vnfd) + rollback = [] + did2, oid = self.topic.new(rollback, fake_session, {}) + db_args = self.db.create.call_args[0] + msg_args = self.msg.write.call_args[0] + + self.assertEqual(len(rollback), 1, "Wrong rollback length") + self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic") + self.assertEqual(msg_args[1], "created", "Wrong message action") + self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content") + self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic") + self.assertEqual(did2, did, "Wrong DB VNFD id") + self.assertIsNotNone(db_args[1]["_admin"]["created"], "Wrong creation time") + self.assertEqual( + db_args[1]["_admin"]["modified"], + db_args[1]["_admin"]["created"], + "Wrong modification time", + ) + self.assertEqual( + db_args[1]["_admin"]["projects_read"], + [test_pid], + "Wrong read-only project list", + ) + self.assertEqual( + db_args[1]["_admin"]["projects_write"], + [test_pid], + "Wrong read-write project list", + ) + + self.db.get_one.side_effect = [ + {"_id": did, "_admin": deepcopy(db_vnfd_content["_admin"])}, + None, + ] + + self.topic.upload_content( + fake_session, did, test_vnfd, {}, {"Content-Type": []} + ) + msg_args = self.msg.write.call_args[0] + test_vnfd["_id"] = did + self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic") + self.assertEqual(msg_args[1], "edited", "Wrong message action") + self.assertEqual(msg_args[2], test_vnfd, "Wrong message content") + + db_args = self.db.get_one.mock_calls[0][1] + self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic") + self.assertEqual(db_args[1]["_id"], did, "Wrong DB VNFD id") + + db_args = self.db.replace.call_args[0] + self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic") + self.assertEqual(db_args[1], did, "Wrong DB VNFD id") + + admin = db_args[2]["_admin"] + db_admin = deepcopy(db_vnfd_content["_admin"]) + self.assertEqual(admin["type"], "vnfd", "Wrong descriptor type") + self.assertEqual(admin["created"], db_admin["created"], "Wrong creation time") + self.assertGreater( + admin["modified"], db_admin["created"], "Wrong modification time" + ) + self.assertEqual( + admin["projects_read"], + db_admin["projects_read"], + "Wrong read-only project list", + ) + self.assertEqual( + admin["projects_write"], + db_admin["projects_write"], + "Wrong read-write project list", + ) + self.assertEqual( + admin["onboardingState"], "ONBOARDED", "Wrong onboarding state" + ) + self.assertEqual( + admin["operationalState"], "ENABLED", "Wrong operational state" + ) + self.assertEqual(admin["usageState"], "NOT_IN_USE", "Wrong usage state") + + storage = admin["storage"] + self.assertEqual(storage["folder"], did + ":1", "Wrong storage folder") + self.assertEqual(storage["descriptor"], "package", "Wrong storage descriptor") + self.assertEqual(admin["revision"], 1, "Wrong revision number") + compare_desc(self, test_vnfd, db_args[2], "VNFD") + + @patch("osm_nbi.descriptor_topics.shutil") + @patch("osm_nbi.descriptor_topics.os.rename") + def test_new_vnfd_check_pyangbind_validation_additional_properties( + self, mock_rename, mock_shutil + ): + did, test_vnfd = self.prepare_vnfd_creation() + test_vnfd = self.prepare_test_vnfd(test_vnfd) + self.topic.upload_content( + fake_session, did, test_vnfd, {}, {"Content-Type": []} + ) + test_vnfd["_id"] = did + test_vnfd["extra-property"] = 0 + self.db.get_one.side_effect = lambda table, filter, fail_on_empty=None, fail_on_more=None: { + "_id": did, + "_admin": deepcopy(db_vnfd_content["_admin"]), + } + + with self.assertRaises( + EngineException, msg="Accepted VNFD with an additional property" + ) as e: + self.topic.upload_content( + fake_session, did, test_vnfd, {}, {"Content-Type": []} ) - tmp1 = test_vnfd["vdu"][0]["cloud-init-file"] - tmp2 = test_vnfd["df"][0]["lcm-operations-configuration"][ - "operate-vnf-op-config" - ]["day1-2"][0]["execution-environment-list"][0]["juju"] - del test_vnfd["vdu"][0]["cloud-init-file"] - del test_vnfd["df"][0]["lcm-operations-configuration"][ - "operate-vnf-op-config" - ]["day1-2"][0]["execution-environment-list"][0]["juju"] - try: - self.db.get_one.side_effect = [ - {"_id": did, "_admin": deepcopy(db_vnfd_content["_admin"])}, - None, - ] - self.topic.upload_content( - fake_session, did, test_vnfd, {}, {"Content-Type": []} - ) - msg_args = self.msg.write.call_args[0] - test_vnfd["_id"] = did - self.assertEqual( - msg_args[0], self.topic.topic_msg, "Wrong message topic" - ) - self.assertEqual(msg_args[1], "edited", "Wrong message action") - self.assertEqual(msg_args[2], test_vnfd, "Wrong message content") - db_args = self.db.get_one.mock_calls[0][1] - self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic") - self.assertEqual(db_args[1]["_id"], did, "Wrong DB VNFD id") - db_args = self.db.replace.call_args[0] - self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic") - self.assertEqual(db_args[1], did, "Wrong DB VNFD id") - admin = db_args[2]["_admin"] - db_admin = deepcopy(db_vnfd_content["_admin"]) - self.assertEqual(admin["type"], "vnfd", "Wrong descriptor type") - self.assertEqual( - admin["created"], db_admin["created"], "Wrong creation time" - ) - self.assertGreater( - admin["modified"], db_admin["created"], "Wrong modification time" - ) - self.assertEqual( - admin["projects_read"], - db_admin["projects_read"], - "Wrong read-only project list", - ) - self.assertEqual( - admin["projects_write"], - db_admin["projects_write"], - "Wrong read-write project list", - ) - self.assertEqual( - admin["onboardingState"], "ONBOARDED", "Wrong onboarding state" - ) - self.assertEqual( - admin["operationalState"], "ENABLED", "Wrong operational state" - ) - self.assertEqual(admin["usageState"], "NOT_IN_USE", "Wrong usage state") - storage = admin["storage"] - self.assertEqual(storage["folder"], did + ":1", "Wrong storage folder") - self.assertEqual( - storage["descriptor"], "package", "Wrong storage descriptor" - ) - self.assertEqual( - admin["revision"], 1, "Wrong revision number" + self.assertEqual( + e.exception.http_code, + HTTPStatus.UNPROCESSABLE_ENTITY, + "Wrong HTTP status code", + ) + self.assertIn( + norm( + "Error in pyangbind validation: {} ({})".format( + "json object contained a key that did not exist", "extra-property" ) + ), + norm(str(e.exception)), + "Wrong exception text", + ) + db_args = self.db.replace.call_args[0] + admin = db_args[2]["_admin"] + self.assertEqual(admin["revision"], 1, "Wrong revision number") - compare_desc(self, test_vnfd, db_args[2], "VNFD") - finally: - test_vnfd["vdu"][0]["cloud-init-file"] = tmp1 - test_vnfd["df"][0]["lcm-operations-configuration"][ - "operate-vnf-op-config" - ]["day1-2"][0]["execution-environment-list"][0]["juju"] = tmp2 - self.db.get_one.side_effect = ( - lambda table, filter, fail_on_empty=None, fail_on_more=None: { - "_id": did, - "_admin": deepcopy(db_vnfd_content["_admin"]), - } - ) - with self.subTest(i=2, t="Check Pyangbind Validation: additional properties"): - test_vnfd["extra-property"] = 0 - try: - with self.assertRaises( - EngineException, msg="Accepted VNFD with an additional property" - ) as e: - self.topic.upload_content( - fake_session, did, test_vnfd, {}, {"Content-Type": []} - ) - self.assertEqual( - e.exception.http_code, - HTTPStatus.UNPROCESSABLE_ENTITY, - "Wrong HTTP status code", - ) - self.assertIn( - norm( - "Error in pyangbind validation: {} ({})".format( - "json object contained a key that did not exist", - "extra-property", - ) - ), - norm(str(e.exception)), - "Wrong exception text", - ) - db_args = self.db.replace.call_args[0] - admin = db_args[2]["_admin"] - self.assertEqual( - admin["revision"], 1, "Wrong revision number" + @patch("osm_nbi.descriptor_topics.shutil") + @patch("osm_nbi.descriptor_topics.os.rename") + def test_new_vnfd_check_pyangbind_validation_property_types( + self, mock_rename, mock_shutil + ): + did, test_vnfd = self.prepare_vnfd_creation() + test_vnfd = self.prepare_test_vnfd(test_vnfd) + test_vnfd["_id"] = did + test_vnfd["product-name"] = {"key": 0} + + with self.assertRaises( + EngineException, msg="Accepted VNFD with a wrongly typed property" + ) as e: + self.topic.upload_content( + fake_session, did, test_vnfd, {}, {"Content-Type": []} + ) + self.assertEqual( + e.exception.http_code, + HTTPStatus.UNPROCESSABLE_ENTITY, + "Wrong HTTP status code", + ) + self.assertIn( + norm( + "Error in pyangbind validation: {} ({})".format( + "json object contained a key that did not exist", "key" ) + ), + norm(str(e.exception)), + "Wrong exception text", + ) - finally: - del test_vnfd["extra-property"] - with self.subTest(i=3, t="Check Pyangbind Validation: property types"): - tmp = test_vnfd["product-name"] - test_vnfd["product-name"] = {"key": 0} - try: - with self.assertRaises( - EngineException, msg="Accepted VNFD with a wrongly typed property" - ) as e: - self.topic.upload_content( - fake_session, did, test_vnfd, {}, {"Content-Type": []} - ) - self.assertEqual( - e.exception.http_code, - HTTPStatus.UNPROCESSABLE_ENTITY, - "Wrong HTTP status code", - ) - self.assertIn( - norm( - "Error in pyangbind validation: {} ({})".format( - "json object contained a key that did not exist", "key" - ) - ), - norm(str(e.exception)), - "Wrong exception text", - ) - finally: - test_vnfd["product-name"] = tmp - with self.subTest(i=4, t="Check Input Validation: cloud-init"): - with self.assertRaises( - EngineException, msg="Accepted non-existent cloud_init file" - ) as e: - self.topic.upload_content( - fake_session, did, test_vnfd, {}, {"Content-Type": []} - ) - self.assertEqual( - e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code" + @patch("osm_nbi.descriptor_topics.shutil") + @patch("osm_nbi.descriptor_topics.os.rename") + def test_new_vnfd_check_input_validation_cloud_init(self, mock_rename, mock_shutil): + did, test_vnfd = self.prepare_vnfd_creation() + del test_vnfd["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][ + "day1-2" + ][0]["execution-environment-list"][0]["juju"] + + with self.assertRaises( + EngineException, msg="Accepted non-existent cloud_init file" + ) as e: + self.topic.upload_content( + fake_session, did, test_vnfd, {}, {"Content-Type": []} ) - self.assertIn( - norm( - "{} defined in vnf[id={}]:vdu[id={}] but not present in package".format( - "cloud-init", test_vnfd["id"], test_vnfd["vdu"][0]["id"] - ) - ), - norm(str(e.exception)), - "Wrong exception text", + self.assertEqual( + e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code" + ) + self.assertIn( + norm( + "{} defined in vnf[id={}]:vdu[id={}] but not present in package".format( + "cloud-init", test_vnfd["id"], test_vnfd["vdu"][0]["id"] + ) + ), + norm(str(e.exception)), + "Wrong exception text", + ) + + @patch("osm_nbi.descriptor_topics.shutil") + @patch("osm_nbi.descriptor_topics.os.rename") + def test_new_vnfd_check_input_validation_day12_configuration( + self, mock_rename, mock_shutil + ): + did, test_vnfd = self.prepare_vnfd_creation() + del test_vnfd["vdu"][0]["cloud-init-file"] + + with self.assertRaises( + EngineException, msg="Accepted non-existent charm in VNF configuration" + ) as e: + self.topic.upload_content( + fake_session, did, test_vnfd, {}, {"Content-Type": []} ) - with self.subTest(i=5, t="Check Input Validation: day1-2 configuration[juju]"): - del test_vnfd["vdu"][0]["cloud-init-file"] - with self.assertRaises( - EngineException, msg="Accepted non-existent charm in VNF configuration" - ) as e: - self.topic.upload_content( - fake_session, did, test_vnfd, {}, {"Content-Type": []} + self.assertEqual( + e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code" + ) + self.assertIn( + norm( + "{} defined in vnf[id={}] but not present in package".format( + "charm", test_vnfd["id"] ) - self.assertEqual( - e.exception.http_code, HTTPStatus.BAD_REQUEST, "Wrong HTTP status code" + ), + norm(str(e.exception)), + "Wrong exception text", + ) + + @patch("osm_nbi.descriptor_topics.shutil") + @patch("osm_nbi.descriptor_topics.os.rename") + def test_new_vnfd_check_input_validation_mgmt_cp(self, mock_rename, mock_shutil): + did, test_vnfd = self.prepare_vnfd_creation() + test_vnfd = self.prepare_test_vnfd(test_vnfd) + del test_vnfd["mgmt-cp"] + + with self.assertRaises( + EngineException, msg="Accepted VNFD without management interface" + ) as e: + self.topic.upload_content( + fake_session, did, test_vnfd, {}, {"Content-Type": []} ) - self.assertIn( - norm( - "{} defined in vnf[id={}] but not present in package".format( - "charm", test_vnfd["id"] - ) - ), - norm(str(e.exception)), - "Wrong exception text", + self.assertEqual( + e.exception.http_code, + HTTPStatus.UNPROCESSABLE_ENTITY, + "Wrong HTTP status code", + ) + self.assertIn( + norm("'{}' is a mandatory field and it is not defined".format("mgmt-cp")), + norm(str(e.exception)), + "Wrong exception text", + ) + + @patch("osm_nbi.descriptor_topics.shutil") + @patch("osm_nbi.descriptor_topics.os.rename") + def test_new_vnfd_check_input_validation_mgmt_cp_connection_point( + self, mock_rename, mock_shutil + ): + did, test_vnfd = self.prepare_vnfd_creation() + test_vnfd = self.prepare_test_vnfd(test_vnfd) + test_vnfd["mgmt-cp"] = "wrong-cp" + + with self.assertRaises( + EngineException, msg="Accepted wrong mgmt-cp connection point" + ) as e: + self.topic.upload_content( + fake_session, did, test_vnfd, {}, {"Content-Type": []} ) - del test_vnfd["df"][0]["lcm-operations-configuration"][ - "operate-vnf-op-config" - ]["day1-2"][0]["execution-environment-list"][0]["juju"] - with self.subTest(i=6, t="Check Input Validation: mgmt-cp"): - tmp = test_vnfd["mgmt-cp"] - del test_vnfd["mgmt-cp"] - try: - with self.assertRaises( - EngineException, msg="Accepted VNFD without management interface" - ) as e: - self.topic.upload_content( - fake_session, did, test_vnfd, {}, {"Content-Type": []} - ) - self.assertEqual( - e.exception.http_code, - HTTPStatus.UNPROCESSABLE_ENTITY, - "Wrong HTTP status code", - ) - self.assertIn( - norm( - "'{}' is a mandatory field and it is not defined".format( - "mgmt-cp" - ) - ), - norm(str(e.exception)), - "Wrong exception text", - ) - finally: - test_vnfd["mgmt-cp"] = tmp - with self.subTest(i=7, t="Check Input Validation: mgmt-cp connection point"): - tmp = test_vnfd["mgmt-cp"] - test_vnfd["mgmt-cp"] = "wrong-cp" - try: - with self.assertRaises( - EngineException, msg="Accepted wrong mgmt-cp connection point" - ) as e: - self.topic.upload_content( - fake_session, did, test_vnfd, {}, {"Content-Type": []} - ) - self.assertEqual( - e.exception.http_code, - HTTPStatus.UNPROCESSABLE_ENTITY, - "Wrong HTTP status code", - ) - self.assertIn( - norm( - "mgmt-cp='{}' must match an existing ext-cpd".format( - test_vnfd["mgmt-cp"] - ) - ), - norm(str(e.exception)), - "Wrong exception text", - ) - finally: - test_vnfd["mgmt-cp"] = tmp - with self.subTest(i=8, t="Check Input Validation: vdu int-cpd"): - ext_cpd = test_vnfd["ext-cpd"][1] - tmp = ext_cpd["int-cpd"]["cpd"] - ext_cpd["int-cpd"]["cpd"] = "wrong-cpd" - try: - with self.assertRaises( - EngineException, - msg="Accepted wrong ext-cpd internal connection point", - ) as e: - self.topic.upload_content( - fake_session, did, test_vnfd, {}, {"Content-Type": []} - ) - self.assertEqual( - e.exception.http_code, - HTTPStatus.UNPROCESSABLE_ENTITY, - "Wrong HTTP status code", - ) - self.assertIn( - norm( - "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format( - ext_cpd["id"] - ) - ), - norm(str(e.exception)), - "Wrong exception text", - ) - finally: - ext_cpd["int-cpd"]["cpd"] = tmp - with self.subTest(i=9, t="Check Input Validation: Duplicated VLD"): - test_vnfd["int-virtual-link-desc"].insert(0, {"id": "internal"}) - try: - with self.assertRaises( - EngineException, msg="Accepted duplicated VLD name" - ) as e: - self.topic.upload_content( - fake_session, did, test_vnfd, {}, {"Content-Type": []} - ) - self.assertEqual( - e.exception.http_code, - HTTPStatus.UNPROCESSABLE_ENTITY, - "Wrong HTTP status code", - ) - self.assertIn( - norm( - "identifier id '{}' is not unique".format( - test_vnfd["int-virtual-link-desc"][0]["id"] - ) - ), - norm(str(e.exception)), - "Wrong exception text", - ) - finally: - del test_vnfd["int-virtual-link-desc"][0] - with self.subTest(i=10, t="Check Input Validation: vdu int-virtual-link-desc"): - vdu = test_vnfd["vdu"][0] - int_cpd = vdu["int-cpd"][1] - tmp = int_cpd["int-virtual-link-desc"] - int_cpd["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc" - try: - with self.assertRaises( - EngineException, msg="Accepted int-virtual-link-desc" - ) as e: - self.topic.upload_content( - fake_session, did, test_vnfd, {}, {"Content-Type": []} - ) - self.assertEqual( - e.exception.http_code, - HTTPStatus.UNPROCESSABLE_ENTITY, - "Wrong HTTP status code", - ) - self.assertIn( - norm( - "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing " - "int-virtual-link-desc".format( - vdu["id"], int_cpd["id"], int_cpd["int-virtual-link-desc"] - ) - ), - norm(str(e.exception)), - "Wrong exception text", - ) - finally: - int_cpd["int-virtual-link-desc"] = tmp - with self.subTest(i=11, t="Check Input Validation: virtual-link-profile)"): - fake_ivld_profile = {"id": "fake-profile-ref", "flavour": "fake-flavour"} - df = test_vnfd["df"][0] - df["virtual-link-profile"] = [fake_ivld_profile] - try: - with self.assertRaises( - EngineException, msg="Accepted non-existent Profile Ref" - ) as e: - self.topic.upload_content( - fake_session, did, test_vnfd, {}, {"Content-Type": []} - ) - self.assertEqual( - e.exception.http_code, - HTTPStatus.UNPROCESSABLE_ENTITY, - "Wrong HTTP status code", - ) - self.assertIn( - norm( - "df[id='{}']:virtual-link-profile='{}' must match an existing " - "int-virtual-link-desc".format( - df["id"], fake_ivld_profile["id"] - ) - ), - norm(str(e.exception)), - "Wrong exception text", - ) - finally: - del df["virtual-link-profile"] - with self.subTest( - i=12, t="Check Input Validation: scaling-criteria monitoring-param-ref" - ): - vdu = test_vnfd["vdu"][1] - affected_df = test_vnfd["df"][0] - sa = affected_df["scaling-aspect"][0] - sp = sa["scaling-policy"][0] - sc = sp["scaling-criteria"][0] - tmp = vdu.pop("monitoring-parameter") - try: - with self.assertRaises( - EngineException, - msg="Accepted non-existent Scaling Group Policy Criteria", - ) as e: - self.topic.upload_content( - fake_session, did, test_vnfd, {}, {"Content-Type": []} - ) - self.assertEqual( - e.exception.http_code, - HTTPStatus.UNPROCESSABLE_ENTITY, - "Wrong HTTP status code", + self.assertEqual( + e.exception.http_code, + HTTPStatus.UNPROCESSABLE_ENTITY, + "Wrong HTTP status code", + ) + self.assertIn( + norm( + "mgmt-cp='{}' must match an existing ext-cpd".format( + test_vnfd["mgmt-cp"] ) - self.assertIn( - norm( - "df[id='{}']:scaling-aspect[id='{}']:scaling-policy" - "[name='{}']:scaling-criteria[name='{}']: " - "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format( - affected_df["id"], - sa["id"], - sp["name"], - sc["name"], - sc["vnf-monitoring-param-ref"], - ) - ), - norm(str(e.exception)), - "Wrong exception text", + ), + norm(str(e.exception)), + "Wrong exception text", + ) + + @patch("osm_nbi.descriptor_topics.shutil") + @patch("osm_nbi.descriptor_topics.os.rename") + def test_new_vnfd_check_input_validation_vdu_int_cpd( + self, mock_rename, mock_shutil + ): + did, test_vnfd = self.prepare_vnfd_creation() + test_vnfd = self.prepare_test_vnfd(test_vnfd) + ext_cpd = test_vnfd["ext-cpd"][1] + ext_cpd["int-cpd"]["cpd"] = "wrong-cpd" + + with self.assertRaises( + EngineException, msg="Accepted wrong ext-cpd internal connection point" + ) as e: + self.topic.upload_content( + fake_session, did, test_vnfd, {}, {"Content-Type": []} + ) + self.assertEqual( + e.exception.http_code, + HTTPStatus.UNPROCESSABLE_ENTITY, + "Wrong HTTP status code", + ) + self.assertIn( + norm( + "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format( + ext_cpd["id"] ) - finally: - vdu["monitoring-parameter"] = tmp - with self.subTest( - i=13, t="Check Input Validation: scaling-aspect vnf-configuration" - ): - df = test_vnfd["df"][0] - tmp = test_vnfd["df"][0]["lcm-operations-configuration"][ - "operate-vnf-op-config" - ]["day1-2"].pop() - try: - with self.assertRaises( - EngineException, - msg="Accepted non-existent Scaling Group VDU ID Reference", - ) as e: - self.topic.upload_content( - fake_session, did, test_vnfd, {}, {"Content-Type": []} - ) - self.assertEqual( - e.exception.http_code, - HTTPStatus.UNPROCESSABLE_ENTITY, - "Wrong HTTP status code", + ), + norm(str(e.exception)), + "Wrong exception text", + ) + + @patch("osm_nbi.descriptor_topics.shutil") + @patch("osm_nbi.descriptor_topics.os.rename") + def test_new_vnfd_check_input_validation_duplicated_vld( + self, mock_rename, mock_shutil + ): + did, test_vnfd = self.prepare_vnfd_creation() + test_vnfd = self.prepare_test_vnfd(test_vnfd) + test_vnfd["int-virtual-link-desc"].insert(0, {"id": "internal"}) + + with self.assertRaises( + EngineException, msg="Accepted duplicated VLD name" + ) as e: + self.topic.upload_content( + fake_session, did, test_vnfd, {}, {"Content-Type": []} + ) + self.assertEqual( + e.exception.http_code, + HTTPStatus.UNPROCESSABLE_ENTITY, + "Wrong HTTP status code", + ) + self.assertIn( + norm( + "identifier id '{}' is not unique".format( + test_vnfd["int-virtual-link-desc"][0]["id"] ) - self.assertIn( - norm( - "'day1-2 configuration' not defined in the descriptor but it is referenced " - "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format( - df["id"], df["scaling-aspect"][0]["id"] - ) - ), - norm(str(e.exception)), - "Wrong exception text", + ), + norm(str(e.exception)), + "Wrong exception text", + ) + + @patch("osm_nbi.descriptor_topics.shutil") + @patch("osm_nbi.descriptor_topics.os.rename") + def test_new_vnfd_check_input_validation_vdu_int_virtual_link_desc( + self, mock_rename, mock_shutil + ): + did, test_vnfd = self.prepare_vnfd_creation() + test_vnfd = self.prepare_test_vnfd(test_vnfd) + vdu = test_vnfd["vdu"][0] + int_cpd = vdu["int-cpd"][1] + int_cpd["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc" + + with self.assertRaises( + EngineException, msg="Accepted int-virtual-link-desc" + ) as e: + self.topic.upload_content( + fake_session, did, test_vnfd, {}, {"Content-Type": []} + ) + self.assertEqual( + e.exception.http_code, + HTTPStatus.UNPROCESSABLE_ENTITY, + "Wrong HTTP status code", + ) + self.assertIn( + norm( + "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing " + "int-virtual-link-desc".format( + vdu["id"], int_cpd["id"], int_cpd["int-virtual-link-desc"] ) - finally: - test_vnfd["df"][0]["lcm-operations-configuration"][ - "operate-vnf-op-config" - ]["day1-2"].append(tmp) - with self.subTest(i=14, t="Check Input Validation: scaling-config-action"): - df = test_vnfd["df"][0] - tmp = ( - test_vnfd["df"][0] - .get("lcm-operations-configuration") - .get("operate-vnf-op-config")["day1-2"][0]["config-primitive"] + ), + norm(str(e.exception)), + "Wrong exception text", + ) + + @patch("osm_nbi.descriptor_topics.shutil") + @patch("osm_nbi.descriptor_topics.os.rename") + def test_new_vnfd_check_input_validation_virtual_link_profile( + self, mock_rename, mock_shutil + ): + did, test_vnfd = self.prepare_vnfd_creation() + test_vnfd = self.prepare_test_vnfd(test_vnfd) + fake_ivld_profile = {"id": "fake-profile-ref", "flavour": "fake-flavour"} + df = test_vnfd["df"][0] + df["virtual-link-profile"] = [fake_ivld_profile] + + with self.assertRaises( + EngineException, msg="Accepted non-existent Profile Ref" + ) as e: + self.topic.upload_content( + fake_session, did, test_vnfd, {}, {"Content-Type": []} ) - test_vnfd["df"][0].get("lcm-operations-configuration").get( - "operate-vnf-op-config" - )["day1-2"][0]["config-primitive"] = [{"name": "wrong-primitive"}] - try: - with self.assertRaises( - EngineException, - msg="Accepted non-existent Scaling Group VDU ID Reference", - ) as e: - self.topic.upload_content( - fake_session, did, test_vnfd, {}, {"Content-Type": []} - ) - self.assertEqual( - e.exception.http_code, - HTTPStatus.UNPROCESSABLE_ENTITY, - "Wrong HTTP status code", + self.assertEqual( + e.exception.http_code, + HTTPStatus.UNPROCESSABLE_ENTITY, + "Wrong HTTP status code", + ) + self.assertIn( + norm( + "df[id='{}']:virtual-link-profile='{}' must match an existing " + "int-virtual-link-desc".format(df["id"], fake_ivld_profile["id"]) + ), + norm(str(e.exception)), + "Wrong exception text", + ) + + @patch("osm_nbi.descriptor_topics.shutil") + @patch("osm_nbi.descriptor_topics.os.rename") + def test_new_vnfd_check_input_validation_scaling_criteria_monitoring_param_ref( + self, mock_rename, mock_shutil + ): + did, test_vnfd = self.prepare_vnfd_creation() + test_vnfd = self.prepare_test_vnfd(test_vnfd) + vdu = test_vnfd["vdu"][1] + affected_df = test_vnfd["df"][0] + sa = affected_df["scaling-aspect"][0] + sp = sa["scaling-policy"][0] + sc = sp["scaling-criteria"][0] + vdu.pop("monitoring-parameter") + + with self.assertRaises( + EngineException, msg="Accepted non-existent Scaling Group Policy Criteria" + ) as e: + self.topic.upload_content( + fake_session, did, test_vnfd, {}, {"Content-Type": []} + ) + self.assertEqual( + e.exception.http_code, + HTTPStatus.UNPROCESSABLE_ENTITY, + "Wrong HTTP status code", + ) + self.assertIn( + norm( + "df[id='{}']:scaling-aspect[id='{}']:scaling-policy" + "[name='{}']:scaling-criteria[name='{}']: " + "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format( + affected_df["id"], + sa["id"], + sp["name"], + sc["name"], + sc["vnf-monitoring-param-ref"], ) - self.assertIn( - norm( - "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-" - "config-primitive-name-ref='{}' does not match any " - "day1-2 configuration:config-primitive:name".format( - df["id"], - df["scaling-aspect"][0]["id"], - sa["scaling-config-action"][0][ - "vnf-config-primitive-name-ref" - ], - ) - ), - norm(str(e.exception)), - "Wrong exception text", + ), + norm(str(e.exception)), + "Wrong exception text", + ) + + @patch("osm_nbi.descriptor_topics.shutil") + @patch("osm_nbi.descriptor_topics.os.rename") + def test_new_vnfd_check_input_validation_scaling_aspect_vnf_configuration( + self, mock_rename, mock_shutil + ): + did, test_vnfd = self.prepare_vnfd_creation() + test_vnfd = self.prepare_test_vnfd(test_vnfd) + test_vnfd["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][ + "day1-2" + ].pop() + df = test_vnfd["df"][0] + + with self.assertRaises( + EngineException, msg="Accepted non-existent Scaling Group VDU ID Reference" + ) as e: + self.topic.upload_content( + fake_session, did, test_vnfd, {}, {"Content-Type": []} + ) + self.assertEqual( + e.exception.http_code, + HTTPStatus.UNPROCESSABLE_ENTITY, + "Wrong HTTP status code", + ) + self.assertIn( + norm( + "'day1-2 configuration' not defined in the descriptor but it is referenced " + "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format( + df["id"], df["scaling-aspect"][0]["id"] ) - finally: - test_vnfd["df"][0].get("lcm-operations-configuration").get( - "operate-vnf-op-config" - )["day1-2"][0]["config-primitive"] = tmp - with self.subTest(i=15, t="Check Input Validation: everything right"): - test_vnfd["id"] = "fake-vnfd-id" - test_vnfd["df"][0].get("lcm-operations-configuration").get( - "operate-vnf-op-config" - )["day1-2"][0]["id"] = "fake-vnfd-id" - self.db.get_one.side_effect = [ - {"_id": did, "_admin": deepcopy(db_vnfd_content["_admin"])}, - None, - ] - rc = self.topic.upload_content( + ), + norm(str(e.exception)), + "Wrong exception text", + ) + + @patch("osm_nbi.descriptor_topics.shutil") + @patch("osm_nbi.descriptor_topics.os.rename") + def test_new_vnfd_check_input_validation_scaling_config_action( + self, mock_rename, mock_shutil + ): + did, test_vnfd = self.prepare_vnfd_creation() + test_vnfd = self.prepare_test_vnfd(test_vnfd) + df = test_vnfd["df"][0] + affected_df = test_vnfd["df"][0] + sa = affected_df["scaling-aspect"][0] + test_vnfd["df"][0].get("lcm-operations-configuration").get( + "operate-vnf-op-config" + )["day1-2"][0]["config-primitive"] = [{"name": "wrong-primitive"}] + + with self.assertRaises( + EngineException, msg="Accepted non-existent Scaling Group VDU ID Reference" + ) as e: + self.topic.upload_content( fake_session, did, test_vnfd, {}, {"Content-Type": []} ) - self.assertTrue(rc, "Input Validation: Unexpected failure") - return + self.assertEqual( + e.exception.http_code, + HTTPStatus.UNPROCESSABLE_ENTITY, + "Wrong HTTP status code", + ) + self.assertIn( + norm( + "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-" + "config-primitive-name-ref='{}' does not match any " + "day1-2 configuration:config-primitive:name".format( + df["id"], + df["scaling-aspect"][0]["id"], + sa["scaling-config-action"][0]["vnf-config-primitive-name-ref"], + ) + ), + norm(str(e.exception)), + "Wrong exception text", + ) + + @patch("osm_nbi.descriptor_topics.shutil") + @patch("osm_nbi.descriptor_topics.os.rename") + def test_new_vnfd_check_input_validation_everything_right( + self, mock_rename, mock_shutil + ): + did, test_vnfd = self.prepare_vnfd_creation() + test_vnfd = self.prepare_test_vnfd(test_vnfd) + test_vnfd["id"] = "fake-vnfd-id" + test_vnfd["df"][0].get("lcm-operations-configuration").get( + "operate-vnf-op-config" + )["day1-2"][0]["id"] = "fake-vnfd-id" + self.db.get_one.side_effect = [ + {"_id": did, "_admin": deepcopy(db_vnfd_content["_admin"])}, + None, + ] + rc = self.topic.upload_content( + fake_session, did, test_vnfd, {}, {"Content-Type": []} + ) + self.assertTrue(rc, "Input Validation: Unexpected failure") def test_edit_vnfd(self): vnfd_content = deepcopy(db_vnfd_content) @@ -722,15 +763,15 @@ class Test_VnfdTopic(TestCase): "Wrong DB NSD vnfd-id", ) - db_del_args = self.db.del_list.call_args[0] + self.db.del_list.call_args[0] self.assertEqual( self.db.del_list.call_args[0][0], - self.topic.topic+"_revisions", + self.topic.topic + "_revisions", "Wrong DB topic", ) self.assertEqual( - self.db.del_list.call_args[0][1]['_id']['$regex'], + self.db.del_list.call_args[0][1]["_id"]["$regex"], did, "Wrong ID for rexep delete", ) @@ -814,51 +855,53 @@ class Test_VnfdTopic(TestCase): self.fs.file_delete.assert_not_called() return + def prepare_vnfd_validation(self): + descriptor_name = "test_descriptor" + self.fs.file_open.side_effect = lambda path, mode: open( + "/tmp/" + str(uuid4()), "a+b" + ) + old_vnfd, new_vnfd = self.create_desc_temp(db_vnfd_content) + return descriptor_name, old_vnfd, new_vnfd + + @patch("osm_nbi.descriptor_topics.yaml") + def test_validate_vnfd_changes_day12_config_primitive_changed(self, mock_yaml): + descriptor_name, old_vnfd, new_vnfd = self.prepare_vnfd_validation() + new_vnfd["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][ + "day1-2" + ][0]["config-primitive"][0]["name"] = "new_action" + mock_yaml.load.side_effect = [old_vnfd, new_vnfd] + with self.assertNotRaises(EngineException): + self.topic._validate_descriptor_changes(descriptor_name, "/tmp/", "/tmp:1/") + + @patch("osm_nbi.descriptor_topics.yaml") + def test_validate_vnfd_changes_sw_version_changed(self, mock_yaml): + # old vnfd uses the default software version: 1.0 + descriptor_name, old_vnfd, new_vnfd = self.prepare_vnfd_validation() + new_vnfd["software-version"] = "1.3" + new_vnfd["sw-image-desc"][0]["name"] = "new-image" + mock_yaml.load.side_effect = [old_vnfd, new_vnfd] + with self.assertNotRaises(EngineException): + self.topic._validate_descriptor_changes(descriptor_name, "/tmp/", "/tmp:1/") + @patch("osm_nbi.descriptor_topics.yaml") - def test_validate_descriptor_changes(self, mock_yaml): - descriptor_name = "test_descriptor" - self.fs.file_open.side_effect = lambda path, mode: open( - "/tmp/" + str(uuid4()), "a+b" + def test_validate_vnfd_changes_sw_version_not_changed_mgm_cp_changed( + self, mock_yaml + ): + descriptor_name, old_vnfd, new_vnfd = self.prepare_vnfd_validation() + new_vnfd["mgmt-cp"] = "new-mgmt-cp" + mock_yaml.load.side_effect = [old_vnfd, new_vnfd] + with self.assertRaises(EngineException) as e: + self.topic._validate_descriptor_changes(descriptor_name, "/tmp/", "/tmp:1/") + self.assertEqual( + e.exception.http_code, + HTTPStatus.UNPROCESSABLE_ENTITY, + "Wrong HTTP status code", + ) + self.assertIn( + norm("there are disallowed changes in the vnf descriptor"), + norm(str(e.exception)), + "Wrong exception text", ) - with self.subTest(i=1, t="VNFD has changes in day1-2 config primitive"): - old_vnfd, new_vnfd = self.create_desc_temp(db_vnfd_content) - new_vnfd["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][ - "day1-2" - ][0]["config-primitive"][0]["name"] = "new_action" - mock_yaml.load.side_effect = [old_vnfd, new_vnfd] - with self.assertNotRaises(EngineException): - self.topic._validate_descriptor_changes( - descriptor_name, "/tmp/", "/tmp:1/" - ) - with self.subTest(i=2, t="VNFD sw version changed"): - # old vnfd uses the default software version: 1.0 - new_vnfd["software-version"] = "1.3" - new_vnfd["sw-image-desc"][0]["name"] = "new-image" - mock_yaml.load.side_effect = [old_vnfd, new_vnfd] - with self.assertNotRaises(EngineException): - self.topic._validate_descriptor_changes( - descriptor_name, "/tmp/", "/tmp:1/" - ) - with self.subTest( - i=3, t="VNFD sw version is not changed and mgmt-cp has changed" - ): - old_vnfd, new_vnfd = self.create_desc_temp(db_vnfd_content) - new_vnfd["mgmt-cp"] = "new-mgmt-cp" - mock_yaml.load.side_effect = [old_vnfd, new_vnfd] - with self.assertRaises(EngineException) as e: - self.topic._validate_descriptor_changes( - descriptor_name, "/tmp/", "/tmp:1/" - ) - self.assertEqual( - e.exception.http_code, - HTTPStatus.UNPROCESSABLE_ENTITY, - "Wrong HTTP status code", - ) - self.assertIn( - norm("there are disallowed changes in the vnf descriptor"), - norm(str(e.exception)), - "Wrong exception text", - ) def test_validate_mgmt_interface_connection_point_on_valid_descriptor(self): indata = deepcopy(db_vnfd_content) @@ -1246,8 +1289,9 @@ class Test_VnfdTopic(TestCase): rollback = [] did2, oid = self.topic.new(rollback, fake_session, {}) db_args = self.db.create.call_args[0] - self.assertEqual(db_args[1]['_admin']['revision'], 0, - "New package should be at revision 0") + self.assertEqual( + db_args[1]["_admin"]["revision"], 0, "New package should be at revision 0" + ) @patch("osm_nbi.descriptor_topics.shutil") @patch("osm_nbi.descriptor_topics.os.rename") @@ -1267,25 +1311,20 @@ class Test_VnfdTopic(TestCase): did2, oid = self.topic.new(rollback, fake_session, {}) del new_vnfd["vdu"][0]["cloud-init-file"] del new_vnfd["df"][0]["lcm-operations-configuration"]["operate-vnf-op-config"][ - "day1-2"][0]["execution-environment-list"][0]["juju"] - + "day1-2" + ][0]["execution-environment-list"][0]["juju"] - old_vnfd = { - "_id": did, - "_admin": deepcopy(db_vnfd_content["_admin"]) - } + old_vnfd = {"_id": did, "_admin": deepcopy(db_vnfd_content["_admin"])} old_vnfd["_admin"]["revision"] = old_revision - self.db.get_one.side_effect = [ - old_vnfd, - None, - ] - self.topic.upload_content( - fake_session, did, new_vnfd, {}, {"Content-Type": []} - ) + self.db.get_one.side_effect = [old_vnfd, None] + self.topic.upload_content(fake_session, did, new_vnfd, {}, {"Content-Type": []}) db_args = self.db.replace.call_args[0] - self.assertEqual(db_args[2]['_admin']['revision'], old_revision + 1, - "Revision should increment") + self.assertEqual( + db_args[2]["_admin"]["revision"], + old_revision + 1, + "Revision should increment", + ) class Test_NsdTopic(TestCase): @@ -1317,301 +1356,297 @@ class Test_NsdTopic(TestCase): new_desc = deepcopy(template) return old_desc, new_desc - @patch("osm_nbi.descriptor_topics.shutil") - @patch("osm_nbi.descriptor_topics.os.rename") - def test_new_nsd(self, mock_rename, mock_shutil): + def prepare_nsd_creation(self): + self.fs.path = "" did = db_nsd_content["_id"] self.fs.get_params.return_value = {} self.fs.file_exists.return_value = False self.fs.file_open.side_effect = lambda path, mode: open( "/tmp/" + str(uuid4()), "a+b" ) + self.db.get_one.side_effect = [ + {"_id": did, "_admin": deepcopy(db_nsd_content["_admin"])}, + None, + ] test_nsd = deepcopy(db_nsd_content) del test_nsd["_id"] del test_nsd["_admin"] - with self.subTest(i=1, t="Normal Creation"): - self.db.create.return_value = did - rollback = [] - did2, oid = self.topic.new(rollback, fake_session, {}) - db_args = self.db.create.call_args[0] - msg_args = self.msg.write.call_args[0] - self.assertEqual(len(rollback), 1, "Wrong rollback length") - self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic") - self.assertEqual(msg_args[1], "created", "Wrong message action") - self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content") - self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic") - self.assertEqual(did2, did, "Wrong DB NSD id") - self.assertIsNotNone(db_args[1]["_admin"]["created"], "Wrong creation time") - self.assertEqual( - db_args[1]["_admin"]["modified"], - db_args[1]["_admin"]["created"], - "Wrong modification time", - ) - self.assertEqual( - db_args[1]["_admin"]["projects_read"], - [test_pid], - "Wrong read-only project list", + return did, test_nsd + + @patch("osm_nbi.descriptor_topics.shutil") + @patch("osm_nbi.descriptor_topics.os.rename") + def test_new_nsd_normal_creation(self, mock_rename, mock_shutil): + did, test_nsd = self.prepare_nsd_creation() + self.db.create.return_value = did + rollback = [] + + did2, oid = self.topic.new(rollback, fake_session, {}) + db_args = self.db.create.call_args[0] + msg_args = self.msg.write.call_args[0] + self.assertEqual(len(rollback), 1, "Wrong rollback length") + self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic") + self.assertEqual(msg_args[1], "created", "Wrong message action") + self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content") + self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic") + self.assertEqual(did2, did, "Wrong DB NSD id") + self.assertIsNotNone(db_args[1]["_admin"]["created"], "Wrong creation time") + self.assertEqual( + db_args[1]["_admin"]["modified"], + db_args[1]["_admin"]["created"], + "Wrong modification time", + ) + self.assertEqual( + db_args[1]["_admin"]["projects_read"], + [test_pid], + "Wrong read-only project list", + ) + self.assertEqual( + db_args[1]["_admin"]["projects_write"], + [test_pid], + "Wrong read-write project list", + ) + + self.db.get_list.return_value = [db_vnfd_content] + + self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []}) + msg_args = self.msg.write.call_args[0] + test_nsd["_id"] = did + self.assertEqual(msg_args[0], self.topic.topic_msg, "Wrong message topic") + self.assertEqual(msg_args[1], "edited", "Wrong message action") + self.assertEqual(msg_args[2], test_nsd, "Wrong message content") + + db_args = self.db.get_one.mock_calls[0][1] + self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic") + self.assertEqual(db_args[1]["_id"], did, "Wrong DB NSD id") + + db_args = self.db.replace.call_args[0] + self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic") + self.assertEqual(db_args[1], did, "Wrong DB NSD id") + + admin = db_args[2]["_admin"] + db_admin = db_nsd_content["_admin"] + self.assertEqual(admin["created"], db_admin["created"], "Wrong creation time") + self.assertGreater( + admin["modified"], db_admin["created"], "Wrong modification time" + ) + self.assertEqual( + admin["projects_read"], + db_admin["projects_read"], + "Wrong read-only project list", + ) + self.assertEqual( + admin["projects_write"], + db_admin["projects_write"], + "Wrong read-write project list", + ) + self.assertEqual( + admin["onboardingState"], "ONBOARDED", "Wrong onboarding state" + ) + self.assertEqual( + admin["operationalState"], "ENABLED", "Wrong operational state" + ) + self.assertEqual(admin["usageState"], "NOT_IN_USE", "Wrong usage state") + + storage = admin["storage"] + self.assertEqual(storage["folder"], did + ":1", "Wrong storage folder") + self.assertEqual(storage["descriptor"], "package", "Wrong storage descriptor") + + compare_desc(self, test_nsd, db_args[2], "NSD") + revision_args = self.db.create.call_args[0] + self.assertEqual( + revision_args[0], self.topic.topic + "_revisions", "Wrong topic" + ) + self.assertEqual(revision_args[1]["id"], db_args[2]["id"], "Wrong revision id") + self.assertEqual( + revision_args[1]["_id"], db_args[2]["_id"] + ":1", "Wrong revision _id" + ) + + @patch("osm_nbi.descriptor_topics.shutil") + @patch("osm_nbi.descriptor_topics.os.rename") + def test_new_nsd_check_pyangbind_validation_required_properties( + self, mock_rename, mock_shutil + ): + did, test_nsd = self.prepare_nsd_creation() + del test_nsd["id"] + + with self.assertRaises( + EngineException, msg="Accepted NSD with a missing required property" + ) as e: + self.topic.upload_content( + fake_session, did, test_nsd, {}, {"Content-Type": []} ) - self.assertEqual( - db_args[1]["_admin"]["projects_write"], - [test_pid], - "Wrong read-write project list", + self.assertEqual( + e.exception.http_code, + HTTPStatus.UNPROCESSABLE_ENTITY, + "Wrong HTTP status code", + ) + self.assertIn( + norm("Error in pyangbind validation: '{}'".format("id")), + norm(str(e.exception)), + "Wrong exception text", + ) + + @patch("osm_nbi.descriptor_topics.shutil") + @patch("osm_nbi.descriptor_topics.os.rename") + def test_new_nsd_check_pyangbind_validation_additional_properties( + self, mock_rename, mock_shutil + ): + did, test_nsd = self.prepare_nsd_creation() + test_nsd["extra-property"] = 0 + + with self.assertRaises( + EngineException, msg="Accepted NSD with an additional property" + ) as e: + self.topic.upload_content( + fake_session, did, test_nsd, {}, {"Content-Type": []} ) - try: - self.fs.path = "" - self.db.get_one.side_effect = [ - {"_id": did, "_admin": db_nsd_content["_admin"]}, - None, - ] - self.db.get_list.return_value = [db_vnfd_content] - self.topic.upload_content( - fake_session, did, test_nsd, {}, {"Content-Type": []} - ) - msg_args = self.msg.write.call_args[0] - test_nsd["_id"] = did - self.assertEqual( - msg_args[0], self.topic.topic_msg, "Wrong message topic" - ) - self.assertEqual(msg_args[1], "edited", "Wrong message action") - self.assertEqual(msg_args[2], test_nsd, "Wrong message content") - db_args = self.db.get_one.mock_calls[0][1] - self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic") - self.assertEqual(db_args[1]["_id"], did, "Wrong DB NSD id") - db_args = self.db.replace.call_args[0] - self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic") - self.assertEqual(db_args[1], did, "Wrong DB NSD id") - admin = db_args[2]["_admin"] - db_admin = db_nsd_content["_admin"] - self.assertEqual( - admin["created"], db_admin["created"], "Wrong creation time" - ) - self.assertGreater( - admin["modified"], db_admin["created"], "Wrong modification time" - ) - self.assertEqual( - admin["projects_read"], - db_admin["projects_read"], - "Wrong read-only project list", - ) - self.assertEqual( - admin["projects_write"], - db_admin["projects_write"], - "Wrong read-write project list", - ) - self.assertEqual( - admin["onboardingState"], "ONBOARDED", "Wrong onboarding state" - ) - self.assertEqual( - admin["operationalState"], "ENABLED", "Wrong operational state" - ) - self.assertEqual(admin["usageState"], "NOT_IN_USE", "Wrong usage state") - storage = admin["storage"] - self.assertEqual(storage["folder"], did + ":1", "Wrong storage folder") - self.assertEqual( - storage["descriptor"], "package", "Wrong storage descriptor" - ) - compare_desc(self, test_nsd, db_args[2], "NSD") - revision_args = self.db.create.call_args[0] - self.assertEqual( - revision_args[0], self.topic.topic + "_revisions", "Wrong topic" - ) - self.assertEqual( - revision_args[1]["id"], db_args[2]["id"], "Wrong revision id" - ) - self.assertEqual( - revision_args[1]["_id"], - db_args[2]["_id"] + ":1", - "Wrong revision _id" + self.assertEqual( + e.exception.http_code, + HTTPStatus.UNPROCESSABLE_ENTITY, + "Wrong HTTP status code", + ) + self.assertIn( + norm( + "Error in pyangbind validation: {} ({})".format( + "json object contained a key that did not exist", "extra-property" ) + ), + norm(str(e.exception)), + "Wrong exception text", + ) - finally: - pass - self.db.get_one.side_effect = ( - lambda table, filter, fail_on_empty=None, fail_on_more=None: { - "_id": did, - "_admin": db_nsd_content["_admin"], - } - ) - with self.subTest(i=2, t="Check Pyangbind Validation: required properties"): - tmp = test_nsd["id"] - del test_nsd["id"] - try: - with self.assertRaises( - EngineException, msg="Accepted NSD with a missing required property" - ) as e: - self.topic.upload_content( - fake_session, did, test_nsd, {}, {"Content-Type": []} - ) - self.assertEqual( - e.exception.http_code, - HTTPStatus.UNPROCESSABLE_ENTITY, - "Wrong HTTP status code", - ) - self.assertIn( - norm("Error in pyangbind validation: '{}'".format("id")), - norm(str(e.exception)), - "Wrong exception text", - ) - finally: - test_nsd["id"] = tmp - with self.subTest(i=3, t="Check Pyangbind Validation: additional properties"): - test_nsd["extra-property"] = 0 - try: - with self.assertRaises( - EngineException, msg="Accepted NSD with an additional property" - ) as e: - self.topic.upload_content( - fake_session, did, test_nsd, {}, {"Content-Type": []} - ) - self.assertEqual( - e.exception.http_code, - HTTPStatus.UNPROCESSABLE_ENTITY, - "Wrong HTTP status code", - ) - self.assertIn( - norm( - "Error in pyangbind validation: {} ({})".format( - "json object contained a key that did not exist", - "extra-property", - ) - ), - norm(str(e.exception)), - "Wrong exception text", - ) - finally: - del test_nsd["extra-property"] - with self.subTest(i=4, t="Check Pyangbind Validation: property types"): - tmp = test_nsd["designer"] - test_nsd["designer"] = {"key": 0} - try: - with self.assertRaises( - EngineException, msg="Accepted NSD with a wrongly typed property" - ) as e: - self.topic.upload_content( - fake_session, did, test_nsd, {}, {"Content-Type": []} - ) - self.assertEqual( - e.exception.http_code, - HTTPStatus.UNPROCESSABLE_ENTITY, - "Wrong HTTP status code", - ) - self.assertIn( - norm( - "Error in pyangbind validation: {} ({})".format( - "json object contained a key that did not exist", "key" - ) - ), - norm(str(e.exception)), - "Wrong exception text", - ) - finally: - test_nsd["designer"] = tmp - with self.subTest( - i=5, t="Check Input Validation: mgmt-network+virtual-link-protocol-data" - ): - df = test_nsd["df"][0] - mgmt_profile = { - "id": "id", - "virtual-link-desc-id": "mgmt", - "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"}, - } - df["virtual-link-profile"] = [mgmt_profile] - try: - with self.assertRaises( - EngineException, msg="Accepted VLD with mgmt-network+ip-profile" - ) as e: - self.topic.upload_content( - fake_session, did, test_nsd, {}, {"Content-Type": []} - ) - self.assertEqual( - e.exception.http_code, - HTTPStatus.UNPROCESSABLE_ENTITY, - "Wrong HTTP status code", - ) - self.assertIn( - norm( - "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data" - " You cannot set a virtual-link-protocol-data when mgmt-network is True".format( - df["id"], mgmt_profile["id"] - ) - ), - norm(str(e.exception)), - "Wrong exception text", - ) - finally: - del df["virtual-link-profile"] - with self.subTest(i=6, t="Check Descriptor Dependencies: vnfd-id[]"): - self.db.get_one.side_effect = [ - {"_id": did, "_admin": db_nsd_content["_admin"]}, - None, - ] - self.db.get_list.return_value = [] - try: - with self.assertRaises( - EngineException, msg="Accepted wrong VNFD ID reference" - ) as e: - self.topic.upload_content( - fake_session, did, test_nsd, {}, {"Content-Type": []} - ) - self.assertEqual( - e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code" + @patch("osm_nbi.descriptor_topics.shutil") + @patch("osm_nbi.descriptor_topics.os.rename") + def test_new_nsd_check_pyangbind_validation_property_types( + self, mock_rename, mock_shutil + ): + did, test_nsd = self.prepare_nsd_creation() + test_nsd["designer"] = {"key": 0} + + with self.assertRaises( + EngineException, msg="Accepted NSD with a wrongly typed property" + ) as e: + self.topic.upload_content( + fake_session, did, test_nsd, {}, {"Content-Type": []} + ) + self.assertEqual( + e.exception.http_code, + HTTPStatus.UNPROCESSABLE_ENTITY, + "Wrong HTTP status code", + ) + self.assertIn( + norm( + "Error in pyangbind validation: {} ({})".format( + "json object contained a key that did not exist", "key" ) - self.assertIn( - norm( - "'vnfd-id'='{}' references a non existing vnfd".format( - test_nsd["vnfd-id"][0] - ) - ), - norm(str(e.exception)), - "Wrong exception text", + ), + norm(str(e.exception)), + "Wrong exception text", + ) + + @patch("osm_nbi.descriptor_topics.shutil") + @patch("osm_nbi.descriptor_topics.os.rename") + def test_new_nsd_check_input_validation_mgmt_network_virtual_link_protocol_data( + self, mock_rename, mock_shutil + ): + did, test_nsd = self.prepare_nsd_creation() + df = test_nsd["df"][0] + mgmt_profile = { + "id": "id", + "virtual-link-desc-id": "mgmt", + "virtual-link-protocol-data": {"associated-layer-protocol": "ipv4"}, + } + df["virtual-link-profile"] = [mgmt_profile] + + with self.assertRaises( + EngineException, msg="Accepted VLD with mgmt-network+ip-profile" + ) as e: + self.topic.upload_content( + fake_session, did, test_nsd, {}, {"Content-Type": []} + ) + self.assertEqual( + e.exception.http_code, + HTTPStatus.UNPROCESSABLE_ENTITY, + "Wrong HTTP status code", + ) + self.assertIn( + norm( + "Error at df[id='{}']:virtual-link-profile[id='{}']:virtual-link-protocol-data" + " You cannot set a virtual-link-protocol-data when mgmt-network is True".format( + df["id"], mgmt_profile["id"] ) - finally: - pass - with self.subTest( - i=7, - t="Check Descriptor Dependencies: " - "vld[vnfd-connection-point-ref][vnfd-connection-point-ref]", - ): - vnfd_descriptor = deepcopy(db_vnfd_content) - df = test_nsd["df"][0] - affected_vnf_profile = df["vnf-profile"][0] - affected_virtual_link = affected_vnf_profile["virtual-link-connectivity"][1] - affected_cpd = vnfd_descriptor["ext-cpd"].pop() - self.db.get_one.side_effect = [ - {"_id": did, "_admin": db_nsd_content["_admin"]}, - None, - ] - self.db.get_list.return_value = [vnfd_descriptor] - try: - with self.assertRaises( - EngineException, msg="Accepted wrong VLD CP reference" - ) as e: - self.topic.upload_content( - fake_session, did, test_nsd, {}, {"Content-Type": []} - ) - self.assertEqual( - e.exception.http_code, - HTTPStatus.UNPROCESSABLE_ENTITY, - "Wrong HTTP status code", + ), + norm(str(e.exception)), + "Wrong exception text", + ) + + @patch("osm_nbi.descriptor_topics.shutil") + @patch("osm_nbi.descriptor_topics.os.rename") + def test_new_nsd_check_descriptor_dependencies_vnfd_id( + self, mock_rename, mock_shutil + ): + did, test_nsd = self.prepare_nsd_creation() + self.db.get_list.return_value = [] + + with self.assertRaises( + EngineException, msg="Accepted wrong VNFD ID reference" + ) as e: + self.topic.upload_content( + fake_session, did, test_nsd, {}, {"Content-Type": []} + ) + self.assertEqual( + e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code" + ) + self.assertIn( + norm( + "'vnfd-id'='{}' references a non existing vnfd".format( + test_nsd["vnfd-id"][0] ) - self.assertIn( - norm( - "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity" - "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a " - "non existing ext-cpd:id inside vnfd '{}'".format( - df["id"], - affected_vnf_profile["id"], - affected_virtual_link["virtual-link-profile-id"], - affected_cpd["id"], - vnfd_descriptor["id"], - ) - ), - norm(str(e.exception)), - "Wrong exception text", + ), + norm(str(e.exception)), + "Wrong exception text", + ) + + @patch("osm_nbi.descriptor_topics.shutil") + @patch("osm_nbi.descriptor_topics.os.rename") + def test_new_nsd_check_descriptor_dependencies_vld_vnfd_connection_point_ref( + self, mock_rename, mock_shutil + ): + # Check Descriptor Dependencies: "vld[vnfd-connection-point-ref][vnfd-connection-point-ref] + did, test_nsd = self.prepare_nsd_creation() + vnfd_descriptor = deepcopy(db_vnfd_content) + df = test_nsd["df"][0] + affected_vnf_profile = df["vnf-profile"][0] + affected_virtual_link = affected_vnf_profile["virtual-link-connectivity"][1] + affected_cpd = vnfd_descriptor["ext-cpd"].pop() + self.db.get_list.return_value = [vnfd_descriptor] + + with self.assertRaises( + EngineException, msg="Accepted wrong VLD CP reference" + ) as e: + self.topic.upload_content( + fake_session, did, test_nsd, {}, {"Content-Type": []} + ) + self.assertEqual( + e.exception.http_code, + HTTPStatus.UNPROCESSABLE_ENTITY, + "Wrong HTTP status code", + ) + self.assertIn( + norm( + "Error at df[id='{}']:vnf-profile[id='{}']:virtual-link-connectivity" + "[virtual-link-profile-id='{}']:constituent-cpd-id='{}' references a " + "non existing ext-cpd:id inside vnfd '{}'".format( + df["id"], + affected_vnf_profile["id"], + affected_virtual_link["virtual-link-profile-id"], + affected_cpd["id"], + vnfd_descriptor["id"], ) - finally: - pass - return + ), + norm(str(e.exception)), + "Wrong exception text", + ) def test_edit_nsd(self): nsd_content = deepcopy(db_nsd_content) @@ -1689,6 +1724,7 @@ class Test_NsdTopic(TestCase): norm(str(e.exception)), "Wrong exception text", ) + self.db.reset_mock() return def test_delete_nsd(self): @@ -1800,47 +1836,51 @@ class Test_NsdTopic(TestCase): "Wrong DB pull_list dictionary", ) self.fs.file_delete.assert_not_called() + self.db.reset_mock() return - @patch("osm_nbi.descriptor_topics.yaml") - def test_validate_descriptor_changes(self, mock_yaml): + def prepare_nsd_validation(self): descriptor_name = "test_ns_descriptor" self.fs.file_open.side_effect = lambda path, mode: open( "/tmp/" + str(uuid4()), "a+b" ) - with self.subTest( - i=1, t="NSD has changes in ns-configuration:config-primitive" - ): - old_nsd, new_nsd = self.create_desc_temp(db_nsd_content) - old_nsd.update( - {"ns-configuration": {"config-primitive": [{"name": "add-user"}]}} - ) - new_nsd.update( - {"ns-configuration": {"config-primitive": [{"name": "del-user"}]}} - ) - mock_yaml.load.side_effect = [old_nsd, new_nsd] - with self.assertNotRaises(EngineException): - self.topic._validate_descriptor_changes( - descriptor_name, "/tmp", "/tmp:1" - ) - with self.subTest(i=2, t="NSD name has changed"): - old_nsd, new_nsd = self.create_desc_temp(db_nsd_content) - new_nsd["name"] = "nscharm-ns2" - mock_yaml.load.side_effect = [old_nsd, new_nsd] - with self.assertRaises(EngineException) as e: - self.topic._validate_descriptor_changes( - descriptor_name, "/tmp", "/tmp:1" - ) - self.assertEqual( - e.exception.http_code, - HTTPStatus.UNPROCESSABLE_ENTITY, - "Wrong HTTP status code", - ) - self.assertIn( - norm("there are disallowed changes in the ns descriptor"), - norm(str(e.exception)), - "Wrong exception text", - ) + old_nsd, new_nsd = self.create_desc_temp(db_nsd_content) + return descriptor_name, old_nsd, new_nsd + + @patch("osm_nbi.descriptor_topics.yaml") + def test_validate_descriptor_ns_configuration_changed(self, mock_yaml): + # NSD has changes in ns-configuration:config-primitive + descriptor_name, old_nsd, new_nsd = self.prepare_nsd_validation() + old_nsd.update( + {"ns-configuration": {"config-primitive": [{"name": "add-user"}]}} + ) + new_nsd.update( + {"ns-configuration": {"config-primitive": [{"name": "del-user"}]}} + ) + mock_yaml.load.side_effect = [old_nsd, new_nsd] + + with self.assertNotRaises(EngineException): + self.topic._validate_descriptor_changes(descriptor_name, "/tmp", "/tmp:1") + + @patch("osm_nbi.descriptor_topics.yaml") + def test_validate_descriptor_nsd_name_changed(self, mock_yaml): + descriptor_name, old_nsd, new_nsd = self.prepare_nsd_validation() + new_nsd["name"] = "nscharm-ns2" + mock_yaml.load.side_effect = [old_nsd, new_nsd] + + with self.assertRaises(EngineException) as e: + self.topic._validate_descriptor_changes(descriptor_name, "/tmp", "/tmp:1") + + self.assertEqual( + e.exception.http_code, + HTTPStatus.UNPROCESSABLE_ENTITY, + "Wrong HTTP status code", + ) + self.assertIn( + norm("there are disallowed changes in the ns descriptor"), + norm(str(e.exception)), + "Wrong exception text", + ) def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_on_valid_descriptor( self, diff --git a/requirements-test.in b/requirements-test.in index 99848ac..e9c47c6 100644 --- a/requirements-test.in +++ b/requirements-test.in @@ -17,5 +17,5 @@ asynctest coverage deepdiff nose2 -requests==2.25.1 +requests pyang diff --git a/requirements-test.txt b/requirements-test.txt index fbbf879..24cfd87 100644 --- a/requirements-test.txt +++ b/requirements-test.txt @@ -38,7 +38,7 @@ coverage==6.4.1 # nose2 deepdiff==5.8.1 # via -r requirements-test.in -idna==2.10 +idna==3.3 # via # requests # yarl @@ -54,7 +54,7 @@ ordered-set==4.1.0 # via deepdiff pyang==2.5.3 # via -r requirements-test.in -requests==2.25.1 +requests==2.28.0 # via -r requirements-test.in six==1.16.0 # via nose2 -- 2.17.1