+ def test_validate_mgmt_interface_connection_point_when_missing_connection_point(
+ self,
+ ):
+ indata = deepcopy(db_vnfd_content)
+ indata["ext-cpd"] = []
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_mgmt_interface_connection_point(indata)
+ 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(indata["mgmt-cp"])
+ ),
+ norm(str(e.exception)),
+ "Wrong exception text",
+ )
+
+ def test_validate_mgmt_interface_connection_point_when_missing_mgmt_cp(self):
+ indata = deepcopy(db_vnfd_content)
+ indata.pop("mgmt-cp")
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_mgmt_interface_connection_point(indata)
+ self.assertEqual(
+ e.exception.http_code,
+ HTTPStatus.UNPROCESSABLE_ENTITY,
+ "Wrong HTTP status code",
+ )
+ self.assertIn(
+ norm("'mgmt-cp' is a mandatory field and it is not defined"),
+ norm(str(e.exception)),
+ "Wrong exception text",
+ )
+
+ def test_validate_vdu_internal_connection_points_on_valid_descriptor(self):
+ indata = db_vnfd_content
+ vdu = indata["vdu"][0]
+ self.topic.validate_vdu_internal_connection_points(vdu)
+
+ def test_validate_external_connection_points_on_valid_descriptor(self):
+ indata = db_vnfd_content
+ self.topic.validate_external_connection_points(indata)
+
+ def test_validate_external_connection_points_when_missing_internal_connection_point(
+ self,
+ ):
+ indata = deepcopy(db_vnfd_content)
+ vdu = indata["vdu"][0]
+ vdu.pop("int-cpd")
+ affected_ext_cpd = indata["ext-cpd"][0]
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_external_connection_points(indata)
+ 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(
+ affected_ext_cpd["id"]
+ )
+ ),
+ norm(str(e.exception)),
+ "Wrong exception text",
+ )
+
+ def test_validate_vdu_internal_connection_points_on_duplicated_internal_connection_point(
+ self,
+ ):
+ indata = deepcopy(db_vnfd_content)
+ vdu = indata["vdu"][0]
+ duplicated_cpd = {
+ "id": "vnf-mgmt",
+ "order": 3,
+ "virtual-network-interface-requirement": [{"name": "duplicated"}],
+ }
+ vdu["int-cpd"].insert(0, duplicated_cpd)
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_vdu_internal_connection_points(vdu)
+ self.assertEqual(
+ e.exception.http_code,
+ HTTPStatus.UNPROCESSABLE_ENTITY,
+ "Wrong HTTP status code",
+ )
+ self.assertIn(
+ norm(
+ "vdu[id='{}']:int-cpd[id='{}'] is already used by other int-cpd".format(
+ vdu["id"], duplicated_cpd["id"]
+ )
+ ),
+ norm(str(e.exception)),
+ "Wrong exception text",
+ )
+
+ def test_validate_external_connection_points_on_duplicated_external_connection_point(
+ self,
+ ):
+ indata = deepcopy(db_vnfd_content)
+ duplicated_cpd = {
+ "id": "vnf-mgmt-ext",
+ "int-cpd": {"vdu-id": "dataVM", "cpd": "vnf-data"},
+ }
+ indata["ext-cpd"].insert(0, duplicated_cpd)
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_external_connection_points(indata)
+ self.assertEqual(
+ e.exception.http_code,
+ HTTPStatus.UNPROCESSABLE_ENTITY,
+ "Wrong HTTP status code",
+ )
+ self.assertIn(
+ norm(
+ "ext-cpd[id='{}'] is already used by other ext-cpd".format(
+ duplicated_cpd["id"]
+ )
+ ),
+ norm(str(e.exception)),
+ "Wrong exception text",
+ )
+
+ def test_validate_internal_virtual_links_on_valid_descriptor(self):
+ indata = db_vnfd_content
+ self.topic.validate_internal_virtual_links(indata)
+
+ def test_validate_internal_virtual_links_on_duplicated_ivld(self):
+ indata = deepcopy(db_vnfd_content)
+ duplicated_vld = {"id": "internal"}
+ indata["int-virtual-link-desc"].insert(0, duplicated_vld)
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_internal_virtual_links(indata)
+ self.assertEqual(
+ e.exception.http_code,
+ HTTPStatus.UNPROCESSABLE_ENTITY,
+ "Wrong HTTP status code",
+ )
+ self.assertIn(
+ norm(
+ "Duplicated VLD id in int-virtual-link-desc[id={}]".format(
+ duplicated_vld["id"]
+ )
+ ),
+ norm(str(e.exception)),
+ "Wrong exception text",
+ )
+
+ def test_validate_internal_virtual_links_when_missing_ivld_on_connection_point(
+ self,
+ ):
+ indata = deepcopy(db_vnfd_content)
+ vdu = indata["vdu"][0]
+ affected_int_cpd = vdu["int-cpd"][0]
+ affected_int_cpd["int-virtual-link-desc"] = "non-existing-int-virtual-link-desc"
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_internal_virtual_links(indata)
+ 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"],
+ affected_int_cpd["id"],
+ affected_int_cpd["int-virtual-link-desc"],
+ )
+ ),
+ norm(str(e.exception)),
+ "Wrong exception text",
+ )
+
+ def test_validate_internal_virtual_links_when_missing_ivld_on_profile(self):
+ indata = deepcopy(db_vnfd_content)
+ affected_ivld_profile = {"id": "non-existing-int-virtual-link-desc"}
+ df = indata["df"][0]
+ df["virtual-link-profile"] = [affected_ivld_profile]
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_internal_virtual_links(indata)
+ 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"], affected_ivld_profile["id"])
+ ),
+ norm(str(e.exception)),
+ "Wrong exception text",
+ )
+
+ def test_validate_monitoring_params_on_valid_descriptor(self):
+ indata = db_vnfd_content
+ self.topic.validate_monitoring_params(indata)
+
+ def test_validate_monitoring_params_on_duplicated_ivld_monitoring_param(self):
+ indata = deepcopy(db_vnfd_content)
+ duplicated_mp = {"id": "cpu", "name": "cpu", "performance_metric": "cpu"}
+ affected_ivld = indata["int-virtual-link-desc"][0]
+ affected_ivld["monitoring-parameters"] = [duplicated_mp, duplicated_mp]
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_monitoring_params(indata)
+ self.assertEqual(
+ e.exception.http_code,
+ HTTPStatus.UNPROCESSABLE_ENTITY,
+ "Wrong HTTP status code",
+ )
+ self.assertIn(
+ norm(
+ "Duplicated monitoring-parameter id in "
+ "int-virtual-link-desc[id='{}']:monitoring-parameters[id='{}']".format(
+ affected_ivld["id"], duplicated_mp["id"]
+ )
+ ),
+ norm(str(e.exception)),
+ "Wrong exception text",
+ )
+
+ def test_validate_monitoring_params_on_duplicated_vdu_monitoring_param(self):
+ indata = deepcopy(db_vnfd_content)
+ duplicated_mp = {
+ "id": "dataVM_cpu_util",
+ "name": "dataVM_cpu_util",
+ "performance_metric": "cpu",
+ }
+ affected_vdu = indata["vdu"][1]
+ affected_vdu["monitoring-parameter"].insert(0, duplicated_mp)
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_monitoring_params(indata)
+ self.assertEqual(
+ e.exception.http_code,
+ HTTPStatus.UNPROCESSABLE_ENTITY,
+ "Wrong HTTP status code",
+ )
+ self.assertIn(
+ norm(
+ "Duplicated monitoring-parameter id in "
+ "vdu[id='{}']:monitoring-parameter[id='{}']".format(
+ affected_vdu["id"], duplicated_mp["id"]
+ )
+ ),
+ norm(str(e.exception)),
+ "Wrong exception text",
+ )
+
+ def test_validate_monitoring_params_on_duplicated_df_monitoring_param(self):
+ indata = deepcopy(db_vnfd_content)
+ duplicated_mp = {
+ "id": "memory",
+ "name": "memory",
+ "performance_metric": "memory",
+ }
+ affected_df = indata["df"][0]
+ affected_df["monitoring-parameter"] = [duplicated_mp, duplicated_mp]
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_monitoring_params(indata)
+ self.assertEqual(
+ e.exception.http_code,
+ HTTPStatus.UNPROCESSABLE_ENTITY,
+ "Wrong HTTP status code",
+ )
+ self.assertIn(
+ norm(
+ "Duplicated monitoring-parameter id in "
+ "df[id='{}']:monitoring-parameter[id='{}']".format(
+ affected_df["id"], duplicated_mp["id"]
+ )
+ ),
+ norm(str(e.exception)),
+ "Wrong exception text",
+ )