.format(desc_item_id, list_item[desc_item_id],
position), HTTPStatus.UNPROCESSABLE_ENTITY)
used_ids.append(list_item[desc_item_id])
+
_check_unique_id_name(final_content)
# 1. validate again with pyangbind
# 1.1. remove internal keys
break
file_pkg.write(indata_text)
if content_range_text:
- if indata_len != end-start:
+ if indata_len != end - start:
raise EngineException("Mismatch between Content-Range header {}-{} and body length of {}".format(
- start, end-1, indata_len), HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE)
+ start, end - 1, indata_len), HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE)
if end != total:
# TODO update to UPLOADING
return False
"onboardingState is {}".format(content["_admin"]["onboardingState"]),
http_code=HTTPStatus.CONFLICT)
storage = content["_admin"]["storage"]
- if path is not None and path != "$DESCRIPTOR": # artifacts
+ if path is not None and path != "$DESCRIPTOR": # artifacts
if not storage.get('pkg-dir'):
raise EngineException("Packages does not contains artifacts", http_code=HTTPStatus.BAD_REQUEST)
if self.fs.file_exists((storage['folder'], storage['pkg-dir'], *path), 'dir'):
return folder_content, "text/plain"
# TODO manage folders in http
else:
- return self.fs.file_open((storage['folder'], storage['pkg-dir'], *path), "rb"),\
+ return self.fs.file_open((storage['folder'], storage['pkg-dir'], *path), "rb"), \
"application/octet-stream"
# pkgtype accept ZIP TEXT -> result
raise EngineException("userDefinedData should be an object, but is '{}' instead"
.format(type(data)),
http_code=HTTPStatus.BAD_REQUEST)
-
+
if ("operationalState" in indata["_admin"] and
content["_admin"]["operationalState"] == indata["_admin"]["operationalState"]):
raise EngineException("operationalState already {}".format(content["_admin"]["operationalState"]),
indata = self.pyangbind_validation("vnfds", indata, force)
# Cross references validation in the descriptor
- if indata.get("vdu"):
- if not indata.get("mgmt-interface"):
- raise EngineException("'mgmt-interface' is a mandatory field and it is not defined",
+ self.validate_mgmt_interfaces_connection_points(indata)
+
+ for vdu in get_iterable(indata.get("vdu")):
+ self.validate_vdu_connection_point_refs(vdu, indata)
+ self._validate_vdu_charms_in_package(storage_params, vdu, indata)
+ self._validate_vdu_cloud_init_in_package(storage_params, vdu, indata)
+
+ self._validate_vnf_charms_in_package(storage_params, indata)
+
+ self.validate_internal_vlds(indata)
+ self.validate_monitoring_params(indata)
+ self.validate_scaling_group_descriptor(indata)
+
+ return indata
+
+ @staticmethod
+ def validate_mgmt_interfaces_connection_points(indata):
+ if not indata.get("vdu"):
+ return
+ if not indata.get("mgmt-interface"):
+ raise EngineException("'mgmt-interface' is a mandatory field and it is not defined",
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
+ if indata["mgmt-interface"].get("cp"):
+ for cp in get_iterable(indata.get("connection-point")):
+ if cp["name"] == indata["mgmt-interface"]["cp"]:
+ break
+ else:
+ raise EngineException("mgmt-interface:cp='{}' must match an existing connection-point"
+ .format(indata["mgmt-interface"]["cp"]),
http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
- if indata["mgmt-interface"].get("cp"):
+
+ @staticmethod
+ def validate_vdu_connection_point_refs(vdu, indata):
+ icp_refs = []
+ ecp_refs = []
+ for interface in get_iterable(vdu.get("interface")):
+ if interface.get("external-connection-point-ref"):
+ if interface.get("external-connection-point-ref") in ecp_refs:
+ raise EngineException("vdu[id='{}']:interface[name='{}']:external-connection-point-ref='{}' "
+ "is referenced by other interface"
+ .format(vdu["id"], interface["name"],
+ interface["external-connection-point-ref"]),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
+ ecp_refs.append(interface.get("external-connection-point-ref"))
for cp in get_iterable(indata.get("connection-point")):
- if cp["name"] == indata["mgmt-interface"]["cp"]:
+ if cp["name"] == interface["external-connection-point-ref"]:
break
else:
- raise EngineException("mgmt-interface:cp='{}' must match an existing connection-point"
- .format(indata["mgmt-interface"]["cp"]),
+ raise EngineException("vdu[id='{}']:interface[name='{}']:external-connection-point-ref='{}' "
+ "must match an existing connection-point"
+ .format(vdu["id"], interface["name"],
+ interface["external-connection-point-ref"]),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
+ elif interface.get("internal-connection-point-ref"):
+ if interface.get("internal-connection-point-ref") in icp_refs:
+ raise EngineException("vdu[id='{}']:interface[name='{}']:internal-connection-point-ref='{}' "
+ "is referenced by other interface"
+ .format(vdu["id"], interface["name"],
+ interface["internal-connection-point-ref"]),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
+ icp_refs.append(interface.get("internal-connection-point-ref"))
+ for internal_cp in get_iterable(vdu.get("internal-connection-point")):
+ if interface["internal-connection-point-ref"] == internal_cp.get("id"):
+ break
+ else:
+ raise EngineException("vdu[id='{}']:interface[name='{}']:internal-connection-point-ref='{}' "
+ "must match an existing vdu:internal-connection-point"
+ .format(vdu["id"], interface["name"],
+ interface["internal-connection-point-ref"]),
http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
- for vdu in get_iterable(indata.get("vdu")):
- icp_refs = []
- ecp_refs = []
- for interface in get_iterable(vdu.get("interface")):
- if interface.get("external-connection-point-ref"):
- if interface.get("external-connection-point-ref") in ecp_refs:
- raise EngineException("vdu[id='{}']:interface[name='{}']:external-connection-point-ref='{}' "
- "is referenced by other interface"
- .format(vdu["id"], interface["name"],
- interface["external-connection-point-ref"]),
- http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
- ecp_refs.append(interface.get("external-connection-point-ref"))
- for cp in get_iterable(indata.get("connection-point")):
- if cp["name"] == interface["external-connection-point-ref"]:
- break
- else:
- raise EngineException("vdu[id='{}']:interface[name='{}']:external-connection-point-ref='{}' "
- "must match an existing connection-point"
- .format(vdu["id"], interface["name"],
- interface["external-connection-point-ref"]),
- http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
- elif interface.get("internal-connection-point-ref"):
- if interface.get("internal-connection-point-ref") in icp_refs:
- raise EngineException("vdu[id='{}']:interface[name='{}']:internal-connection-point-ref='{}' "
- "is referenced by other interface"
- .format(vdu["id"], interface["name"],
- interface["internal-connection-point-ref"]),
- http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
- icp_refs.append(interface.get("internal-connection-point-ref"))
- for internal_cp in get_iterable(vdu.get("internal-connection-point")):
- if interface["internal-connection-point-ref"] == internal_cp.get("id"):
- break
- else:
- raise EngineException("vdu[id='{}']:interface[name='{}']:internal-connection-point-ref='{}' "
- "must match an existing vdu:internal-connection-point"
- .format(vdu["id"], interface["name"],
- interface["internal-connection-point-ref"]),
- http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
- # Validate that if descriptor contains charms, artifacts _admin.storage."pkg-dir" is not none
- if vdu.get("vdu-configuration"):
- if vdu["vdu-configuration"].get("juju"):
- if not self._validate_package_folders(storage_params, 'charms'):
- raise EngineException("Charm defined in vnf[id={}]:vdu[id={}] but not present in "
- "package".format(indata["id"], vdu["id"]))
- # Validate that if descriptor contains cloud-init, artifacts _admin.storage."pkg-dir" is not none
- if vdu.get("cloud-init-file"):
- if not self._validate_package_folders(storage_params, 'cloud_init', vdu["cloud-init-file"]):
- raise EngineException("Cloud-init defined in vnf[id={}]:vdu[id={}] but not present in "
- "package".format(indata["id"], vdu["id"]))
- # Validate that if descriptor contains charms, artifacts _admin.storage."pkg-dir" is not none
- if indata.get("vnf-configuration"):
- if indata["vnf-configuration"].get("juju"):
- if not self._validate_package_folders(storage_params, 'charms'):
- raise EngineException("Charm defined in vnf[id={}] but not present in "
- "package".format(indata["id"]))
+ def _validate_vdu_charms_in_package(self, storage_params, vdu, indata):
+ if not vdu.get("vdu-configuration"):
+ return
+ if vdu["vdu-configuration"].get("juju"):
+ if not self._validate_package_folders(storage_params, 'charms'):
+ raise EngineException("Charm defined in vnf[id={}]:vdu[id={}] but not present in "
+ "package".format(indata["id"], vdu["id"]))
+
+ def _validate_vdu_cloud_init_in_package(self, storage_params, vdu, indata):
+ if not vdu.get("cloud-init-file"):
+ return
+ if not self._validate_package_folders(storage_params, 'cloud_init', vdu["cloud-init-file"]):
+ raise EngineException("Cloud-init defined in vnf[id={}]:vdu[id={}] but not present in "
+ "package".format(indata["id"], vdu["id"]))
+
+ def _validate_vnf_charms_in_package(self, storage_params, indata):
+ if not indata.get("vnf-configuration"):
+ return
+ if indata["vnf-configuration"].get("juju"):
+ if not self._validate_package_folders(storage_params, 'charms'):
+ raise EngineException("Charm defined in vnf[id={}] but not present in "
+ "package".format(indata["id"]))
+
+ def _validate_package_folders(self, storage_params, folder, file=None):
+ if not storage_params or not storage_params.get("pkg-dir"):
+ return False
+ else:
+ if self.fs.file_exists("{}_".format(storage_params["folder"]), 'dir'):
+ f = "{}_/{}/{}".format(storage_params["folder"], storage_params["pkg-dir"], folder)
+ else:
+ f = "{}/{}/{}".format(storage_params["folder"], storage_params["pkg-dir"], folder)
+ if file:
+ return self.fs.file_exists("{}/{}".format(f, file), 'file')
+ else:
+ if self.fs.file_exists(f, 'dir'):
+ if self.fs.dir_ls(f):
+ return True
+ return False
+
+ @staticmethod
+ def validate_internal_vlds(indata):
vld_names = [] # For detection of duplicated VLD names
for ivld in get_iterable(indata.get("internal-vld")):
- # BEGIN Detection of duplicated VLD names
ivld_name = ivld.get("name")
- if ivld_name:
- if ivld_name in vld_names:
- raise EngineException("Duplicated VLD name '{}' in vnfd[id={}]:internal-vld[id={}]"
- .format(ivld["name"], indata["id"], ivld["id"]),
- http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
- else:
- vld_names.append(ivld_name)
- # END Detection of duplicated VLD names
+ if ivld_name and ivld_name in vld_names:
+ raise EngineException("Duplicated VLD name '{}' in vnfd[id={}]:internal-vld[id={}]"
+ .format(ivld["name"], indata["id"], ivld["id"]),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
+ else:
+ vld_names.append(ivld_name)
+
for icp in get_iterable(ivld.get("internal-connection-point")):
icp_mark = False
for vdu in get_iterable(indata.get("vdu")):
raise EngineException("internal-vld[id='{}']:ip-profile-ref='{}' does not exist".format(
ivld["id"], ivld["ip-profile-ref"]),
http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
+
+ @staticmethod
+ def validate_monitoring_params(indata):
for mp in get_iterable(indata.get("monitoring-param")):
if mp.get("vdu-monitoring-param"):
mp_vmp_mark = False
for vdu in get_iterable(indata.get("vdu")):
for vmp in get_iterable(vdu.get("monitoring-param")):
- if vmp["id"] == mp["vdu-monitoring-param"].get("vdu-monitoring-param-ref") and vdu["id"] ==\
+ if vmp["id"] == mp["vdu-monitoring-param"].get("vdu-monitoring-param-ref") and vdu["id"] == \
mp["vdu-monitoring-param"]["vdu-ref"]:
mp_vmp_mark = True
break
mp["vdu-metric"]["vdu-ref"]),
http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
+ @staticmethod
+ def validate_scaling_group_descriptor(indata):
for sgd in get_iterable(indata.get("scaling-group-descriptor")):
for sp in get_iterable(sgd.get("scaling-policy")):
for sc in get_iterable(sp.get("scaling-criteria")):
"vnf-configuration:config-primitive:name"
.format(sgd["name"], sca["vnf-config-primitive-name-ref"]),
http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
- return indata
-
- def _validate_package_folders(self, storage_params, folder, file=None):
- if not storage_params or not storage_params.get("pkg-dir"):
- return False
- else:
- if self.fs.file_exists("{}_".format(storage_params["folder"]), 'dir'):
- f = "{}_/{}/{}".format(storage_params["folder"], storage_params["pkg-dir"], folder)
- else:
- f = "{}/{}/{}".format(storage_params["folder"], storage_params["pkg-dir"], folder)
- if file:
- return self.fs.file_exists("{}/{}".format(f, file), 'file')
- else:
- if self.fs.file_exists(f, 'dir'):
- if self.fs.dir_ls(f):
- return True
- return False
def delete_extra(self, session, _id, db_content, not_send_msg=None):
"""
"""
super().delete_extra(session, _id, db_content, not_send_msg)
self.db.del_list("vnfpkgops", {"vnfPkgId": _id})
-
+
def sol005_projection(self, data):
data["onboardingState"] = data["_admin"]["onboardingState"]
data["operationalState"] = data["_admin"]["operationalState"]
links["vnfd"] = {"href": "/vnfpkgm/v1/vnf_packages/{}/vnfd".format(data["_id"])}
links["packageContent"] = {"href": "/vnfpkgm/v1/vnf_packages/{}/package_content".format(data["_id"])}
data["_links"] = links
-
+
return super().sol005_projection(data)
# Cross references validation in the descriptor
# TODO validata that if contains cloud-init-file or charms, have artifacts _admin.storage."pkg-dir" is not none
for vld in get_iterable(indata.get("vld")):
- if vld.get("mgmt-network") and vld.get("ip-profile-ref"):
- raise EngineException("Error at vld[id='{}']:ip-profile-ref"
- " You cannot set an ip-profile when mgmt-network is True"
- .format(vld["id"]), http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
- for vnfd_cp in get_iterable(vld.get("vnfd-connection-point-ref")):
- for constituent_vnfd in get_iterable(indata.get("constituent-vnfd")):
- if vnfd_cp["member-vnf-index-ref"] == constituent_vnfd["member-vnf-index"]:
- if vnfd_cp.get("vnfd-id-ref") and vnfd_cp["vnfd-id-ref"] != constituent_vnfd["vnfd-id-ref"]:
- raise EngineException("Error at vld[id='{}']:vnfd-connection-point-ref[vnfd-id-ref='{}'] "
- "does not match constituent-vnfd[member-vnf-index='{}']:vnfd-id-ref"
- " '{}'".format(vld["id"], vnfd_cp["vnfd-id-ref"],
- constituent_vnfd["member-vnf-index"],
- constituent_vnfd["vnfd-id-ref"]),
- http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
- break
- else:
- raise EngineException("Error at vld[id='{}']:vnfd-connection-point-ref[member-vnf-index-ref='{}'] "
- "does not match any constituent-vnfd:member-vnf-index"
- .format(vld["id"], vnfd_cp["member-vnf-index-ref"]),
- http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
- # Check VNFFGD
+ self.validate_vld_mgmt_network_with_ip_profile_ref(vld)
+ self.validate_vld_connection_point_refs(vld, indata)
+
for fgd in get_iterable(indata.get("vnffgd")):
- for cls in get_iterable(fgd.get("classifier")):
- rspref = cls.get("rsp-id-ref")
- for rsp in get_iterable(fgd.get("rsp")):
- rspid = rsp.get("id")
- if rspid and rspref and rspid == rspref:
- break
- else:
- raise EngineException(
- "Error at vnffgd[id='{}']:classifier[id='{}']:rsp-id-ref '{}' does not match any rsp:id"
- .format(fgd["id"], cls["id"], rspref),
- http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
+ self.validate_fgd_classifiers(fgd)
+
return indata
+ @staticmethod
+ def validate_vld_mgmt_network_with_ip_profile_ref(vld):
+ if vld.get("mgmt-network") and vld.get("ip-profile-ref"):
+ raise EngineException("Error at vld[id='{}']:ip-profile-ref"
+ " You cannot set an ip-profile when mgmt-network is True"
+ .format(vld["id"]), http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
+
+ @staticmethod
+ def validate_vld_connection_point_refs(vld, indata):
+ for vnfd_cp in get_iterable(vld.get("vnfd-connection-point-ref")):
+ for constituent_vnfd in get_iterable(indata.get("constituent-vnfd")):
+ if vnfd_cp["member-vnf-index-ref"] == constituent_vnfd["member-vnf-index"]:
+ if vnfd_cp.get("vnfd-id-ref") and vnfd_cp["vnfd-id-ref"] != constituent_vnfd["vnfd-id-ref"]:
+ raise EngineException("Error at vld[id='{}']:vnfd-connection-point-ref[vnfd-id-ref='{}'] "
+ "does not match constituent-vnfd[member-vnf-index='{}']:vnfd-id-ref"
+ " '{}'".format(vld["id"], vnfd_cp["vnfd-id-ref"],
+ constituent_vnfd["member-vnf-index"],
+ constituent_vnfd["vnfd-id-ref"]),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
+ break
+ else:
+ raise EngineException("Error at vld[id='{}']:vnfd-connection-point-ref[member-vnf-index-ref='{}'] "
+ "does not match any constituent-vnfd:member-vnf-index"
+ .format(vld["id"], vnfd_cp["member-vnf-index-ref"]),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
+
+ @staticmethod
+ def validate_fgd_classifiers(fgd):
+ for cls in get_iterable(fgd.get("classifier")):
+ rspref = cls.get("rsp-id-ref")
+ for rsp in get_iterable(fgd.get("rsp")):
+ rspid = rsp.get("id")
+ if rspid and rspref and rspid == rspref:
+ break
+ else:
+ raise EngineException(
+ "Error at vnffgd[id='{}']:classifier[id='{}']:rsp-id-ref '{}' does not match any rsp:id"
+ .format(fgd["id"], cls["id"], rspref),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
+
def _validate_input_edit(self, indata, content, force=False):
# not needed to validate with pyangbind becuase it will be validated at check_conflict_on_edit
"""
"""
if session["force"]:
return
+ member_vnfd_index = self._get_descriptor_constituent_vnfds_by_member_vnfd_index(session, descriptor)
+
+ # Cross references validation in the descriptor and vnfd connection point validation
+ for vld in get_iterable(descriptor.get("vld")):
+ self.validate_vld_connection_point_refs_vnfd_connection_points(vld, member_vnfd_index)
+
+ def _get_descriptor_constituent_vnfds_by_member_vnfd_index(self, session, descriptor):
member_vnfd_index = {}
if descriptor.get("constituent-vnfd") and not session["force"]:
for vnf in descriptor["constituent-vnfd"]:
vnfd_id = vnf["vnfd-id-ref"]
- filter_q = self._get_project_filter(session)
- filter_q["id"] = vnfd_id
- vnf_list = self.db.get_list("vnfds", filter_q)
+ query_filter = self._get_project_filter(session)
+ query_filter["id"] = vnfd_id
+ vnf_list = self.db.get_list("vnfds", query_filter)
if not vnf_list:
raise EngineException("Descriptor error at 'constituent-vnfd':'vnfd-id-ref'='{}' references a non "
"existing vnfd".format(vnfd_id), http_code=HTTPStatus.CONFLICT)
- # elif len(vnf_list) > 1:
- # raise EngineException("More than one vnfd found for id='{}'".format(vnfd_id),
- # http_code=HTTPStatus.CONFLICT)
+
member_vnfd_index[vnf["member-vnf-index"]] = vnf_list[0]
+ return member_vnfd_index
- # Cross references validation in the descriptor and vnfd connection point validation
- for vld in get_iterable(descriptor.get("vld")):
- for referenced_vnfd_cp in get_iterable(vld.get("vnfd-connection-point-ref")):
- # look if this vnfd contains this connection point
- vnfd = member_vnfd_index.get(referenced_vnfd_cp["member-vnf-index-ref"])
- for vnfd_cp in get_iterable(vnfd.get("connection-point")):
- if referenced_vnfd_cp.get("vnfd-connection-point-ref") == vnfd_cp["name"]:
- break
- else:
- raise EngineException(
- "Error at vld[id='{}']:vnfd-connection-point-ref[member-vnf-index-ref='{}']:vnfd-"
- "connection-point-ref='{}' references a non existing conection-point:name inside vnfd '{}'"
- .format(vld["id"], referenced_vnfd_cp["member-vnf-index-ref"],
- referenced_vnfd_cp["vnfd-connection-point-ref"], vnfd["id"]),
- http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
+ @staticmethod
+ def validate_vld_connection_point_refs_vnfd_connection_points(vld, member_vnfd_index):
+ for referenced_vnfd_cp in get_iterable(vld.get("vnfd-connection-point-ref")):
+ # look if this vnfd contains this connection point
+ vnfd = member_vnfd_index.get(referenced_vnfd_cp["member-vnf-index-ref"])
+ for vnfd_cp in get_iterable(vnfd.get("connection-point")):
+ if referenced_vnfd_cp.get("vnfd-connection-point-ref") == vnfd_cp["name"]:
+ break
+ else:
+ raise EngineException(
+ "Error at vld[id='{}']:vnfd-connection-point-ref[member-vnf-index-ref='{}']:vnfd-"
+ "connection-point-ref='{}' references a non existing conection-point:name inside vnfd '{}'"
+ .format(vld["id"], referenced_vnfd_cp["member-vnf-index-ref"],
+ referenced_vnfd_cp["vnfd-connection-point-ref"], vnfd["id"]),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
def check_conflict_on_edit(self, session, final_content, edit_content, _id):
super().check_conflict_on_edit(session, final_content, edit_content, _id)
if self.db.get_list("nsts", _filter):
raise EngineException("There is at least one NetSlice Template referencing this descriptor",
http_code=HTTPStatus.CONFLICT)
-
+
def sol005_projection(self, data):
data["nsdOnboardingState"] = data["_admin"]["onboardingState"]
data["nsdOperationalState"] = data["_admin"]["operationalState"]
links["self"] = {"href": "/nsd/v1/ns_descriptors/{}".format(data["_id"])}
links["nsd_content"] = {"href": "/nsd/v1/ns_descriptors/{}/nsd_content".format(data["_id"])}
data["_links"] = links
-
+
return super().sol005_projection(data)
from osm_common.dbbase import DbException
import yaml
-
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]
if isinstance(d1, dict) and isinstance(d2, dict):
for key in d1.keys():
if key in d2:
- compare_desc(tc, d1[key], d2[key], k+"[{}]".format(key))
+ compare_desc(tc, d1[key], d2[key], k + "[{}]".format(key))
elif isinstance(d1, list) and isinstance(d2, list) and len(d1) == len(d2):
for i in range(len(d1)):
- compare_desc(tc, d1[i], d2[i], k+"[{}]".format(i))
+ compare_desc(tc, d1[i], d2[i], k + "[{}]".format(i))
else:
tc.assertEqual(d1, d2, "Wrong descriptor content: {}".format(k))
finally:
test_vnfd["vdu"][0]["cloud-init-file"] = tmp1
test_vnfd["vnf-configuration"]["juju"] = tmp2
- self.db.get_one.side_effect = lambda table, filter, fail_on_empty=None, fail_on_more=None:\
+ self.db.get_one.side_effect = lambda table, filter, fail_on_empty=None, fail_on_more=None: \
{"_id": did, "_admin": db_vnfd_content["_admin"]}
with self.subTest(i=2, t='Check Pyangbind Validation: required properties'):
tmp = test_vnfd["id"]
sg["vdu"][0]["vdu-id-ref"] = test_vnfd["vdu"][0]["id"]
sg["scaling-config-action"] = [{"trigger": "pre-scale-in"}]
try:
- with self.assertRaises(EngineException, msg="Accepted non-existent Scaling Group VDU ID Reference")\
+ with self.assertRaises(EngineException, msg="Accepted non-existent Scaling Group VDU ID Reference") \
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.db.set_one.assert_not_called()
fs_del_calls = self.fs.file_delete.call_args_list
self.assertEqual(fs_del_calls[0][0][0], did, "Wrong FS file id")
- self.assertEqual(fs_del_calls[1][0][0], did+'_', "Wrong FS folder id")
+ self.assertEqual(fs_del_calls[1][0][0], did + '_', "Wrong FS folder id")
with self.subTest(i=2, t='Conflict on Delete - VNFD in use by VNFR'):
self.db.get_list.return_value = [{"_id": str(uuid4()), "name": "fake-vnfr"}]
with self.assertRaises(EngineException, msg="Accepted VNFD in use by VNFR") as e:
self.fs.file_delete.assert_not_called()
return
+ 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")
+
class Test_NsdTopic(TestCase):
compare_desc(self, test_nsd, db_args[2], "NSD")
finally:
pass
- self.db.get_one.side_effect = lambda table, filter, fail_on_empty=None, fail_on_more=None:\
+ self.db.get_one.side_effect = lambda table, filter, fail_on_empty=None, fail_on_more=None: \
{"_id": did, "_admin": db_nsd_content["_admin"]}
with self.subTest(i=2, t='Check Pyangbind Validation: required properties'):
tmp = test_nsd["id"]
.format(test_nsd["vld"][0]["id"],
test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["member-vnf-index-ref"],
test_nsd["vld"][0]["vnfd-connection-point-ref"][0]
- ["vnfd-connection-point-ref"],
+ ["vnfd-connection-point-ref"],
db_vnfd_content["id"])),
norm(str(e.exception)), "Wrong exception text")
finally:
test_nsd["vld"][0]["vnfd-connection-point-ref"][0]["vnfd-connection-point-ref"] = tmp
return
- with self.subTest(i=11, t='Check Input Validation: everything right'):
- test_nsd["id"] = "fake-nsd-id"
- self.db.get_one.side_effect = [{"_id": did, "_admin": db_nsd_content["_admin"]}, None]
- self.db.get_list.return_value = [db_vnfd_content]
- rc = self.topic.upload_content(fake_session, did, test_nsd, {}, {"Content-Type": []})
- self.assertTrue(rc, "Input Validation: Unexpected failure")
- return
def test_edit_nsd(self):
did = db_nsd_content["_id"]
self.db.set_one.assert_not_called()
fs_del_calls = self.fs.file_delete.call_args_list
self.assertEqual(fs_del_calls[0][0][0], did, "Wrong FS file id")
- self.assertEqual(fs_del_calls[1][0][0], did+'_', "Wrong FS folder id")
+ self.assertEqual(fs_del_calls[1][0][0], did + '_', "Wrong FS folder id")
with self.subTest(i=2, t='Conflict on Delete - NSD in use by nsr'):
self.db.get_list.return_value = [{"_id": str(uuid4()), "name": "fake-nsr"}]
with self.assertRaises(EngineException, msg="Accepted NSD in use by NSR") as e:
self.fs.file_delete.assert_not_called()
return
+ def test_validate_vld_mgmt_network_with_ip_profile_ref_on_valid_descriptor(self):
+ indata = deepcopy(db_nsd_content)
+ vld = indata['vld'][0]
+ self.topic.validate_vld_mgmt_network_with_ip_profile_ref(vld)
+
+ def test_validate_vld_mgmt_network_with_ip_profile_ref_when_both_defined(self):
+ indata = deepcopy(db_nsd_content)
+ vld = indata['vld'][0]
+ vld['ip-profile-ref'] = 'a-profile-ref'
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_vld_mgmt_network_with_ip_profile_ref(vld)
+ self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
+ self.assertIn(norm("Error at vld[id='{}']:ip-profile-ref"
+ " You cannot set an ip-profile when mgmt-network is True"
+ .format(vld["id"])),
+ norm(str(e.exception)), "Wrong exception text")
+
+ def test_validate_vld_connection_point_refs_on_valid_descriptor(self):
+ indata = deepcopy(db_nsd_content)
+ vld = indata['vld'][0]
+ self.topic.validate_vld_connection_point_refs(vld, indata)
+
+ def test_validate_vld_connection_point_refs_when_missing_constituent_vnfd(self):
+ indata = deepcopy(db_nsd_content)
+ vld = indata['vld'][0]
+ vnfd_cp = vld['vnfd-connection-point-ref'][0]
+ indata['constituent-vnfd'] = []
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_vld_connection_point_refs(vld, indata)
+ self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
+ self.assertIn(norm("Error at vld[id='{}']:vnfd-connection-point-ref[member-vnf-index-ref='{}'] "
+ "does not match any constituent-vnfd:member-vnf-index"
+ .format(vld["id"], vnfd_cp["member-vnf-index-ref"])),
+ norm(str(e.exception)), "Wrong exception text")
+
+ def test_validate_vld_connection_point_refs_on_unmatched_constituent_vnfd(self):
+ indata = deepcopy(db_nsd_content)
+ vld = indata['vld'][0]
+ vnfd_cp = vld['vnfd-connection-point-ref'][0]
+ constituent_vnfd = indata['constituent-vnfd'][0]
+ constituent_vnfd['vnfd-id-ref'] = 'unmatched-vnfd-id-ref'
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_vld_connection_point_refs(vld, indata)
+ self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
+ self.assertIn(norm("Error at vld[id='{}']:vnfd-connection-point-ref[vnfd-id-ref='{}'] "
+ "does not match constituent-vnfd[member-vnf-index='{}']:vnfd-id-ref"
+ " '{}'".format(vld["id"], vnfd_cp["vnfd-id-ref"],
+ constituent_vnfd["member-vnf-index"],
+ constituent_vnfd["vnfd-id-ref"])),
+ norm(str(e.exception)), "Wrong exception text")
+
+ def test_validate_vld_connection_point_refs_vnfd_connection_points_on_valid_descriptor(self):
+ nsd_descriptor = deepcopy(db_nsd_content)
+ vnfd_1_descriptor = deepcopy(db_vnfd_content)
+ vnfd_2_descriptor = deepcopy(db_vnfd_content)
+ vld = nsd_descriptor['vld'][0]
+ member_vnfd_index = {'1': vnfd_1_descriptor, '2': vnfd_2_descriptor}
+ self.topic.validate_vld_connection_point_refs_vnfd_connection_points(vld, member_vnfd_index)
+
+ def test_validate_vld_connection_point_refs_vnfd_connection_points_when_missing_connection_point(self):
+ nsd_descriptor = deepcopy(db_nsd_content)
+ vnfd_1_descriptor = deepcopy(db_vnfd_content)
+ vnfd_2_descriptor = deepcopy(db_vnfd_content)
+ vld = nsd_descriptor['vld'][0]
+ referenced_vnfd_cp = vld['vnfd-connection-point-ref'][0]
+ member_vnfd_index = {'1': vnfd_1_descriptor, '2': vnfd_2_descriptor}
+ vnfd_1_descriptor['connection-point'] = []
+ with self.assertRaises(EngineException) as e:
+ self.topic.validate_vld_connection_point_refs_vnfd_connection_points(vld, member_vnfd_index)
+ self.assertEqual(e.exception.http_code, HTTPStatus.UNPROCESSABLE_ENTITY, "Wrong HTTP status code")
+ self.assertIn(norm("Error at vld[id='{}']:vnfd-connection-point-ref[member-vnf-index-ref='{}']:vnfd-"
+ "connection-point-ref='{}' references a non existing conection-point:name inside vnfd '{}'"
+ .format(vld["id"], referenced_vnfd_cp["member-vnf-index-ref"],
+ referenced_vnfd_cp["vnfd-connection-point-ref"], vnfd_1_descriptor["id"])),
+ norm(str(e.exception)), "Wrong exception text")
+
+ def test_check_conflict_on_edit_when_missing_vnfd_index(self):
+ nsd_descriptor = deepcopy(db_nsd_content)
+ invalid_vnfd_id_ref = 'invalid-vnfd-id-ref'
+ nsd_descriptor['id'] = 'invalid-vnfd-id-ns'
+ nsd_descriptor['constituent-vnfd'][0]['vnfd-id-ref'] = invalid_vnfd_id_ref
+ nsd_descriptor['vld'][0]['vnfd-connection-point-ref'][0]['vnfd-id-ref'] = invalid_vnfd_id_ref
+ nsd_descriptor['vld'][1]['vnfd-connection-point-ref'][0]['vnfd-id-ref'] = invalid_vnfd_id_ref
+ with self.assertRaises(EngineException) as e:
+ self.db.get_list.return_value = []
+ self.topic.check_conflict_on_edit(fake_session, nsd_descriptor, [], 'id')
+ self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code")
+ self.assertIn(norm("Descriptor error at 'constituent-vnfd':'vnfd-id-ref'='{}' references a non "
+ "existing vnfd".format(invalid_vnfd_id_ref)),
+ norm(str(e.exception)), "Wrong exception text")
+
if __name__ == '__main__':
unittest.main()
projects / osm / NBI.git / commitdiff