nsi_instantiate,
ns_migrate,
ns_verticalscale,
+ nslcmop_cancel,
)
from osm_nbi.base_topic import (
BaseTopic,
def __init__(self, db, fs, msg, auth):
BaseTopic.__init__(self, db, fs, msg, auth)
- def _check_descriptor_dependencies(self, session, descriptor):
- """
- Check that the dependent descriptors exist on a new descriptor or edition
- :param session: client session information
- :param descriptor: descriptor to be inserted or edit
- :return: None or raises exception
- """
- if not descriptor.get("nsdId"):
- return
- nsd_id = descriptor["nsdId"]
- if not self.get_item_list(session, "nsds", {"id": nsd_id}):
- raise EngineException(
- "Descriptor error at nsdId='{}' references a non exist nsd".format(
- nsd_id
- ),
- http_code=HTTPStatus.CONFLICT,
- )
-
@staticmethod
def format_on_new(content, project_id=None, make_public=False):
BaseTopic.format_on_new(content, project_id=project_id, make_public=make_public)
for config in df["lcm-operations-configuration"][
"operate-vnf-op-config"
].get("day1-2", []):
- # Verify the target object (VNF|NS|VDU|KDU) where we need to populate
+ # Verify the target object (VNF|NS|VDU|KDU) where we need to populate
# the params with the additional ones given by the user
if config.get("id") == selector:
for primitive in get_iterable(
"<rw_mgmt_ip>",
"<VDU_SCALE_INFO>",
"<ns_config_info>",
- "<OSM>"
+ "<OSM>",
):
continue
if (
EngineException, ValidationError, DbException, FsException, MsgException.
Note: Exceptions are not captured on purpose. They should be captured at called
"""
+ step = "checking quotas" # first step must be defined outside try
try:
- step = "checking quotas"
self.check_quota(session)
step = "validating input parameters"
return ns_k8s_namespace
- def _add_flavor_to_nsr(self, vdu, vnfd, nsr_descriptor, member_vnf_index, revision=None):
+ def _add_shared_volumes_to_nsr(
+ self, vdu, vnfd, nsr_descriptor, member_vnf_index, revision=None
+ ):
+ svsd = []
+ for vsd in vnfd.get("virtual-storage-desc", ()):
+ if vsd.get("vdu-storage-requirements"):
+ if (
+ vsd.get("vdu-storage-requirements")[0].get("key") == "multiattach"
+ and vsd.get("vdu-storage-requirements")[0].get("value") == "True"
+ ):
+ # Avoid setting the volume name multiple times
+ if not match(f"shared-.*-{vnfd['id']}", vsd["id"]):
+ vsd["id"] = f"shared-{vsd['id']}-{vnfd['id']}"
+ svsd.append(vsd)
+ if svsd:
+ nsr_descriptor["shared-volumes"] = svsd
+
+ def _add_flavor_to_nsr(
+ self, vdu, vnfd, nsr_descriptor, member_vnf_index, revision=None
+ ):
flavor_data = {}
guest_epa = {}
# Find this vdu compute and storage descriptors
if vsd.get("id") == vdu.get("virtual-storage-desc", [[]])[0]:
vdu_virtual_storage = vsd
# Get this vdu vcpus, memory and storage info for flavor_data
- if vdu_virtual_compute.get("virtual-cpu", {}).get(
- "num-virtual-cpu"
- ):
+ if vdu_virtual_compute.get("virtual-cpu", {}).get("num-virtual-cpu"):
flavor_data["vcpu-count"] = vdu_virtual_compute["virtual-cpu"][
"num-virtual-cpu"
]
if vdu_virtual_compute.get("virtual-memory", {}).get("size"):
flavor_data["memory-mb"] = (
- float(vdu_virtual_compute["virtual-memory"]["size"])
- * 1024.0
+ float(vdu_virtual_compute["virtual-memory"]["size"]) * 1024.0
)
if vdu_virtual_storage.get("size-of-storage"):
- flavor_data["storage-gb"] = vdu_virtual_storage[
- "size-of-storage"
- ]
+ flavor_data["storage-gb"] = vdu_virtual_storage["size-of-storage"]
# Get this vdu EPA info for guest_epa
if vdu_virtual_compute.get("virtual-cpu", {}).get("cpu-quota"):
- guest_epa["cpu-quota"] = vdu_virtual_compute["virtual-cpu"][
- "cpu-quota"
- ]
+ guest_epa["cpu-quota"] = vdu_virtual_compute["virtual-cpu"]["cpu-quota"]
if vdu_virtual_compute.get("virtual-cpu", {}).get("pinning"):
vcpu_pinning = vdu_virtual_compute["virtual-cpu"]["pinning"]
if vcpu_pinning.get("thread-policy"):
- guest_epa["cpu-thread-pinning-policy"] = vcpu_pinning[
- "thread-policy"
- ]
+ guest_epa["cpu-thread-pinning-policy"] = vcpu_pinning["thread-policy"]
if vcpu_pinning.get("policy"):
cpu_policy = (
- "SHARED"
- if vcpu_pinning["policy"] == "dynamic"
- else "DEDICATED"
+ "SHARED" if vcpu_pinning["policy"] == "dynamic" else "DEDICATED"
)
guest_epa["cpu-pinning-policy"] = cpu_policy
if vdu_virtual_compute.get("virtual-memory", {}).get("mem-quota"):
- guest_epa["mem-quota"] = vdu_virtual_compute["virtual-memory"][
- "mem-quota"
+ guest_epa["mem-quota"] = vdu_virtual_compute["virtual-memory"]["mem-quota"]
+ if vdu_virtual_compute.get("virtual-memory", {}).get("mempage-size"):
+ guest_epa["mempage-size"] = vdu_virtual_compute["virtual-memory"][
+ "mempage-size"
]
- if vdu_virtual_compute.get("virtual-memory", {}).get(
- "mempage-size"
- ):
- guest_epa["mempage-size"] = vdu_virtual_compute[
- "virtual-memory"
- ]["mempage-size"]
- if vdu_virtual_compute.get("virtual-memory", {}).get(
- "numa-node-policy"
- ):
- guest_epa["numa-node-policy"] = vdu_virtual_compute[
- "virtual-memory"
- ]["numa-node-policy"]
- if vdu_virtual_storage.get("disk-io-quota"):
- guest_epa["disk-io-quota"] = vdu_virtual_storage[
- "disk-io-quota"
+ if vdu_virtual_compute.get("virtual-memory", {}).get("numa-node-policy"):
+ guest_epa["numa-node-policy"] = vdu_virtual_compute["virtual-memory"][
+ "numa-node-policy"
]
+ if vdu_virtual_storage.get("disk-io-quota"):
+ guest_epa["disk-io-quota"] = vdu_virtual_storage["disk-io-quota"]
if guest_epa:
flavor_data["guest-epa"] = guest_epa
revision = revision if revision is not None else 1
- flavor_data["name"] = vdu["id"][:56] + "-" + member_vnf_index + "-" + str(revision) + "-flv"
+ flavor_data["name"] = (
+ vdu["id"][:56] + "-" + member_vnf_index + "-" + str(revision) + "-flv"
+ )
flavor_data["id"] = str(len(nsr_descriptor["flavor"]))
nsr_descriptor["flavor"].append(flavor_data)
"flavor": [],
"image": [],
"affinity-or-anti-affinity-group": [],
+ "shared-volumes": [],
+ "vnffgd": [],
}
if "revision" in nsd["_admin"]:
nsr_descriptor["revision"] = nsd["_admin"]["revision"]
for vdu in vnfd.get("vdu", ()):
member_vnf_index = vnf_profile.get("id")
self._add_flavor_to_nsr(vdu, vnfd, nsr_descriptor, member_vnf_index)
+ self._add_shared_volumes_to_nsr(
+ vdu, vnfd, nsr_descriptor, member_vnf_index
+ )
sw_image_id = vdu.get("sw-image-desc")
if sw_image_id:
image_data = self._get_image_data_from_vnfd(vnfd, sw_image_id)
)
vld["name"] = vld["id"]
nsr_descriptor["vld"] = nsr_vld
+ if nsd.get("vnffgd"):
+ vnffgd = nsd.get("vnffgd")
+ for vnffg in vnffgd:
+ info = {}
+ for k, v in vnffg.items():
+ if k == "id":
+ info.update({k: v})
+ if k == "nfpd":
+ info.update({k: v})
+ nsr_descriptor["vnffgd"].append(info)
return nsr_descriptor
if "revision" in vnfd:
vnfr_descriptor["revision"] = vnfd["revision"]
-
vnf_k8s_namespace = ns_k8s_namespace
if vnf_params:
if vnf_params.get("k8s-namespace"):
try:
vdu_virtual_storage_descriptors = utils.filter_in_list(
vnfd.get("virtual-storage-desc", []),
- lambda stg_desc: stg_desc["id"] in vdu["virtual-storage-desc"]
+ lambda stg_desc: stg_desc["id"] in vdu["virtual-storage-desc"],
)
except Exception:
vdu_virtual_storage_descriptors = []
"interfaces": [],
"additionalParams": additional_params,
"vdu-name": vdu["name"],
- "virtual-storages": vdu_virtual_storage_descriptors
+ "virtual-storages": vdu_virtual_storage_descriptors,
}
if vdu_params and vdu_params.get("config-units"):
vdur["config-units"] = vdu_params["config-units"]
# Name, mac-address and interface position is taken from VNFD
# and included into VNFR. By this way RO can process this information
# while creating the VDU.
- iface_fields = ("name", "mac-address", "position")
+ iface_fields = ("name", "mac-address", "position", "ip-address")
vdu_iface = {
x: iface[x] for x in iface_fields if iface.get(x) is not None
}
if (
cpd.get("constituent-cpd-id")
== iface_ext_cp
- ):
+ ) and vnf_profile.get("id") == vnf_index:
vdu_iface["ns-vld-id"] = vlc.get(
"virtual-link-profile-id"
)
vdur["alt-image-ids"] = alt_image_ids
revision = revision if revision is not None else 1
- flavor_data_name = vdu["id"][:56] + "-" + vnf_index + "-" + str(revision) + "-flv"
+ flavor_data_name = (
+ vdu["id"][:56] + "-" + vnf_index + "-" + str(revision) + "-flv"
+ )
nsr_flavor_desc = utils.find_in_list(
nsr_descriptor["flavor"],
lambda flavor: flavor["name"] == flavor_data_name,
if nsr_flavor_desc:
vdur["ns-flavor-id"] = nsr_flavor_desc["id"]
+ # Adding Shared Volume information to vdur
+ if vdur.get("virtual-storages"):
+ nsr_sv = []
+ for vsd in vdur["virtual-storages"]:
+ if vsd.get("vdu-storage-requirements"):
+ if (
+ vsd["vdu-storage-requirements"][0].get("key")
+ == "multiattach"
+ and vsd["vdu-storage-requirements"][0].get("value")
+ == "True"
+ ):
+ nsr_sv.append(vsd["id"])
+ if nsr_sv:
+ vdur["shared-volumes-id"] = nsr_sv
+
# Adding Affinity groups information to vdur
try:
vdu_profile_affinity_group = utils.find_in_list(
vdur["id"] = vdur["_id"]
vdur["count-index"] = index
vnfr_descriptor["vdur"].append(vdur)
-
return vnfr_descriptor
def vca_status_refresh(self, session, ns_instance_content, filter_q):
:param filter_q: dict: query parameter containing vcaStatus-refresh as true or false
:return: None
"""
- time_now, time_delta = time(), time() - ns_instance_content["_admin"]["modified"]
- force_refresh = isinstance(filter_q, dict) and filter_q.get('vcaStatusRefresh') == 'true'
+ time_now, time_delta = (
+ time(),
+ time() - ns_instance_content["_admin"]["modified"],
+ )
+ force_refresh = (
+ isinstance(filter_q, dict) and filter_q.get("vcaStatusRefresh") == "true"
+ )
threshold_reached = time_delta > 120
if force_refresh or threshold_reached:
operation, _id = "vca_status_refresh", ns_instance_content["_id"]
ns_instance_content["_admin"]["modified"] = time_now
self.db.set_one(self.topic, {"_id": _id}, ns_instance_content)
nslcmop_desc = NsLcmOpTopic._create_nslcmop(_id, operation, None)
- self.format_on_new(nslcmop_desc, session["project_id"], make_public=session["public"])
+ self.format_on_new(
+ nslcmop_desc, session["project_id"], make_public=session["public"]
+ )
nslcmop_desc["_admin"].pop("nsState")
self.msg.write("ns", operation, nslcmop_desc)
return
"terminate": ns_terminate,
"migrate": ns_migrate,
"verticalscale": ns_verticalscale,
+ "cancel": nslcmop_cancel,
}
def __init__(self, db, fs, msg, auth):
vnfd_id_2update = indata["changeVnfPackageData"]["vnfdId"]
if vnf_instance_id not in nsr["constituent-vnfr-ref"]:
-
raise EngineException(
f"Error in validating ns-update request: vnf {vnf_instance_id} does not "
f"belong to NS {ns_instance_id}",
# Check the given vnfd-id belongs to given vnf instance
if constituent_vnfd_id and (vnfd_id_2update != constituent_vnfd_id):
-
raise EngineException(
f"Error in validating ns-update request: vnfd-id {vnfd_id_2update} does not "
f"match with the vnfd-id: {constituent_vnfd_id} of VNF instance: {vnf_instance_id}",
"nsd:constituent-vnfd".format(member_vnf_index)
)
- ## Backwards compatibility: if there is no revision, get it from the one and only VNFD entry
+ # Backwards compatibility: if there is no revision, get it from the one and only VNFD entry
if "revision" in vnfr:
vnfd_revision = vnfr["vnfd-id"] + ":" + str(vnfr["revision"])
- vnfd = self.db.get_one("vnfds_revisions", {"_id": vnfd_revision}, fail_on_empty=False)
+ vnfd = self.db.get_one(
+ "vnfds_revisions", {"_id": vnfd_revision}, fail_on_empty=False
+ )
else:
- vnfd = self.db.get_one("vnfds", {"_id": vnfr["vnfd-id"]}, fail_on_empty=False)
+ vnfd = self.db.get_one(
+ "vnfds", {"_id": vnfr["vnfd-id"]}, fail_on_empty=False
+ )
if not vnfd:
raise EngineException(
ivld.get("id"): set()
for ivld in get_iterable(vnfd.get("int-virtual-link-desc"))
}
- for vdu in get_iterable(vnfd.get("vdu")):
- for cpd in get_iterable(vnfd.get("int-cpd")):
+ for vdu in vnfd.get("vdu", {}):
+ for cpd in vdu.get("int-cpd", {}):
if cpd.get("int-virtual-link-desc"):
vnfd_ivlds_cpds[cpd.get("int-virtual-link-desc")] = cpd.get("id")
return self.db.get_one("vim_accounts", db_filter)
except Exception:
raise EngineException(
- "Invalid vimAccountId='{}' not present for the project".format(
- vim_id
- )
+ "Invalid vimAccountId='{}' not present for the project".format(vim_id)
)
def _check_valid_wim_account(self, wim_account, wim_accounts, session):
return ifaces_forcing_vim_network
def _update_vnfrs_from_nsd(self, nsr):
+ step = "Getting vnf_profiles from nsd" # first step must be defined outside try
try:
nsr_id = nsr["_id"]
nsd = nsr["nsd"]
- step = "Getting vnf_profiles from nsd"
vnf_profiles = nsd.get("df", [{}])[0].get("vnf-profile", ())
vld_fixed_ip_connection_point_data = {}
for cpd in vlc.get("constituent-cpd-id", ()):
if cpd.get("ip-address"):
step = "Storing ip-address info"
- vld_fixed_ip_connection_point_data.update({vlc.get("virtual-link-profile-id") + '.' + cpd.get("constituent-base-element-id"): {
- "vnfd-connection-point-ref": cpd.get(
- "constituent-cpd-id"),
- "ip-address": cpd.get(
- "ip-address")}})
+ vld_fixed_ip_connection_point_data.update(
+ {
+ vlc.get("virtual-link-profile-id")
+ + "."
+ + cpd.get("constituent-base-element-id"): {
+ "vnfd-connection-point-ref": cpd.get(
+ "constituent-cpd-id"
+ ),
+ "ip-address": cpd.get("ip-address"),
+ }
+ }
+ )
# Inserting ip address to vnfr
if len(vld_fixed_ip_connection_point_data) > 0:
vnfrs = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
for item in vld_fixed_ip_connection_point_data.keys():
step = "Filtering vnfrs"
- vnfr = next(filter(lambda vnfr: vnfr["member-vnf-index-ref"] == item.split('.')[1], vnfrs), None)
+ vnfr = next(
+ filter(
+ lambda vnfr: vnfr["member-vnf-index-ref"]
+ == item.split(".")[1],
+ vnfrs,
+ ),
+ None,
+ )
if vnfr:
vnfr_update = {}
for vdur_index, vdur in enumerate(vnfr["vdur"]):
for iface_index, iface in enumerate(vdur["interfaces"]):
step = "Looking for matched interface"
if (
- iface.get("external-connection-point-ref")
- == vld_fixed_ip_connection_point_data[item].get("vnfd-connection-point-ref") and
- iface.get("ns-vld-id") == item.split('.')[0]
-
+ iface.get("external-connection-point-ref")
+ == vld_fixed_ip_connection_point_data[item].get(
+ "vnfd-connection-point-ref"
+ )
+ and iface.get("ns-vld-id") == item.split(".")[0]
):
vnfr_update_text = "vdur.{}.interfaces.{}".format(
vdur_index, iface_index
step = "Storing info in order to update vnfr"
vnfr_update[
vnfr_update_text + ".ip-address"
- ] = increment_ip_mac(
- vld_fixed_ip_connection_point_data[item].get("ip-address"),
- vdur.get("count-index", 0), )
+ ] = increment_ip_mac(
+ vld_fixed_ip_connection_point_data[item].get(
+ "ip-address"
+ ),
+ vdur.get("count-index", 0),
+ )
vnfr_update[vnfr_update_text + ".fixed-ip"] = True
step = "updating vnfr at database"
self.db.set_one("vnfrs", {"_id": vnfr["_id"]}, vnfr_update)
except (
- ValidationError,
- EngineException,
- DbException,
- MsgException,
- FsException,
+ ValidationError,
+ EngineException,
+ DbException,
+ MsgException,
+ FsException,
) as e:
raise type(e)("{} while '{}'".format(e, step), http_code=e.http_code)
HTTPStatus.CONFLICT,
)
self._check_ns_operation(session, nsr, operation, indata)
- if (indata.get("primitive_params")):
+ if indata.get("primitive_params"):
indata["primitive_params"] = json.dumps(indata["primitive_params"])
- elif (indata.get("additionalParamsForVnf")):
- indata["additionalParamsForVnf"] = json.dumps(indata["additionalParamsForVnf"])
+ elif indata.get("additionalParamsForVnf"):
+ indata["additionalParamsForVnf"] = json.dumps(
+ indata["additionalParamsForVnf"]
+ )
if operation == "instantiate":
self._update_vnfrs_from_nsd(nsr)
vnfd = self.db.get_one("vnfds", {"_id": vnfd_id})
nsd = self.db.get_one("nsds", {"_id": nsr["nsd-id"]})
ns_request = nsr["instantiate_params"]
- vnfr = self.db.get_one("vnfrs", {"_id": indata["changeVnfPackageData"]["vnfInstanceId"]})
+ vnfr = self.db.get_one(
+ "vnfrs", {"_id": indata["changeVnfPackageData"]["vnfInstanceId"]}
+ )
latest_vnfd_revision = vnfd["_admin"].get("revision", 1)
vnfr_vnfd_revision = vnfr.get("revision", 1)
if latest_vnfd_revision != vnfr_vnfd_revision:
old_vnfd_id = vnfd_id + ":" + str(vnfr_vnfd_revision)
- old_db_vnfd = self.db.get_one("vnfds_revisions", {"_id": old_vnfd_id})
+ old_db_vnfd = self.db.get_one(
+ "vnfds_revisions", {"_id": old_vnfd_id}
+ )
old_sw_version = old_db_vnfd.get("software-version", "1.0")
new_sw_version = vnfd.get("software-version", "1.0")
if new_sw_version != old_sw_version:
vnf_index = vnfr["member-vnf-index-ref"]
self.logger.info("nsr {}".format(nsr))
for vdu in vnfd["vdu"]:
- self.nsrtopic._add_flavor_to_nsr(vdu, vnfd, nsr, vnf_index, latest_vnfd_revision)
+ self.nsrtopic._add_shared_volumes_to_nsr(
+ vdu, vnfd, nsr, vnf_index, latest_vnfd_revision
+ )
+ self.nsrtopic._add_flavor_to_nsr(
+ vdu, vnfd, nsr, vnf_index, latest_vnfd_revision
+ )
sw_image_id = vdu.get("sw-image-desc")
if sw_image_id:
- image_data = self.nsrtopic._get_image_data_from_vnfd(vnfd, sw_image_id)
+ image_data = self.nsrtopic._get_image_data_from_vnfd(
+ vnfd, sw_image_id
+ )
self.nsrtopic._add_image_to_nsr(nsr, image_data)
for alt_image in vdu.get("alternative-sw-image-desc", ()):
- image_data = self.nsrtopic._get_image_data_from_vnfd(vnfd, alt_image)
+ image_data = self.nsrtopic._get_image_data_from_vnfd(
+ vnfd, alt_image
+ )
self.nsrtopic._add_image_to_nsr(nsr, image_data)
nsr_update["image"] = nsr["image"]
nsr_update["flavor"] = nsr["flavor"]
+ nsr_update["shared-volumes"] = nsr["shared-volumes"]
self.db.set_one("nsrs", {"_id": nsr["_id"]}, nsr_update)
- ns_k8s_namespace = self.nsrtopic._get_ns_k8s_namespace(nsd, ns_request, session)
- vnfr_descriptor = self.nsrtopic._create_vnfr_descriptor_from_vnfd(
- nsd,
- vnfd,
- vnfd_id,
- vnf_index,
- nsr,
- ns_request,
- ns_k8s_namespace,
- latest_vnfd_revision,
+ ns_k8s_namespace = self.nsrtopic._get_ns_k8s_namespace(
+ nsd, ns_request, session
+ )
+ vnfr_descriptor = (
+ self.nsrtopic._create_vnfr_descriptor_from_vnfd(
+ nsd,
+ vnfd,
+ vnfd_id,
+ vnf_index,
+ nsr,
+ ns_request,
+ ns_k8s_namespace,
+ latest_vnfd_revision,
+ )
)
indata["newVdur"] = vnfr_descriptor["vdur"]
nslcmop_desc = self._create_nslcmop(nsInstanceId, operation, indata)
# except DbException as e:
# raise EngineException("Cannot get ns_instance '{}': {}".format(e), HTTPStatus.NOT_FOUND)
+ def cancel(self, rollback, session, indata=None, kwargs=None, headers=None):
+ validate_input(indata, self.operation_schema["cancel"])
+ # Override descriptor with query string kwargs
+ self._update_input_with_kwargs(indata, kwargs, yaml_format=True)
+ nsLcmOpOccId = indata["nsLcmOpOccId"]
+ cancelMode = indata["cancelMode"]
+ # get nslcmop from nsLcmOpOccId
+ _filter = BaseTopic._get_project_filter(session)
+ _filter["_id"] = nsLcmOpOccId
+ nslcmop = self.db.get_one("nslcmops", _filter)
+ # Fail is this is not an ongoing nslcmop
+ if nslcmop.get("operationState") not in [
+ "STARTING",
+ "PROCESSING",
+ "ROLLING_BACK",
+ ]:
+ raise EngineException(
+ "Operation is not in STARTING, PROCESSING or ROLLING_BACK state",
+ http_code=HTTPStatus.CONFLICT,
+ )
+ nsInstanceId = nslcmop["nsInstanceId"]
+ update_dict = {
+ "isCancelPending": True,
+ "cancelMode": cancelMode,
+ }
+ self.db.set_one(
+ "nslcmops", q_filter=_filter, update_dict=update_dict, fail_on_empty=False
+ )
+ data = {
+ "_id": nsLcmOpOccId,
+ "nsInstanceId": nsInstanceId,
+ "cancelMode": cancelMode,
+ }
+ self.msg.write("nslcmops", "cancel", data)
+
def delete(self, session, _id, dry_run=False, not_send_msg=None):
raise EngineException(
"Method delete called directly", HTTPStatus.INTERNAL_SERVER_ERROR
additional_params[k] = "!!yaml " + safe_dump(v)
return additional_params
- 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("nst-ref"):
- return
- nstd_id = descriptor["nst-ref"]
- if not self.get_item_list(session, "nsts", {"id": nstd_id}):
- raise EngineException(
- "Descriptor error at nst-ref='{}' references a non exist nstd".format(
- nstd_id
- ),
- http_code=HTTPStatus.CONFLICT,
- )
-
def check_conflict_on_del(self, session, _id, db_content):
"""
Check that NSI is not instantiated
:return: the _id of nsi descriptor created at database
"""
+ step = "checking quotas" # first step must be defined outside try
try:
- step = "checking quotas"
self.check_quota(session)
step = ""
self.db.create("nsis", nsi_descriptor)
rollback.append({"topic": "nsis", "_id": nsi_id})
return nsi_id, None
+ except ValidationError as e:
+ raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)
except Exception as e: # TODO remove try Except, it is captured at nbi.py
self.logger.exception(
"Exception {} at NsiTopic.new()".format(e), exc_info=True
)
raise EngineException("Error {}: {}".format(step, e))
- except ValidationError as e:
- raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)
def edit(self, session, _id, indata=None, kwargs=None, content=None):
raise EngineException(
projects / osm / NBI.git / blobdiff