# no yes -> error
# onefile yes no -> zip
# X yes -> text
-
- if accept_text and (not storage.get('pkg-dir') or path == "$DESCRIPTOR"):
+ contain_many_files = False
+ if storage.get('pkg-dir'):
+ # check if there are more than one file in the package, ignoring checksums.txt.
+ pkg_files = self.fs.dir_ls((storage['folder'], storage['pkg-dir']))
+ if len(pkg_files) >= 3 or (len(pkg_files) == 2 and 'checksums.txt' not in pkg_files):
+ contain_many_files = True
+ if accept_text and (not contain_many_files or path == "$DESCRIPTOR"):
return self.fs.file_open((storage['folder'], storage['descriptor']), "r"), "text/plain"
- elif storage.get('pkg-dir') and not accept_zip:
+ elif contain_many_files and not accept_zip:
raise EngineException("Packages that contains several files need to be retrieved with 'application/zip'"
"Accept header", http_code=HTTPStatus.NOT_ACCEPTABLE)
else:
raise EngineException("Error in pyangbind validation: {}".format(str(e)),
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 "_id" in indata:
+ indata.pop("_id")
+ if "_admin" not in indata:
+ indata["_admin"] = {}
+
+ if "operationalState" in indata:
+ if indata["operationalState"] in ("ENABLED", "DISABLED"):
+ indata["_admin"]["operationalState"] = indata.pop("operationalState")
+ else:
+ raise EngineException("State '{}' is not a valid operational state"
+ .format(indata["operationalState"]),
+ http_code=HTTPStatus.BAD_REQUEST)
+
+ # In the case of user defined data, we need to put the data in the root of the object
+ # to preserve current expected behaviour
+ if "userDefinedData" in indata:
+ data = indata.pop("userDefinedData")
+ if type(data) == dict:
+ indata["_admin"]["userDefinedData"] = data
+ else:
+ 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"]),
+ http_code=HTTPStatus.CONFLICT)
+
+ return indata
+
class VnfdTopic(DescriptorTopic):
topic = "vnfds"
http_code=HTTPStatus.CONFLICT)
def _validate_input_new(self, indata, storage_params, force=False):
+ indata.pop("onboardingState", None)
+ indata.pop("operationalState", None)
+ indata.pop("usageState", None)
+
+ indata.pop("links", None)
+
indata = self.pyangbind_validation("vnfds", indata, force)
# Cross references validation in the descriptor
if indata.get("vdu"):
http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
return indata
- def _validate_input_edit(self, indata, force=False):
- # not needed to validate with pyangbind becuase it will be validated at check_conflict_on_edit
- 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
"""
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"]
+ data["usageState"] = data["_admin"]["usageState"]
+
+ links = {}
+ links["self"] = {"href": "/vnfpkgm/v1/vnf_packages/{}".format(data["_id"])}
+ 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)
class NsdTopic(DescriptorTopic):
return clean_indata
def _validate_input_new(self, indata, storage_params, force=False):
+ indata.pop("nsdOnboardingState", None)
+ indata.pop("nsdOperationalState", None)
+ indata.pop("nsdUsageState", None)
+
+ indata.pop("links", None)
+
indata = self.pyangbind_validation("nsds", indata, force)
# 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
http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
return indata
- def _validate_input_edit(self, indata, force=False):
+ 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
+ """
+ indata looks as follows:
+ - In the new case (conformant)
+ {'nsdOperationalState': 'DISABLED', 'userDefinedData': {'id': 'string23',
+ '_id': 'c6ddc544-cede-4b94-9ebe-be07b298a3c1', 'name': 'simon46'}}
+ - In the old case (backwards-compatible)
+ {'id': 'string23', '_id': 'c6ddc544-cede-4b94-9ebe-be07b298a3c1', 'name': 'simon46'}
+ """
+ if "_admin" not in indata:
+ indata["_admin"] = {}
+
+ if "nsdOperationalState" in indata:
+ if indata["nsdOperationalState"] in ("ENABLED", "DISABLED"):
+ indata["_admin"]["operationalState"] = indata.pop("nsdOperationalState")
+ else:
+ raise EngineException("State '{}' is not a valid operational state"
+ .format(indata["nsdOperationalState"]),
+ http_code=HTTPStatus.BAD_REQUEST)
+
+ # In the case of user defined data, we need to put the data in the root of the object
+ # to preserve current expected behaviour
+ if "userDefinedData" in indata:
+ data = indata.pop("userDefinedData")
+ if type(data) == dict:
+ indata["_admin"]["userDefinedData"] = data
+ else:
+ 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("nsdOperationalState already {}".format(content["_admin"]["operationalState"]),
+ http_code=HTTPStatus.CONFLICT)
return indata
def _check_descriptor_dependencies(self, session, descriptor):
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"]
+ data["nsdUsageState"] = data["_admin"]["usageState"]
+
+ links = {}
+ 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)
class NstTopic(DescriptorTopic):
clean_indata = clean_indata['nst:nst'][0]
return clean_indata
- def _validate_input_edit(self, indata, force=False):
- # TODO validate with pyangbind, serialize
- return indata
-
def _validate_input_new(self, indata, storage_params, force=False):
+ indata.pop("onboardingState", None)
+ indata.pop("operationalState", None)
+ indata.pop("usageState", None)
indata = self.pyangbind_validation("nsts", indata, force)
return indata.copy()
raise EngineException("there is at least one Netslice Instance using this descriptor",
http_code=HTTPStatus.CONFLICT)
+ def sol005_projection(self, data):
+ data["onboardingState"] = data["_admin"]["onboardingState"]
+ data["operationalState"] = data["_admin"]["operationalState"]
+ data["usageState"] = data["_admin"]["usageState"]
+
+ links = {}
+ links["self"] = {"href": "/nst/v1/netslice_templates/{}".format(data["_id"])}
+ links["nst"] = {"href": "/nst/v1/netslice_templates/{}/nst".format(data["_id"])}
+ data["_links"] = links
+
+ return super().sol005_projection(data)
+
class PduTopic(BaseTopic):
topic = "pdus"