X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_nbi%2Ftests%2Ftest_descriptor_topics.py;h=8ab74b8699fa92b007cabd577e7f7acee953a365;hb=10f814e5c393778cbad66346f79208b4c70be2ff;hp=27c0bd3d9a1ffb58deb824d404d6a4a1b282c8bb;hpb=6f18ee4526bd33b54ed8da49dcb62abcf935c905;p=osm%2FNBI.git

diff --git a/osm_nbi/tests/test_descriptor_topics.py b/osm_nbi/tests/test_descriptor_topics.py
index 27c0bd3..8ab74b8 100755
--- a/osm_nbi/tests/test_descriptor_topics.py
+++ b/osm_nbi/tests/test_descriptor_topics.py
@@ -27,15 +27,26 @@ from copy import deepcopy
 from time import time
 from osm_common import dbbase, fsbase, msgbase
 from osm_nbi import authconn
-from osm_nbi.tests.test_pkg_descriptors import db_vnfds_text, db_nsds_text
+from osm_nbi.tests.test_pkg_descriptors import (
+    db_vnfds_text,
+    db_nsds_text,
+    vnfd_exploit_text,
+    vnfd_exploit_fixed_text,
+    db_sfc_nsds_text,
+)
 from osm_nbi.descriptor_topics import VnfdTopic, NsdTopic
 from osm_nbi.engine import EngineException
 from osm_common.dbbase import DbException
 import yaml
+import tempfile
+import collections
+import collections.abc
+
+collections.MutableSequence = collections.abc.MutableSequence
 
 test_name = "test-user"
-db_vnfd_content = yaml.load(db_vnfds_text, Loader=yaml.Loader)[0]
-db_nsd_content = yaml.load(db_nsds_text, Loader=yaml.Loader)[0]
+db_vnfd_content = yaml.safe_load(db_vnfds_text)[0]
+db_nsd_content = yaml.safe_load(db_nsds_text)[0]
 test_pid = db_vnfd_content["_admin"]["projects_read"][0]
 fake_session = {
     "username": test_name,
@@ -46,11 +57,23 @@ fake_session = {
     "public": False,
     "allow_show_user_project_role": True,
 }
+UUID = "00000000-0000-0000-0000-000000000000"
+
+
+def admin_value():
+    return {"projects_read": []}
+
 
+def setup_mock_fs(fs):
+    fs.path = ""
+    fs.get_params.return_value = {}
+    fs.file_exists.return_value = False
+    fs.file_open.side_effect = lambda path, mode: tempfile.TemporaryFile(mode="a+b")
 
-def norm(str):
+
+def norm(s: str):
     """Normalize string for checking"""
-    return " ".join(str.strip().split()).lower()
+    return " ".join(s.strip().split()).lower()
 
 
 def compare_desc(tc, d1, d2, k):
@@ -94,7 +117,7 @@ class Test_VnfdTopic(TestCase):
         self.topic.check_quota = Mock(return_value=None)  # skip quota
 
     @contextmanager
-    def assertNotRaises(self, exception_type):
+    def assertNotRaises(self, exception_type=Exception):
         try:
             yield None
         except exception_type:
@@ -106,12 +129,7 @@ class Test_VnfdTopic(TestCase):
         return old_desc, new_desc
 
     def prepare_vnfd_creation(self):
-        self.fs.path = ""
-        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"
-        )
+        setup_mock_fs(self.fs)
         test_vnfd = deepcopy(db_vnfd_content)
         did = db_vnfd_content["_id"]
         self.db.create.return_value = did
@@ -121,6 +139,16 @@ class Test_VnfdTopic(TestCase):
         ]
         return did, test_vnfd
 
+    def prepare_vnfd(self, vnfd_text):
+        setup_mock_fs(self.fs)
+        test_vnfd = yaml.safe_load(vnfd_text)
+        self.db.create.return_value = UUID
+        self.db.get_one.side_effect = [
+            {"_id": UUID, "_admin": admin_value()},
+            None,
+        ]
+        return UUID, test_vnfd
+
     def prepare_test_vnfd(self, test_vnfd):
         del test_vnfd["_id"]
         del test_vnfd["_admin"]
