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
"""
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
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_resource_profile(vnfd, kdu_profile_id):
+ return list_utils.find_in_list(
+ vnfd.get("df", ())[0]["kdu-resource-profile"],
+ lambda kdu_profile: kdu_profile["id"] == kdu_profile_id,
+ )
def get_configuration(vnfd, entity_id):
if not ops_vnf:
return None
day12ops = ops_vnf.get("day1-2", [])
- list_utils.find_in_list(
- day12ops,
- lambda configuration: configuration["id"] == entity_id)
+ return list_utils.find_in_list(
+ day12ops, lambda configuration: configuration["id"] == entity_id
+ )
+
+
+def get_relation_list(vnfd, entity_id):
+ return (get_configuration(vnfd, entity_id) or {}).get("relation", [])
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:
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,
+ )