for cp in vnfd.get("ext-cpd", ()):
vnf_cp = {
"name": cp.get("id"),
- "connection-point-id": cp.get("int-cpd").get("cpd"),
- "connection-point-vdu-id": cp.get("int-cpd").get("vdu-id"),
+ "connection-point-id": cp.get("int-cpd", {}).get("cpd"),
+ "connection-point-vdu-id": cp.get("int-cpd", {}).get("vdu-id"),
"id": cp.get("id"),
# "ip-address", "mac-address" # filled by LCM
# vim-id # TODO it would be nice having a vim port id
net["external-connection-point-ref"] = all_k8s_cluster_nets_cpds[net.get("id")]
# update kdus
- # TODO: Change for multiple df support
- all_kdu_profiles = vnfd.get("df", [[]])[0].get("kdu-profile", ())
- all_kdu_profiles_models = {profile.get("name"): profile.get("kdu-model-id") for profile in all_kdu_profiles}
- all_kdu_models = vnfd.get("kdu-model", ())
- all_kdu_models = {model.get("id"): model for model in all_kdu_models}
for kdu in get_iterable(vnfd.get("kdu")):
additional_params, kdu_params = self._format_additional_params(ns_request,
vnf_index,
kdur = {
"additionalParams": additional_params,
"k8s-namespace": kdu_k8s_namespace,
- "kdu-name": kdu.get("name"),
+ "kdu-name": kdu["name"],
# TODO "name": "" Name of the VDU in the VIM
"ip-address": None, # mgmt-interface filled by LCM
"k8s-cluster": {},
}
if kdu_params and kdu_params.get("config-units"):
kdur["config-units"] = kdu_params["config-units"]
-
- kdu_model_data = all_kdu_models[all_kdu_profiles_models[kdur["name"]]]
- kdur[kdu_model_data.get("kdu-model-type")] = kdu_model or kdu_model_data
+ if kdu.get("helm-version"):
+ kdur["helm-version"] = kdu["helm-version"]
+ for k8s_type in ("helm-chart", "juju-bundle"):
+ if kdu.get(k8s_type):
+ kdur[k8s_type] = kdu_model or kdu[k8s_type]
if not vnfr_descriptor.get("kdur"):
vnfr_descriptor["kdur"] = []
vnfr_descriptor["kdur"].append(kdur)
elif indata.get("kdu_name"):
self._check_valid_kdu(vnfd, indata["kdu_name"])
# TODO: Change the [0] as kdu-configuration is now a list
- descriptor_configuration = vnfd.get("kdu-configuration", [{}])[0].get("config-primitive")
+ kdud = next((k for k in vnfd["kdu"] if k["name"] == indata["kdu_name"]), None)
+ descriptor_configuration = deep_get(kdud, ("kdu-configuration", "config-primitive"))
else:
# TODO: Change the [0] as vnf-configuration is now a list
descriptor_configuration = vnfd.get("vnf-configuration", [{}])[0].get("config-primitive")
def _check_scale_ns_operation(self, indata, nsr):
vnfd = self._get_vnfd_from_vnf_member_index(indata["scaleVnfData"]["scaleByStepData"]["member-vnf-index"],
nsr["_id"])
- for scaling_group in get_iterable(vnfd.get("scaling-group-descriptor")):
- if indata["scaleVnfData"]["scaleByStepData"]["scaling-group-descriptor"] == scaling_group["name"]:
+ for scaling_aspect in get_iterable(vnfd.get("df", ())[0]["scaling-aspect"]):
+ if indata["scaleVnfData"]["scaleByStepData"]["scaling-group-descriptor"] == scaling_aspect["id"]:
break
else:
raise EngineException("Invalid scaleVnfData:scaleByStepData:scaling-group-descriptor '{}' is not "
- "present at vnfd:scaling-group-descriptor"
+ "present at vnfd:scaling-aspect"
.format(indata["scaleVnfData"]["scaleByStepData"]["scaling-group-descriptor"]))
def _check_instantiate_ns_operation(self, indata, nsr, session):
vnfr_update[vnfr_update_text + ".mac-address"] = increment_ip_mac(
iface_inst_param.get("mac-address"), vdur.get("count-index", 0))
vnfr_update[vnfr_update_text + ".fixed-mac"] = True
+ if iface_inst_param.get("floating-ip-required"):
+ vnfr_update[vnfr_update_text + ".floating-ip-required"] = True
# get vnf.internal-vld.internal-conection-point instantiation params to update vnfr.vdur.interfaces
# TODO update vld with the ip-profile
for ivld_inst_param in get_iterable(vnf_inst_params.get("internal-vld")):