@@ -216,6 +244,26 @@ class Test_VnfdTopic(TestCase):
         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_exploit(self, mock_rename, mock_shutil):
+        id, test_vnfd = self.prepare_vnfd(vnfd_exploit_text)
+
+        with self.assertRaises(EngineException):
+            self.topic.upload_content(
+                fake_session, id, test_vnfd, {}, {"Content-Type": []}
+            )
+
+    @patch("osm_nbi.descriptor_topics.shutil")
+    @patch("osm_nbi.descriptor_topics.os.rename")
+    def test_new_vnfd_valid_helm_chart(self, mock_rename, mock_shutil):
+        id, test_vnfd = self.prepare_vnfd(vnfd_exploit_fixed_text)
+
+        with self.assertNotRaises():
+            self.topic.upload_content(
+                fake_session, id, test_vnfd, {}, {"Content-Type": []}
+            )
+
     @patch("osm_nbi.descriptor_topics.shutil")
     @patch("osm_nbi.descriptor_topics.os.rename")
     def test_new_vnfd_check_pyangbind_validation_additional_properties(
@@ -228,10 +276,12 @@ class Test_VnfdTopic(TestCase):
         )
         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"]),
-        }
+        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"
@@ -401,6 +451,8 @@ class Test_VnfdTopic(TestCase):
     def test_new_vnfd_check_input_validation_vdu_int_cpd(
         self, mock_rename, mock_shutil
     ):
+        """Testing input validation during new vnfd creation
+        for vdu internal connection point"""
         did, test_vnfd = self.prepare_vnfd_creation()
         test_vnfd = self.prepare_test_vnfd(test_vnfd)
         ext_cpd = test_vnfd["ext-cpd"][1]
@@ -432,6 +484,8 @@ class Test_VnfdTopic(TestCase):
     def test_new_vnfd_check_input_validation_duplicated_vld(
         self, mock_rename, mock_shutil
     ):
+        """Testing input validation during new vnfd creation
+        for dublicated virtual link description"""
         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"})
@@ -462,6 +516,8 @@ class Test_VnfdTopic(TestCase):
     def test_new_vnfd_check_input_validation_vdu_int_virtual_link_desc(
         self, mock_rename, mock_shutil
     ):
+        """Testing input validation during new vnfd creation
+        for vdu internal virtual link description"""
         did, test_vnfd = self.prepare_vnfd_creation()
         test_vnfd = self.prepare_test_vnfd(test_vnfd)
         vdu = test_vnfd["vdu"][0]
@@ -495,6 +551,8 @@ class Test_VnfdTopic(TestCase):
     def test_new_vnfd_check_input_validation_virtual_link_profile(
         self, mock_rename, mock_shutil
     ):
+        """Testing input validation during new vnfd creation
+        for virtual link profile"""
         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"}
@@ -521,11 +579,56 @@ class Test_VnfdTopic(TestCase):
             "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_vdu_id(
