X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;ds=sidebyside;f=osm_nbi%2Ftests%2Ftest_instance_topics.py;h=7623e8e2631cb42caeb4c908115133bef9dc9f18;hb=6380e7cbe3ffd525453b5d8ef54c18a9cc14fa5d;hp=cbb80ef7b965886e9e1159b441575e11de65b363;hpb=cee2ebfe3b4a0e4fe7a566eeb6e3f959649e1fce;p=osm%2FNBI.git diff --git a/osm_nbi/tests/test_instance_topics.py b/osm_nbi/tests/test_instance_topics.py index cbb80ef..7623e8e 100644 --- a/osm_nbi/tests/test_instance_topics.py +++ b/osm_nbi/tests/test_instance_topics.py @@ -15,6 +15,7 @@ # contact: esousa@whitestack.com or alfonso.tiernosepulveda@telefonica.com ## +from contextlib import contextmanager import unittest from time import time from unittest.mock import Mock, mock_open # patch, MagicMock @@ -56,7 +57,6 @@ class TestNsLcmOpTopic(unittest.TestCase): self.db.create_list("vnfrs", yaml.load(db_vnfrs_text, Loader=yaml.Loader)) self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader)) self.db.create = Mock(return_value="created_id") - self.db.set_one = Mock(return_value={"updated": 1}) self.nsd = self.db.get_list("nsds")[0] self.nsd_id = self.nsd["_id"] self.nsr = self.db.get_list("nsrs")[0] @@ -67,6 +67,7 @@ class TestNsLcmOpTopic(unittest.TestCase): self.vim_id = self.vim["_id"] def test_create_instantiate(self): + self.db.set_one = Mock(return_value={"updated": 1}) session = { "force": False, "admin": False, @@ -227,6 +228,59 @@ class TestNsLcmOpTopic(unittest.TestCase): "Engine exception bad http_code with {}".format(indata_copy), ) + def test_update_remove_vnf(self): + vnfr_id = self.db.get_list("vnfrs")[0]["_id"] + session = {} + self.db.set_one( + "nsrs", + {"_id": self.nsr_id}, + {"_admin.nsState": "INSTANTIATED"}, + ) + indata = { + "lcmOperationType": "update", + "updateType": "REMOVE_VNF", + "nsInstanceId": self.nsr_id, + "removeVnfInstanceId": vnfr_id + } + + session = { + "force": False, + "admin": False, + "public": False, + "project_id": [self.nsr_project], + "method": "write", + } + rollback = [] + headers = {} + + nslcmop_id, _ = self.nslcmop_topic.new( + rollback, session, indata, kwargs=None, headers=headers + ) + + self.assertEqual( + self.db.create.call_count, + 1, + "database create not called, or called more than once", + ) + _call = self.db.create.call_args_list[0] + self.assertEqual( + _call[0][0], "nslcmops", "nslcmops entry must be created at database" + ) + created_nslcmop = _call[0][1] + self.assertEqual( + self.nsr_id, + created_nslcmop["nsInstanceId"], + "mismatch between nsId '_id' in created nslcmop and database nsr", + ) + self.assertTrue( + created_nslcmop["lcmOperationType"] == "update", + "Database record must contain 'lcmOperationType=update'", + ) + self.assertTrue( + created_nslcmop["operationParams"]["updateType"] == "REMOVE_VNF", + "Database record must contain 'updateType=REMOVE_VNF'", + ) + class TestNsLcmOpTopicWithMock(unittest.TestCase): def setUp(self): @@ -255,6 +309,120 @@ class TestNsLcmOpTopicWithMock(unittest.TestCase): self.assertEqual(self.db.get_one.call_args_list[0][0][0], 'vnfrs', "Incorrect first DB lookup") self.assertEqual(self.db.get_one.call_args_list[1][0][0], 'vnfds_revisions', "Incorrect second DB lookup") + @contextmanager + def assertNotRaises(self, exception_type): + try: + yield None + except exception_type: + raise self.failureException("{} raised".format(exception_type.__name__)) + + def test_check_ns_update_operation(self): + self.db = DbMemory() + self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None) + session = {} + + with self.subTest(i=1, t="VNF instance does not belong to NS"): + test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader) + test_vnfr[0]["revision"] = 2 + test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader) + test_nsr[0]["constituent-vnfr-ref"][ + 0 + ] = "99d90b0c-faff-4b9f-bccd-017f33985984" + self.db.create_list("vnfrs", test_vnfr) + self.db.create_list("nsrs", test_nsr) + nsrs = self.db.get_list("nsrs")[0] + indata = { + "updateType": "CHANGE_VNFPKG", + "changeVnfPackageData": { + "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984", + "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77", + }, + "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85", + } + with self.assertRaises(EngineException) as expected_exception: + self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata) + self.assertEqual( + str(expected_exception.exception), + "Error in validating ns-update request: vnf 88d90b0c-faff-4b9f-bccd-017f33985984" + " does not belong to NS f48163a6-c807-47bc-9682-f72caef5af85", + ) + + with self.subTest(i=2, t="Ns update request validated with no exception"): + test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader) + test_vnfr[0]["revision"] = 2 + test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader) + self.db.create_list("vnfrs", test_vnfr) + self.db.create_list("nsrs", test_nsr) + nsrs = self.db.get_list("nsrs")[1] + indata = { + "updateType": "CHANGE_VNFPKG", + "changeVnfPackageData": { + "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984", + "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77", + }, + "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85", + } + with self.assertNotRaises(EngineException): + self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata) + + with self.subTest( + i=3, t="Ns update request rejected because of too small timeout" + ): + indata = { + "updateType": "CHANGE_VNFPKG", + "changeVnfPackageData": { + "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984", + "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77", + }, + "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85", + "timeout_ns_update": 50, + } + with self.assertRaises(EngineException) as expected_exception: + self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata) + self.assertEqual( + str(expected_exception.exception), + "Error in validating ns-update request: 50 second is not enough " + "to upgrade the VNF instance: 88d90b0c-faff-4b9f-bccd-017f33985984", + ) + + with self.subTest(i=4, t="wrong vnfdid is given as an update parameter"): + test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader) + test_vnfr[0]["revision"] = 2 + test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader) + self.db.create_list("vnfrs", test_vnfr) + self.db.create_list("nsrs", test_nsr) + nsrs = self.db.get_list("nsrs")[2] + indata = { + "updateType": "CHANGE_VNFPKG", + "changeVnfPackageData": { + "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984", + "vnfdId": "9637bcf8-cf14-42dc-ad70-c66fcf1e6e77", + }, + "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85", + } + with self.assertRaises(EngineException) as expected_exception: + self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata) + self.assertEqual( + str(expected_exception.exception), + "Error in validating ns-update request: vnfd-id 9637bcf8-cf14-42dc-ad70-c66fcf1e6e77 does not " + "match with the vnfd-id: 7637bcf8-cf14-42dc-ad70-c66fcf1e6e77 of " + "VNF instance: 88d90b0c-faff-4b9f-bccd-017f33985984", + ) + + with self.subTest(i=5, t="Ns update REMOVE_VNF request validated with no exception"): + test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader) + test_vnfr[0]["revision"] = 2 + test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader) + self.db.create_list("vnfrs", test_vnfr) + self.db.create_list("nsrs", test_nsr) + nsrs = self.db.get_list("nsrs")[1] + indata = { + "updateType": "REMOVE_VNF", + "removeVnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984", + "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85", + } + with self.assertNotRaises(EngineException): + self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata) class TestNsrTopic(unittest.TestCase): def setUp(self):