+ def test_validate_mgmt_interface_connection_point_on_valid_descriptor(self):
+ indata = deepcopy(db_vnfd_content)
+ self.topic.validate_mgmt_interface_connection_point(indata)
+
+ 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")
+
+ def test_validate_scaling_group_descriptor_on_valid_descriptor(self):
+ indata = db_vnfd_content
+ self.topic.validate_scaling_group_descriptor(indata)
+
+ def test_validate_scaling_group_descriptor_when_missing_monitoring_param(self):
+ indata = deepcopy(db_vnfd_content)
+ vdu = indata['vdu'][1]
+ affected_df = indata['df'][0]
+ affected_sa = affected_df['scaling-aspect'][0]
+ affected_sp = affected_sa['scaling-policy'][0]
+ affected_sc = affected_sp['scaling-criteria'][0]
+ vdu.pop('monitoring-parameter')
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_scaling_group_descriptor(indata)
+ 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"], affected_sa["id"], affected_sp["name"], affected_sc["name"],
+ affected_sc["vnf-monitoring-param-ref"])),
+ norm(str(e.exception)), "Wrong exception text")
+
+ def test_validate_scaling_group_descriptor_when_missing_vnf_configuration(self):
+ indata = deepcopy(db_vnfd_content)
+ df = indata['df'][0]
+ affected_sa = df['scaling-aspect'][0]
+ indata.pop('vnf-configuration')
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_scaling_group_descriptor(indata)
+ self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
+ self.assertIn(norm("'vnf-configuration' not defined in the descriptor but it is referenced "
+ "by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action"
+ .format(df["id"], affected_sa["id"])),
+ norm(str(e.exception)), "Wrong exception text")
+
+ def test_validate_scaling_group_descriptor_when_missing_scaling_config_action_primitive(self):
+ indata = deepcopy(db_vnfd_content)
+ df = indata['df'][0]
+ affected_sa = df['scaling-aspect'][0]
+ affected_sca_primitive = affected_sa['scaling-config-action'][0]['vnf-config-primitive-name-ref']
+ indata['vnf-configuration'][0]['config-primitive'] = []
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_scaling_group_descriptor(indata)
+ 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 "
+ "vnf-configuration:config-primitive:name"
+ .format(df["id"], affected_sa["id"], affected_sca_primitive)),
+ norm(str(e.exception)), "Wrong exception text")
+