+ def test_validate_mgmt_interfaces_connection_points_on_valid_descriptor(self):
+ indata = deepcopy(db_vnfd_content)
+ self.topic.validate_mgmt_interfaces_connection_points(indata)
+
+ def test_validate_mgmt_interfaces_connection_points_when_missing_connection_point(self):
+ indata = deepcopy(db_vnfd_content)
+ indata['connection-point'] = []
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_mgmt_interfaces_connection_points(indata)
+ self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
+ self.assertIn(norm("mgmt-interface:cp='{}' must match an existing connection-point"
+ .format(indata["mgmt-interface"]["cp"])),
+ norm(str(e.exception)), "Wrong exception text")
+
+ def test_validate_mgmt_interfaces_connection_points_when_missing_mgmt_interface(self):
+ indata = deepcopy(db_vnfd_content)
+ indata.pop('mgmt-interface')
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_mgmt_interfaces_connection_points(indata)
+ self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
+ self.assertIn(norm("'mgmt-interface' is a mandatory field and it is not defined"),
+ norm(str(e.exception)), "Wrong exception text")
+
+ def test_validate_vdu_connection_point_refs_on_valid_descriptor(self):
+ indata = db_vnfd_content
+ vdu = indata['vdu'][0]
+ self.topic.validate_vdu_connection_point_refs(vdu, indata)
+
+ def test_validate_vdu_connection_point_refs_when_missing_internal_connection_point(self):
+ indata = deepcopy(db_vnfd_content)
+ vdu = indata['vdu'][0]
+ vdu.pop('internal-connection-point')
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_vdu_connection_point_refs(vdu, indata)
+ self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
+ self.assertIn(norm("vdu[id='{}']:interface[name='{}']:internal-connection-point-ref='{}' "
+ "must match an existing vdu:internal-connection-point"
+ .format(vdu["id"], vdu['interface'][1]["name"],
+ vdu['interface'][1]["internal-connection-point-ref"])),
+ norm(str(e.exception)), "Wrong exception text")
+
+ def test_validate_vdu_connection_point_refs_when_missing_external_connection_point(self):
+ indata = deepcopy(db_vnfd_content)
+ vdu = indata['vdu'][0]
+ indata.pop('connection-point')
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_vdu_connection_point_refs(vdu, indata)
+ self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
+ self.assertIn(norm("vdu[id='{}']:interface[name='{}']:external-connection-point-ref='{}' "
+ "must match an existing connection-point"
+ .format(vdu["id"], vdu['interface'][0]["name"],
+ vdu['interface'][0]["external-connection-point-ref"])),
+ norm(str(e.exception)), "Wrong exception text")
+
+ def test_validate_vdu_connection_point_refs_on_duplicated_internal_connection_point_ref(self):
+ indata = deepcopy(db_vnfd_content)
+ vdu = indata['vdu'][0]
+ duplicated_interface = {'name': 'dup-mgmt-eth1', 'position': 3,
+ 'internal-connection-point-ref': 'mgmtVM-internal'}
+ vdu['interface'].insert(0, duplicated_interface)
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_vdu_connection_point_refs(vdu, indata)
+ self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
+ self.assertIn(norm("vdu[id='{}']:interface[name='{}']:internal-connection-point-ref='{}' "
+ "is referenced by other interface"
+ .format(vdu["id"], vdu['interface'][2]["name"],
+ vdu['interface'][2]["internal-connection-point-ref"])),
+ norm(str(e.exception)), "Wrong exception text")
+
+ def test_validate_vdu_connection_point_refs_on_duplicated_external_connection_point_ref(self):
+ indata = deepcopy(db_vnfd_content)
+ vdu = indata['vdu'][0]
+ duplicated_interface = {'name': 'dup-mgmt-eth0', 'position': 3, 'external-connection-point-ref': 'vnf-mgmt'}
+ vdu['interface'].insert(0, duplicated_interface)
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_vdu_connection_point_refs(vdu, indata)
+ self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
+ self.assertIn(norm("vdu[id='{}']:interface[name='{}']:external-connection-point-ref='{}' "
+ "is referenced by other interface"
+ .format(vdu["id"], vdu['interface'][1]["name"],
+ vdu['interface'][1]["external-connection-point-ref"])),
+ norm(str(e.exception)), "Wrong exception text")
+
+ def test_validate_internal_vlds_on_valid_descriptor(self):
+ indata = db_vnfd_content
+ self.topic.validate_internal_vlds(indata)
+
+ def test_validate_internal_vlds_on_duplicated_vld(self):
+ indata = deepcopy(db_vnfd_content)
+ duplicated_vld = {'id': 'internal', 'name': 'internal',
+ 'internal-connection-point': [{'id-ref': 'mgmtVM-internal'}, {'id-ref': 'dataVM-internal'}]}
+ indata['internal-vld'].insert(0, duplicated_vld)
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_internal_vlds(indata)
+ self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
+ self.assertIn(norm("Duplicated VLD name '{}' in vnfd[id={}]:internal-vld[id={}]"
+ .format(indata['internal-vld'][1]["name"], indata["id"], indata['internal-vld'][1]["id"])),
+ norm(str(e.exception)), "Wrong exception text")
+
+ def test_validate_internal_vlds_when_missing_internal_connection_point(self):
+ indata = deepcopy(db_vnfd_content)
+ ivld = indata['internal-vld'][0]
+ icp = ivld['internal-connection-point'][0]
+ indata['vdu'][0]['internal-connection-point'].pop()
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_internal_vlds(indata)
+ self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
+ self.assertIn(norm("internal-vld[id='{}']:internal-connection-point='{}' must match an existing "
+ "vdu:internal-connection-point".format(ivld["id"], icp["id-ref"])),
+ norm(str(e.exception)), "Wrong exception text")
+
+ def test_validate_internal_vlds_when_missing_ip_profile(self):
+ indata = deepcopy(db_vnfd_content)
+ ivld = indata['internal-vld'][0]
+ ivld['ip-profile-ref'] = 'non-existing-ip-profile'
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_internal_vlds(indata)
+ self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
+ self.assertIn(norm("internal-vld[id='{}']:ip-profile-ref='{}' does not exist".format(
+ ivld["id"], ivld["ip-profile-ref"])),
+ 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_when_missing_vdu(self):
+ indata = deepcopy(db_vnfd_content)
+ monitoring_param = indata['monitoring-param'][0]
+ indata['vdu'].pop()
+ 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("monitoring-param:vdu-monitoring-param:vdu-monitoring-param-ref='{}' not "
+ "defined at vdu[id='{}'] or vdu does not exist"
+ .format(monitoring_param["vdu-monitoring-param"]["vdu-monitoring-param-ref"],
+ monitoring_param["vdu-monitoring-param"]["vdu-ref"])),
+ norm(str(e.exception)), "Wrong exception text")
+
+ def test_validate_monitoring_params_when_missing_vdu_monitoring_param_ref(self):
+ indata = deepcopy(db_vnfd_content)
+ monitoring_param = indata['monitoring-param'][0]
+ indata['vdu'][1]['monitoring-param'] = []
+ 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("monitoring-param:vdu-monitoring-param:vdu-monitoring-param-ref='{}' not "
+ "defined at vdu[id='{}'] or vdu does not exist"
+ .format(monitoring_param["vdu-monitoring-param"]["vdu-monitoring-param-ref"],
+ monitoring_param["vdu-monitoring-param"]["vdu-ref"])),
+ 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_vnf_monitoring_param_ref(self):
+ indata = deepcopy(db_vnfd_content)
+ sgd = indata['scaling-group-descriptor'][0]
+ sc = sgd['scaling-policy'][0]['scaling-criteria'][0]
+ indata['monitoring-param'] = []
+ 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("scaling-group-descriptor[name='{}']:scaling-criteria[name='{}']:"
+ "vnf-monitoring-param-ref='{}' not defined in any monitoring-param"
+ .format(sgd["name"], sc["name"], sc["vnf-monitoring-param-ref"])),
+ norm(str(e.exception)), "Wrong exception text")
+
+ def test_validate_scaling_group_descriptor_when_missing_vdu(self):
+ indata = deepcopy(db_vnfd_content)
+ sgd = indata['scaling-group-descriptor'][0]
+ sgd_vdu = sgd['vdu'][0]
+ indata['vdu'].pop()
+ 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("scaling-group-descriptor[name='{}']:vdu-id-ref={} does not match any vdu"
+ .format(sgd["name"], sgd_vdu["vdu-id-ref"])),
+ norm(str(e.exception)), "Wrong exception text")
+
+ def test_validate_scaling_group_descriptor_when_missing_vnf_configuration(self):
+ indata = deepcopy(db_vnfd_content)
+ sgd = indata['scaling-group-descriptor'][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 "
+ "scaling-group-descriptor[name='{}']:scaling-config-action"
+ .format(sgd["name"])),
+ 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)
+ sgd = indata['scaling-group-descriptor'][0]
+ sca = sgd['scaling-config-action'][0]
+ indata['vnf-configuration']['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("scaling-group-descriptor[name='{}']:scaling-config-action:vnf-config-"
+ "primitive-name-ref='{}' does not match any "
+ "vnf-configuration:config-primitive:name"
+ .format(sgd["name"], sca["vnf-config-primitive-name-ref"])),
+ norm(str(e.exception)), "Wrong exception text")
+