if self.db.get_one(item, filter, fail_on_empty=False):
raise EngineException("{} with id '{}' already exists for this tenant".format(item[:-1], indata["id"]),
HTTPStatus.CONFLICT)
+ # TODO validate with pyangbind. Load and dumps to convert data types
+ if item == "nsds":
+ # transform constituent-vnfd:member-vnf-index to string
+ if indata.get("constituent-vnfd"):
+ for constituent_vnfd in indata["constituent-vnfd"]:
+ if "member-vnf-index" in constituent_vnfd:
+ constituent_vnfd["member-vnf-index"] = str(constituent_vnfd["member-vnf-index"])
- # TODO validate with pyangbind
if item == "nsds" and not force:
self._check_descriptor_dependencies(session, "nsds", indata)
elif item == "userDefinedData":
indata["_admin"]["projects_read"] = [session["project_id"]]
if not indata["_admin"].get("projects_write"):
indata["_admin"]["projects_write"] = [session["project_id"]]
- if item == "nsds":
- # transform constituent-vnfd:member-vnf-index to string
- if indata.get("constituent-vnfd"):
- for constituent_vnfd in indata["constituent-vnfd"]:
- if "member-vnf-index" in constituent_vnfd:
- constituent_vnfd["member-vnf-index"] = str(constituent_vnfd["member-vnf-index"])
if item in ("vnfds", "nsds"):
indata["_admin"]["onboardingState"] = "CREATED"
indata["_admin"]["operationalState"] = "DISABLED"
"created-time": now,
# "vnfd": vnfd, # at OSM model.but removed to avoid data duplication TODO: revise
"vnfd-ref": vnfd_id,
- "vnfd-id": vnfr_id, # not at OSM model, but useful
+ "vnfd-id": vnfd["_id"], # not at OSM model, but useful
"vim-account-id": None,
"vdur": [],
"connection-point": [],
return folder_content, "text/plain"
# TODO manage folders in http
else:
- return self.fs.file_open((storage['folder'], storage['pkg-dir'], *path), "rb"), \
- "application/octet-stream"
+ return self.fs.file_open((storage['folder'], storage['pkg-dir'], *path), "rb"),\
+ "application/octet-stream"
# pkgtype accept ZIP TEXT -> result
# manyfiles yes X -> zip
self._add_delete_filter(session, item, filter)
if item in ("vnfds", "nsds") and not force:
descriptor = self.get_item(session, item, _id)
- descriptor_id = descriptor["id"]
- self._check_dependencies_on_descriptor(session, item, descriptor_id)
+ descriptor_id = descriptor.get("id")
+ if descriptor_id:
+ self._check_dependencies_on_descriptor(session, item, descriptor_id)
if item == "nsrs":
nsr = self.db.get_one(item, filter)
- if nsr["_admin"]["nsState"] == "INSTANTIATED" and not force:
+ if nsr["_admin"].get("nsState") == "INSTANTIATED" and not force:
raise EngineException("nsr '{}' cannot be deleted because it is in 'INSTANTIATED' state. "
"Launch 'terminate' operation first; or force deletion".format(_id),
http_code=HTTPStatus.CONFLICT)
self.db.del_list("vnfrs", {"nsr-id-ref": _id})
self.msg.write("ns", "deleted", {"_id": _id})
return v
- if item in ("vim_accounts", "sdns"):
- desc = self.db.get_one(item, filter)
- desc["_admin"]["to_delete"] = True
- self.db.replace(item, _id, desc) # TODO change to set_one
+ if item in ("vim_accounts", "sdns") and not force:
+ self.db.set_one(item, {"_id": _id}, {"_admin.to_delete": True}) # TODO change status
if item == "vim_accounts":
self.msg.write("vim_account", "delete", {"_id": _id})
elif item == "sdns":
return {"deleted": 1} # TODO indicate an offline operation to return 202 ACCEPTED
v = self.db.del_one(item, filter)
- self.fs.file_delete(_id, ignore_non_exist=True)
+ if item in ("vnfds", "nsds"):
+ self.fs.file_delete(_id, ignore_non_exist=True)
+ if item in ("vim_accounts", "sdns", "vnfds", "nsds"):
+ self.msg.write(item[:-1], "deleted", {"_id": _id})
return v
def prune(self):