+ raise EngineException(
+ "There is at least one NS instance using this descriptor",
+ http_code=HTTPStatus.CONFLICT,
+ )
+
+ # check NSD referenced by NST
+ del _filter["nsd-id"]
+ _filter["netslice-subnet.ANYINDEX.nsd-ref"] = descriptor_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 delete_extra(self, session, _id, db_content, not_send_msg=None):
+ """
+ Deletes associate file system storage (via super)
+ Deletes associated vnfpkgops from database.
+ :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
+ :param _id: server internal id
+ :param db_content: The database content of the descriptor
+ :return: None
+ :raises: FsException in case of error while deleting associated storage
+ """
+ super().delete_extra(session, _id, db_content, not_send_msg)
+ self.db.del_list(self.topic + "_revisions", {"_id": {"$regex": _id}})
+
+ @staticmethod
+ def extract_day12_primitives(nsd: dict) -> dict:
+ """Removes the day12 primitives from the NSD descriptors
+
+ Args:
+ nsd (dict): Descriptor as a dictionary
+
+ Returns:
+ nsd (dict): Cleared NSD
+ """
+ if nsd.get("ns-configuration"):
+ for key in [
+ "config-primitive",
+ "initial-config-primitive",
+ "terminate-config-primitive",
+ ]:
+ nsd["ns-configuration"].pop(key, None)
+ return nsd
+
+ def remove_modifiable_items(self, nsd: dict) -> dict:
+ """Removes the modifiable parts from the VNFD descriptors
+
+ It calls different extract functions according to different update types
+ to clear all the modifiable items from NSD
+
+ Args:
+ nsd (dict): Descriptor as a dictionary
+
+ Returns:
+ nsd (dict): Descriptor which does not include modifiable contents
+ """
+ while isinstance(nsd, dict) and nsd.get("nsd"):
+ nsd = nsd["nsd"]
+ if isinstance(nsd, list):
+ nsd = nsd[0]
+ nsd.pop("_admin", None)
+ # If the more extractions need to be done from NSD,
+ # the new extract methods could be appended to below list.
+ for extract_function in [self.extract_day12_primitives]:
+ nsd_temp = extract_function(nsd)
+ nsd = nsd_temp
+ return nsd
+
+ def _validate_descriptor_changes(
+ self,
+ descriptor_id: str,
+ descriptor_file_name: str,
+ old_descriptor_directory: str,
+ new_descriptor_directory: str,
+ ):
+ """Compares the old and new NSD descriptors and validates the new descriptor
+
+ Args:
+ old_descriptor_directory: Directory of descriptor which is in-use
+ new_descriptor_directory: Directory of descriptor which is proposed to update (new revision)
+
+ Returns:
+ None
+
+ Raises:
+ EngineException: In case of error if the changes are not allowed
+ """
+
+ try:
+ # If NSD does not exist in DB, or it is not in use by any NS,
+ # validation is not required.
+ nsd = self.db.get_one("nsds", {"_id": descriptor_id}, fail_on_empty=False)
+ if not nsd or not detect_descriptor_usage(nsd, "nsds", self.db):
+ return
+
+ # Get the old and new descriptor contents in order to compare them.
+ with self.fs.file_open(
+ (old_descriptor_directory.rstrip("/"), descriptor_file_name), "r"
+ ) as old_descriptor_file:
+ with self.fs.file_open(
+ (new_descriptor_directory.rstrip("/"), descriptor_file_name), "r"
+ ) as new_descriptor_file:
+ old_content = yaml.safe_load(old_descriptor_file.read())
+ new_content = yaml.safe_load(new_descriptor_file.read())
+
+ if old_content and new_content:
+ disallowed_change = DeepDiff(
+ self.remove_modifiable_items(old_content),
+ self.remove_modifiable_items(new_content),
+ )
+
+ if disallowed_change:
+ changed_nodes = functools.reduce(
+ lambda a, b: a + ", " + b,
+ [
+ node.lstrip("root")
+ for node in disallowed_change.get(
+ "values_changed"
+ ).keys()
+ ],
+ )
+
+ raise EngineException(
+ f"Error in validating new descriptor: {changed_nodes} cannot be modified, "
+ "there are disallowed changes in the ns descriptor. ",
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+ except (
+ DbException,
+ AttributeError,
+ IndexError,
+ KeyError,
+ ValueError,
+ ) as e:
+ raise type(e)(
+ "NS Descriptor could not be processed with error: {}.".format(e)
+ )
+
+ 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):
+ topic = "nsts"
+ topic_msg = "nst"
+ quota_name = "slice_templates"
+
+ def __init__(self, db, fs, msg, auth):
+ DescriptorTopic.__init__(self, db, fs, msg, auth)
+
+ def pyangbind_validation(self, item, data, force=False):
+ try:
+ mynst = nst_im()
+ pybindJSONDecoder.load_ietf_json(
+ {"nst": [data]},
+ None,
+ None,
+ obj=mynst,
+ path_helper=True,
+ skip_unknown=force,
+ )
+ out = pybindJSON.dumps(mynst, mode="ietf")
+ desc_out = self._remove_envelop(yaml.safe_load(out))
+ return desc_out
+ except Exception as e:
+ raise EngineException(
+ "Error in pyangbind validation: {}".format(str(e)),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+
+ @staticmethod
+ def _remove_envelop(indata=None):
+ if not indata:
+ return {}
+ clean_indata = indata
+
+ if clean_indata.get("nst"):
+ if (
+ not isinstance(clean_indata["nst"], list)
+ or len(clean_indata["nst"]) != 1
+ ):
+ raise EngineException("'nst' must be a list only one element")
+ clean_indata = clean_indata["nst"][0]
+ elif clean_indata.get("nst:nst"):
+ if (
+ not isinstance(clean_indata["nst:nst"], list)
+ or len(clean_indata["nst:nst"]) != 1
+ ):
+ raise EngineException("'nst:nst' must be a list only one element")
+ clean_indata = clean_indata["nst:nst"][0]
+ return clean_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()
+
+ def _check_descriptor_dependencies(self, session, descriptor):
+ """
+ Check that the dependent descriptors exist on a new descriptor or edition
+ :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
+ :param descriptor: descriptor to be inserted or edit
+ :return: None or raises exception
+ """
+ if not descriptor.get("netslice-subnet"):
+ return
+ for nsd in descriptor["netslice-subnet"]:
+ nsd_id = nsd["nsd-ref"]
+ filter_q = self._get_project_filter(session)
+ filter_q["id"] = nsd_id
+ if not self.db.get_list("nsds", filter_q):
+ raise EngineException(
+ "Descriptor error at 'netslice-subnet':'nsd-ref'='{}' references a non "
+ "existing nsd".format(nsd_id),
+ http_code=HTTPStatus.CONFLICT,
+ )
+
+ def check_conflict_on_edit(self, session, final_content, edit_content, _id):
+ final_content = super().check_conflict_on_edit(
+ session, final_content, edit_content, _id
+ )
+
+ self._check_descriptor_dependencies(session, final_content)
+ return final_content
+
+ def check_conflict_on_del(self, session, _id, db_content):
+ """
+ Check that there is not any NSIR that uses this NST. Only NSIRs belonging to this project are considered. Note
+ that NST can be public and be used by other projects.
+ :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
+ :param _id: nst internal id
+ :param db_content: The database content of the _id.
+ :return: None or raises EngineException with the conflict
+ """
+ # TODO: Check this method
+ if session["force"]:
+ return
+ # Get Network Slice Template from Database
+ _filter = self._get_project_filter(session)
+ _filter["_admin.nst-id"] = _id
+ if self.db.get_list("nsis", _filter):
+ 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)