Feature 10908: Update Charms in Running VNF instances

[osm/NBI.git] / osm_nbi / tests / test_instance_topics.py

diff --git a/osm_nbi/tests/test_instance_topics.py b/osm_nbi/tests/test_instance_topics.py
index cbb80ef..b05ee0c 100644 (file)
--- 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
@@ -255,6 +256,106 @@ 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",
+            )
+
 
 class TestNsrTopic(unittest.TestCase):
     def setUp(self):