+        self, mock_rename, mock_shutil
+    ):
+        """Testing input validation during new vnfd creation
+        for scaling criteria with invalid vdu-id"""
+        did, test_vnfd = self.prepare_vnfd_creation()
+        test_vnfd = self.prepare_test_vnfd(test_vnfd)
+        test_vnfd["df"][0]["scaling-aspect"][0]["aspect-delta-details"]["deltas"][0][
+            "vdu-delta"
+        ][0]["id"] = "vdudelta1"
+        affected_df = test_vnfd["df"][0]
+        sa = affected_df["scaling-aspect"][0]
+        delta = sa["aspect-delta-details"]["deltas"][0]
+        vdu_delta = delta["vdu-delta"][0]
+
+        with self.assertRaises(
+            EngineException, msg="Accepted invalid 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='{}']:aspect-delta-details"
+                "[delta='{}']: "
+                "vdu-id='{}' not defined in vdu".format(
+                    affected_df["id"],
+                    sa["id"],
+                    delta["id"],
+                    vdu_delta["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
     ):
+        """Testing input validation during new vnfd creation
+        for scaling criteria without monitoring parameter"""
         did, test_vnfd = self.prepare_vnfd_creation()
         test_vnfd = self.prepare_test_vnfd(test_vnfd)
         vdu = test_vnfd["vdu"][1]
@@ -567,6 +670,8 @@ class Test_VnfdTopic(TestCase):
     def test_new_vnfd_check_input_validation_scaling_aspect_vnf_configuration(
         self, mock_rename, mock_shutil
     ):
+        """Testing input validation during new vnfd creation
+        for scaling criteria without day12 configuration"""
         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"][
@@ -601,6 +706,8 @@ class Test_VnfdTopic(TestCase):
     def test_new_vnfd_check_input_validation_scaling_config_action(
         self, mock_rename, mock_shutil
     ):
+        """Testing input validation during new vnfd creation
+        for scaling criteria wrong config primitive"""
         did, test_vnfd = self.prepare_vnfd_creation()
         test_vnfd = self.prepare_test_vnfd(test_vnfd)
         df = test_vnfd["df"][0]
@@ -635,11 +742,164 @@ class Test_VnfdTopic(TestCase):
             "Wrong exception text",
         )
 
+    @patch("osm_nbi.descriptor_topics.shutil")
+    @patch("osm_nbi.descriptor_topics.os.rename")
+    def test_new_vnfd_check_input_validation_healing_criteria_vdu_id(
+        self, mock_rename, mock_shutil
+    ):
+        """Testing input validation during new vnfd creation
+        for healing criteria with invalid vdu-id"""
+        did, test_vnfd = self.prepare_vnfd_creation()
+        test_vnfd = self.prepare_test_vnfd(test_vnfd)
+        test_vnfd["df"][0]["healing-aspect"][0]["healing-policy"][0][
+            "vdu-id"
+        ] = "vduid1"
+        affected_df = test_vnfd["df"][0]
+        ha = affected_df["healing-aspect"][0]
+        hp = ha["healing-policy"][0]
+        hp_vdu_id = hp["vdu-id"]
+
+        with self.assertRaises(
+            EngineException, msg="Accepted invalid Healing 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='{}']:healing-aspect[id='{}']:healing-policy"
+                "[name='{}']: "
+                "vdu-id='{}' not defined in vdu".format(
+                    affected_df["id"],
+                    ha["id"],
+                    hp["event-name"],
+                    hp_vdu_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_alarm_criteria_monitoring_param_ref(
+        self, mock_rename, mock_shutil
+    ):
+        """Testing input validation during new vnfd creation
+        for alarm with invalid monitoring parameter reference"""
+        did, test_vnfd = self.prepare_vnfd_creation()
+        test_vnfd = self.prepare_test_vnfd(test_vnfd)
+        test_vnfd["vdu"][1]["alarm"][0]["vnf-monitoring-param-ref"] = "unit_test_alarm"
+        vdu = test_vnfd["vdu"][1]
+        alarm = vdu["alarm"][0]
+        alarm_monitoring_param = alarm["vnf-monitoring-param-ref"]
+
+        with self.assertRaises(
+            EngineException, msg="Accepted invalid Alarm 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(
+                "vdu[id='{}']:alarm[id='{}']:"
+                "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
+                    vdu["id"],
+                    alarm["alarm-id"],
+                    alarm_monitoring_param,
+                )
+            ),
+            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_storage_reference_criteria(
+        self, mock_rename, mock_shutil
+    ):
+        """Testing input validation during new vnfd creation
+        for invalid virtual-storge-desc reference"""
+        did, test_vnfd = self.prepare_vnfd_creation()
+        test_vnfd = self.prepare_test_vnfd(test_vnfd)
+        test_vnfd["vdu"][1]["virtual-storage-desc"] = "unit_test_storage"
+        vdu = test_vnfd["vdu"][1]
+        vsd_ref = vdu["virtual-storage-desc"]
+
+        with self.assertRaises(
+            EngineException, msg="Accepted invalid virtual-storage-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[virtual-storage-desc='{}']"
+                "not defined in vnfd".format(
+                    vsd_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_compute_reference_criteria(
+        self, mock_rename, mock_shutil
+    ):
+        """Testing input validation during new vnfd creation
+        for invalid virtual-compute-desc reference"""
+        did, test_vnfd = self.prepare_vnfd_creation()
+        test_vnfd = self.prepare_test_vnfd(test_vnfd)
+        test_vnfd["vdu"][1]["virtual-compute-desc"] = "unit_test_compute"
+        vdu = test_vnfd["vdu"][1]
+        vcd_ref = vdu["virtual-compute-desc"]
+
+        with self.assertRaises(
+            EngineException, msg="Accepted invalid virtual-compute-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[virtual-compute-desc='{}']"
+                "not defined in vnfd".format(
+                    vcd_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
     ):
+        """Testing input validation during new vnfd creation
+        everything correct"""
         did, test_vnfd = self.prepare_vnfd_creation()
         test_vnfd = self.prepare_test_vnfd(test_vnfd)
         test_vnfd["id"] = "fake-vnfd-id"
@@ -763,7 +1023,6 @@ class Test_VnfdTopic(TestCase):
                 "Wrong DB NSD vnfd-id",
             )
 
-            self.db.del_list.call_args[0]
             self.assertEqual(
                 self.db.del_list.call_args[0][0],
                 self.topic.topic + "_revisions",
@@ -863,35 +1122,73 @@ class Test_VnfdTopic(TestCase):
         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):
+    @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
+    @patch("osm_nbi.descriptor_topics.yaml.safe_load")
+    def test_validate_vnfd_changes_day12_config_primitive_changed(
+        self, mock_safe_load, mock_detect_usage
+    ):
+        """Validating VNFD for VNFD updates, day1-2 config primitive has changed"""
         descriptor_name, old_vnfd, new_vnfd = self.prepare_vnfd_validation()
+        did = old_vnfd["_id"]
         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/")
+        mock_safe_load.side_effect = [old_vnfd, new_vnfd]
+        mock_detect_usage.return_value = True
+        self.db.get_one.return_value = old_vnfd
 
-    @patch("osm_nbi.descriptor_topics.yaml")
-    def test_validate_vnfd_changes_sw_version_changed(self, mock_yaml):
+        with self.assertNotRaises(EngineException):
+            self.topic._validate_descriptor_changes(
+                did, descriptor_name, "/tmp/", "/tmp:1/"
+            )
+        self.db.get_one.assert_called_once()
+        mock_detect_usage.assert_called_once()
+        self.assertEqual(mock_safe_load.call_count, 2)
+
+    @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
+    @patch("osm_nbi.descriptor_topics.yaml.safe_load")
+    def test_validate_vnfd_changes_sw_version_changed(
+        self, mock_safe_load, mock_detect_usage
+    ):
+        """Validating VNFD for updates, software version has changed"""
         # old vnfd uses the default software version: 1.0
         descriptor_name, old_vnfd, new_vnfd = self.prepare_vnfd_validation()
+        did = old_vnfd["_id"]
         new_vnfd["software-version"] = "1.3"
         new_vnfd["sw-image-desc"][0]["name"] = "new-image"
-        mock_yaml.load.side_effect = [old_vnfd, new_vnfd]
+        mock_safe_load.side_effect = [old_vnfd, new_vnfd]
+        mock_detect_usage.return_value = True
+        self.db.get_one.return_value = old_vnfd
+
         with self.assertNotRaises(EngineException):
-            self.topic._validate_descriptor_changes(descriptor_name, "/tmp/", "/tmp:1/")
+            self.topic._validate_descriptor_changes(
+                did, descriptor_name, "/tmp/", "/tmp:1/"
+            )
+        self.db.get_one.assert_called_once()
+        mock_detect_usage.assert_called_once()
+        self.assertEqual(mock_safe_load.call_count, 2)
 
-    @patch("osm_nbi.descriptor_topics.yaml")
+    @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
+    @patch("osm_nbi.descriptor_topics.yaml.safe_load")
     def test_validate_vnfd_changes_sw_version_not_changed_mgm_cp_changed(
-        self, mock_yaml
+        self, mock_safe_load, mock_detect_usage
     ):
+        """Validating VNFD for updates, software version has not
+        changed, mgmt-cp has changed."""
         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/")
+        mock_safe_load.side_effect = [old_vnfd, new_vnfd]
+        did = old_vnfd["_id"]
+        mock_detect_usage.return_value = True
+        self.db.get_one.return_value = old_vnfd
+
+        with self.assertRaises(
+            EngineException, msg="there are disallowed changes in the vnf descriptor"
+        ) as e:
+            self.topic._validate_descriptor_changes(
+                did, descriptor_name, "/tmp/", "/tmp:1/"
+            )
+
         self.assertEqual(
             e.exception.http_code,
             HTTPStatus.UNPROCESSABLE_ENTITY,
@@ -902,6 +1199,32 @@ class Test_VnfdTopic(TestCase):
             norm(str(e.exception)),
             "Wrong exception text",
         )
+        self.db.get_one.assert_called_once()
+        mock_detect_usage.assert_called_once()
+        self.assertEqual(mock_safe_load.call_count, 2)
+
+    @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
+    @patch("osm_nbi.descriptor_topics.yaml.safe_load")
+    def test_validate_vnfd_changes_sw_version_not_changed_mgm_cp_changed_vnfd_not_in_use(
+        self, mock_safe_load, mock_detect_usage
+    ):
+        """Validating VNFD for updates, software version has not
+        changed, mgmt-cp has changed, vnfd is not in use."""
+        descriptor_name, old_vnfd, new_vnfd = self.prepare_vnfd_validation()
+        new_vnfd["mgmt-cp"] = "new-mgmt-cp"
+        mock_safe_load.side_effect = [old_vnfd, new_vnfd]
+        did = old_vnfd["_id"]
+        mock_detect_usage.return_value = None
+        self.db.get_one.return_value = old_vnfd
+
+        with self.assertNotRaises(EngineException):
+            self.topic._validate_descriptor_changes(
+                did, descriptor_name, "/tmp/", "/tmp:1/"
+            )
+
+        self.db.get_one.assert_called_once()
+        mock_detect_usage.assert_called_once()
+        mock_safe_load.assert_not_called()
 
     def test_validate_mgmt_interface_connection_point_on_valid_descriptor(self):
         indata = deepcopy(db_vnfd_content)
@@ -1316,7 +1639,8 @@ class Test_VnfdTopic(TestCase):
 
         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.db.get_one.side_effect = [old_vnfd, old_vnfd, None]
         self.topic.upload_content(fake_session, did, new_vnfd, {}, {"Content-Type": []})
 
         db_args = self.db.replace.call_args[0]
@@ -1361,8 +1685,8 @@ class Test_NsdTopic(TestCase):
         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.fs.file_open.side_effect = lambda path, mode: tempfile.TemporaryFile(
+            mode="a+b"
         )
         self.db.get_one.side_effect = [
             {"_id": did, "_admin": deepcopy(db_nsd_content["_admin"])},
@@ -1847,30 +2171,50 @@ class Test_NsdTopic(TestCase):
         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
+    @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
+    @patch("osm_nbi.descriptor_topics.yaml.safe_load")
+    def test_validate_descriptor_ns_configuration_changed(
+        self, mock_safe_load, mock_detect_usage
+    ):
+        """Validating NSD and NSD has changes in ns-configuration:config-primitive"""
         descriptor_name, old_nsd, new_nsd = self.prepare_nsd_validation()
+        mock_safe_load.side_effect = [old_nsd, new_nsd]
+        mock_detect_usage.return_value = True
+        self.db.get_one.return_value = old_nsd
         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):
+            self.topic._validate_descriptor_changes(
+                old_nsd["_id"], descriptor_name, "/tmp", "/tmp:1"
+            )
+        self.db.get_one.assert_called_once()
+        mock_detect_usage.assert_called_once()
+        self.assertEqual(mock_safe_load.call_count, 2)
+
+    @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
+    @patch("osm_nbi.descriptor_topics.yaml.safe_load")
+    def test_validate_descriptor_nsd_name_changed(
+        self, mock_safe_load, mock_detect_usage
+    ):
+        """Validating NSD, NSD name has changed."""
         descriptor_name, old_nsd, new_nsd = self.prepare_nsd_validation()
+        did = old_nsd["_id"]
         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")
+        mock_safe_load.side_effect = [old_nsd, new_nsd]
+        mock_detect_usage.return_value = True
+        self.db.get_one.return_value = old_nsd
 
+        with self.assertRaises(
+            EngineException, msg="there are disallowed changes in the ns descriptor"
+        ) as e:
+            self.topic._validate_descriptor_changes(
+                did, descriptor_name, "/tmp", "/tmp:1"
+            )
         self.assertEqual(
             e.exception.http_code,
             HTTPStatus.UNPROCESSABLE_ENTITY,
@@ -1882,6 +2226,32 @@ class Test_NsdTopic(TestCase):
             "Wrong exception text",
         )
 
+        self.db.get_one.assert_called_once()
+        mock_detect_usage.assert_called_once()
+        self.assertEqual(mock_safe_load.call_count, 2)
+
+    @patch("osm_nbi.descriptor_topics.detect_descriptor_usage")
+    @patch("osm_nbi.descriptor_topics.yaml.safe_load")
+    def test_validate_descriptor_nsd_name_changed_nsd_not_in_use(
+        self, mock_safe_load, mock_detect_usage
+    ):
+        """Validating NSD, NSD name has changed, NSD is not in use."""
+        descriptor_name, old_nsd, new_nsd = self.prepare_nsd_validation()
+        did = old_nsd["_id"]
+        new_nsd["name"] = "nscharm-ns2"
+        mock_safe_load.side_effect = [old_nsd, new_nsd]
+        mock_detect_usage.return_value = None
+        self.db.get_one.return_value = old_nsd
+
+        with self.assertNotRaises(Exception):
+            self.topic._validate_descriptor_changes(
+                did, descriptor_name, "/tmp", "/tmp:1"
+            )
+
+        self.db.get_one.assert_called_once()
+        mock_detect_usage.assert_called_once()
+        mock_safe_load.assert_not_called()
+
     def test_validate_vld_mgmt_network_with_virtual_link_protocol_data_on_valid_descriptor(
         self,
     ):
@@ -2022,6 +2392,61 @@ class Test_NsdTopic(TestCase):
             "Wrong exception text",
         )
 
+    def test_validate_vnffgd_descriptor_on_valid_descriptor(self):
+        indata = yaml.safe_load(db_sfc_nsds_text)[0]
+        vnffgd = indata.get("vnffgd")
+        fg = vnffgd[0]
+        self.topic.validate_vnffgd_data(fg, indata)
+
+    def test_validate_vnffgd_descriptor_not_matching_nfp_position_element(self):
+        indata = yaml.safe_load(db_sfc_nsds_text)[0]
+        vnffgd = indata.get("vnffgd")
+        fg = vnffgd[0]
+        nfpd = fg.get("nfpd")[0]
+        with self.assertRaises(EngineException) as e:
+            fg.update({"nfp-position-element": [{"id": "test1"}]})
+            self.topic.validate_vnffgd_data(fg, indata)
+        self.assertEqual(
+            e.exception.http_code,
+            HTTPStatus.UNPROCESSABLE_ENTITY,
+            "Wrong HTTP status code",
+        )
+        self.assertIn(
+            norm(
+                "Error at vnffgd nfpd[id='{}']:nfp-position-element-id='{}' "
+                "does not match any nfp-position-element".format(nfpd["id"], "test")
+            ),
+            norm(str(e.exception)),
+            "Wrong exception text",
+        )
+
+    def test_validate_vnffgd_descriptor_not_matching_constituent_base_element_id(
+        self,
+    ):
+        indata = yaml.safe_load(db_sfc_nsds_text)[0]
+        vnffgd = indata.get("vnffgd")
+        fg = vnffgd[0]
+        fg["nfpd"][0]["position-desc-id"][0]["cp-profile-id"][0][
+            "constituent-profile-elements"
+        ][0]["constituent-base-element-id"] = "error_vnf"
+        with self.assertRaises(EngineException) as e:
+            self.topic.validate_vnffgd_data(fg, indata)
+        self.assertEqual(
+            e.exception.http_code,
+            HTTPStatus.UNPROCESSABLE_ENTITY,
+            "Wrong HTTP status code",
+        )
+        self.assertIn(
+            norm(
+                "Error at vnffgd constituent_profile[id='{}']:vnfd-id='{}' "
+                "does not match any constituent-base-element-id".format(
+                    "vnf1", "error_vnf"
+                )
+            ),
+            norm(str(e.exception)),
+            "Wrong exception text",
+        )
+
 
 if __name__ == "__main__":
     unittest.main()