BaseTopic.__init__(self, db, fs, msg)
def check_conflict_on_edit(self, session, final_content, edit_content, _id, force=False):
- # check that this id is not present
- _filter = {"id": final_content["id"]}
- if _id:
+ # 1. validate again with pyangbind
+ # 1.1. remove internal keys
+ internal_keys = {}
+ for k in ("_id", "_admin"):
+ if k in final_content:
+ internal_keys[k] = final_content.pop(k)
+ serialized = self._validate_input_new(final_content, force)
+ # 1.2. modify final_content with a serialized version
+ final_content.clear()
+ final_content.update(serialized)
+ # 1.3. restore internal keys
+ for k, v in internal_keys.items():
+ final_content[k] = v
+
+ # 2. check that this id is not present
+ if "id" in edit_content:
+ _filter = self._get_project_filter(session, write=False, show_all=False)
+ _filter["id"] = final_content["id"]
_filter["_id.neq"] = _id
-
- _filter.update(self._get_project_filter(session, write=False, show_all=False))
- if self.db.get_one(self.topic, _filter, fail_on_empty=False):
- raise EngineException("{} with id '{}' already exists for this project".format(self.topic[:-1],
- final_content["id"]),
- HTTPStatus.CONFLICT)
- # TODO validate with pyangbind. Load and dumps to convert data types
+ if self.db.get_one(self.topic, _filter, fail_on_empty=False):
+ raise EngineException("{} with id '{}' already exists for this project".format(self.topic[:-1],
+ final_content["id"]),
+ HTTPStatus.CONFLICT)
@staticmethod
def format_on_new(content, project_id=None, make_public=False):
clean_indata = clean_indata['vnfd:vnfd'][0]
return clean_indata
+ def check_conflict_on_edit(self, session, final_content, edit_content, _id, force=False):
+ super().check_conflict_on_edit(session, final_content, edit_content, _id, force=force)
+
+ # set type of vnfd
+ contains_pdu = False
+ contains_vdu = False
+ for vdu in get_iterable(final_content.get("vdu")):
+ if vdu.get("pdu-type"):
+ contains_pdu = True
+ else:
+ contains_vdu = True
+ if contains_pdu:
+ final_content["_admin"]["type"] = "hnfd" if contains_vdu else "pnfd"
+ elif contains_vdu:
+ final_content["_admin"]["type"] = "vnfd"
+ # if neither vud nor pdu do not fill type
+
def check_conflict_on_del(self, session, _id, force=False):
"""
Check that there is not any NSD that uses this VNFD. Only NSDs belonging to this project are considered. Note
raise EngineException("There is soame NSD that depends on this VNFD", http_code=HTTPStatus.CONFLICT)
def _validate_input_new(self, indata, force=False):
- # TODO validate with pyangbind, serialize
indata = self.pyangbind_validation("vnfds", indata, force)
# Cross references validation in the descriptor
- 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"]),
+ if indata.get("vdu"):
+ 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)
for vdu in get_iterable(indata.get("vdu")):
for interface in get_iterable(vdu.get("interface")):
"vnf-configuration:config-primitive:name"
.format(sgd["name"], sca["vnf-config-primitive-name-ref"]),
http_code=HTTPStatus.UNPROCESSABLE_ENTITY)
+ # TODO validata that if contains cloud-init-file or charms, have artifacts _admin.storage."pkg-dir" is not none
return indata
def _validate_input_edit(self, indata, force=False):
- # TODO validate with pyangbind, serialize
+ # not needed to validate with pyangbind becuase it will be validated at check_conflict_on_edit
return indata
return clean_indata
def _validate_input_new(self, indata, force=False):
-
- # TODO validate with pyangbind, serialize
indata = self.pyangbind_validation("nsds", indata, force)
+ # TODO validata that if contains cloud-init-file or charms, have artifacts _admin.storage."pkg-dir" is not none
return indata
def _validate_input_edit(self, indata, force=False):
- # TODO validate with pyangbind, serialize
+ # not needed to validate with pyangbind becuase it will be validated at check_conflict_on_edit
return indata
def _check_descriptor_dependencies(self, session, descriptor):
def check_conflict_on_edit(self, session, final_content, edit_content, _id, force=False):
super().check_conflict_on_edit(session, final_content, edit_content, _id, force=force)
- self._check_descriptor_dependencies(session, final_content)
+ if not force:
+ self._check_descriptor_dependencies(session, final_content)
def check_conflict_on_del(self, session, _id, force=False):
"""
@staticmethod
def format_on_new(content, project_id=None, make_public=False):
- BaseTopic.format_on_new(content, project_id=None, make_public=make_public)
+ BaseTopic.format_on_new(content, project_id=project_id, make_public=make_public)
content["_admin"]["onboardingState"] = "CREATED"
- content["_admin"]["operationalState"] = "DISABLED"
+ content["_admin"]["operationalState"] = "ENABLED"
content["_admin"]["usageState"] = "NOT_IN_USE"
def check_conflict_on_del(self, session, _id, force=False):