X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_lcm%2Fdata_utils%2Fvnfd.py;h=954a84c0ef645c8ad7aaf55fed296893ce8d7efe;hb=7f26b3f6740e12dd35d8024b14c3fe3a32ec5dbf;hp=668fab9318a80c9293b78098ee2894a954a9bc0a;hpb=832f899a8fa59e67a1e5e34356ef192b835b79a7;p=osm%2FLCM.git diff --git a/osm_lcm/data_utils/vnfd.py b/osm_lcm/data_utils/vnfd.py index 668fab9..954a84c 100644 --- a/osm_lcm/data_utils/vnfd.py +++ b/osm_lcm/data_utils/vnfd.py @@ -25,10 +25,6 @@ from osm_lcm.data_utils import list_utils -def get_lcm_operations_configuration(vnfd): - return vnfd.get("df", ())[0].get("lcm-operations-configuration", ()) - - def get_vdu_list(vnfd): return vnfd.get("vdu", ()) @@ -37,7 +33,19 @@ def get_kdu_list(vnfd): return vnfd.get("kdu", ()) -def get_ee_sorted_initial_config_primitive_list(primitive_list, vca_deployed, ee_descriptor_id): +def get_kdu(vnfd, kdu_name): + return list_utils.find_in_list( + get_kdu_list(vnfd), lambda kdu: kdu["name"] == kdu_name + ) + + +def get_kdu_services(kdu): + return kdu.get("service", []) + + +def get_ee_sorted_initial_config_primitive_list( + primitive_list, vca_deployed, ee_descriptor_id +): """ Generates a list of initial-config-primitive based on the list provided by the descriptor. It includes internal primitives as verify-ssh-credentials, or config when needed @@ -50,10 +58,12 @@ def get_ee_sorted_initial_config_primitive_list(primitive_list, vca_deployed, ee """ primitive_list = primitive_list or [] primitive_list = [ - p for p in primitive_list if p.get("execution-environment-ref", ee_descriptor_id) == ee_descriptor_id + p + for p in primitive_list + if p.get("execution-environment-ref", ee_descriptor_id) == ee_descriptor_id ] if primitive_list: - primitive_list.sort(key=lambda val: int(val['seq'])) + primitive_list.sort(key=lambda val: int(val["seq"])) # look for primitive config, and get the position. None if not present config_position = None @@ -67,54 +77,58 @@ def get_ee_sorted_initial_config_primitive_list(primitive_list, vca_deployed, ee primitive_list.insert(0, {"name": "config", "parameter": []}) config_position = 0 # TODO revise if needed: for VNF/VDU add verify-ssh-credentials after config - if vca_deployed["member-vnf-index"] and config_position is not None and vca_deployed.get("ssh-public-key"): - primitive_list.insert(config_position + 1, {"name": "verify-ssh-credentials", "parameter": []}) + if ( + vca_deployed["member-vnf-index"] + and config_position is not None + and vca_deployed.get("ssh-public-key") + ): + primitive_list.insert( + config_position + 1, {"name": "verify-ssh-credentials", "parameter": []} + ) return primitive_list def get_ee_sorted_terminate_config_primitive_list(primitive_list, ee_descriptor_id): primitive_list = primitive_list or [] primitive_list = [ - p for p in primitive_list if p.get("execution-environment-ref", ee_descriptor_id) == ee_descriptor_id + p + for p in primitive_list + if p.get("execution-environment-ref", ee_descriptor_id) == ee_descriptor_id ] if primitive_list: - primitive_list.sort(key=lambda val: int(val['seq'])) + primitive_list.sort(key=lambda val: int(val["seq"])) return primitive_list def get_vdu_profile(vnfd, vdu_profile_id): return list_utils.find_in_list( vnfd.get("df", ())[0]["vdu-profile"], - lambda vdu_profile: vdu_profile["id"] == vdu_profile_id) + lambda vdu_profile: vdu_profile["id"] == vdu_profile_id, + ) -def get_kdu_profile(vnfd, kdu_profile_id): +def get_kdu_resource_profile(vnfd, kdu_profile_id): return list_utils.find_in_list( - vnfd.get("df", ())[0]["kdu-profile"], - lambda kdu_profile: kdu_profile["id"] == kdu_profile_id) + vnfd.get("df", ())[0]["kdu-resource-profile"], + lambda kdu_profile: kdu_profile["id"] == kdu_profile_id, + ) -def get_vnf_configuration(vnfd): - if "vnf-configuration-id" not in vnfd.get("df")[0]: +def get_configuration(vnfd, entity_id): + lcm_ops_config = vnfd.get("df")[0].get("lcm-operations-configuration") + if not lcm_ops_config: return None - vnf_config_id = vnfd.get("df")[0]["vnf-configuration-id"] - return list_utils.find_in_list( - vnfd.get("vnf-configuration", {}), - lambda vnf_config: vnf_config["id"] == vnf_config_id) - - -def get_vdu_configuration(vnfd, vdu_id): - vdu_profile = get_vdu_profile(vnfd, vdu_id) + ops_vnf = lcm_ops_config.get("operate-vnf-op-config") + if not ops_vnf: + return None + day12ops = ops_vnf.get("day1-2", []) return list_utils.find_in_list( - vnfd.get("vdu-configuration", ()), - lambda vdu_configuration: vdu_configuration["id"] == vdu_profile["vdu-configuration-id"]) + day12ops, lambda configuration: configuration["id"] == entity_id + ) -def get_kdu_configuration(vnfd, kdu_id): - kdu_profile = get_kdu_profile(vnfd, kdu_id) - return list_utils.find_in_list( - vnfd.get("kdu-configuration", ()), - lambda kdu_configuration: kdu_configuration["id"] == kdu_profile["kdu-configuration-id"]) +def get_relation_list(vnfd, entity_id): + return (get_configuration(vnfd, entity_id) or {}).get("relation", []) def get_virtual_link_profiles(vnfd): @@ -122,15 +136,13 @@ def get_virtual_link_profiles(vnfd): def get_vdu(vnfd, vdu_id): - return list_utils.find_in_list( - vnfd.get("vdu", ()), - lambda vdu: vdu["id"] == vdu_id) + return list_utils.find_in_list(vnfd.get("vdu", ()), lambda vdu: vdu["id"] == vdu_id) def get_vdu_index(vnfd, vdu_id): target_vdu = list_utils.find_in_list( - vnfd.get("vdu", ()), - lambda vdu: vdu["id"] == vdu_id) + vnfd.get("vdu", ()), lambda vdu: vdu["id"] == vdu_id + ) if target_vdu: return vnfd.get("vdu", ()).index(target_vdu) else: @@ -143,15 +155,47 @@ def get_scaling_aspect(vnfd): def get_number_of_instances(vnfd, vdu_id): return list_utils.find_in_list( - vnfd.get( - "df", - () - )[0].get( - "instantiation-level", - () - )[0].get( - "vdu-level", - () - ), - lambda a_vdu: a_vdu["vdu-id"] == vdu_id - )["number-of-instances"] + vnfd.get("df", ())[0].get("instantiation-level", ())[0].get("vdu-level", ()), + lambda a_vdu: a_vdu["vdu-id"] == vdu_id, + ).get("number-of-instances", 1) + + +def get_juju_ee_ref(vnfd, entity_id): + return list_utils.find_in_list( + get_configuration(vnfd, entity_id).get("execution-environment-list", []), + lambda ee: "juju" in ee, + ) + + +def find_software_version(vnfd: dict) -> str: + """Find the sotware version in the VNFD descriptors + + Args: + vnfd (dict): Descriptor as a dictionary + + Returns: + software-version (str) + """ + + default_sw_version = "1.0" + + if vnfd.get("vnfd"): + vnfd = vnfd["vnfd"] + + if vnfd.get("software-version"): + return vnfd["software-version"] + + else: + return default_sw_version + + +def check_helm_ee_in_ns(db_vnfds: list) -> bool: + for vnfd in db_vnfds: + descriptor_config = get_configuration(vnfd, vnfd["id"]) + if not ( + descriptor_config and "execution-environment-list" in descriptor_config + ): + continue + ee_list = descriptor_config.get("execution-environment-list", []) + if list_utils.find_in_list(ee_list, lambda ee_item: "helm-chart" in ee_item): + return True