if len(parts) == 2:
kdu_model = parts[0]
if desc_params.get("kdu_atomic_upgrade"):
- atomic_upgrade = desc_params.get("kdu_atomic_upgrade").lower() in ("yes", "true", "1")
+ atomic_upgrade = desc_params.get(
+ "kdu_atomic_upgrade"
+ ).lower() in ("yes", "true", "1")
del desc_params["kdu_atomic_upgrade"]
else:
atomic_upgrade = True
current_charm_artifact_path,
target_charm_artifact_path,
charm_artifact_paths,
- helm_artifacts
+ helm_artifacts,
) = ([], [], [], [])
step = "Checking if revision has changed in VNFD"
# add chart to list and all parameters
step = "Getting helm chart name"
chart_name = ee_item.get("helm-chart")
- if ee_item.get("helm-version") and ee_item.get("helm-version") == "v2":
+ if (
+ ee_item.get("helm-version")
+ and ee_item.get("helm-version") == "v2"
+ ):
vca_type = "helm"
else:
vca_type = "helm-v3"
step = "Setting Helm chart artifact paths"
- helm_artifacts.append({
- "current_artifact_path": get_charm_artifact_path(
- current_base_folder,
- chart_name,
- vca_type,
- current_vnf_revision,
- ),
- "target_artifact_path": get_charm_artifact_path(
- latest_base_folder,
- chart_name,
- vca_type,
- latest_vnfd_revision,
- ),
- "ee_id": ee_id,
- "vca_index": vca_index,
- "vdu_index": vdu_count_index,
- })
+ helm_artifacts.append(
+ {
+ "current_artifact_path": get_charm_artifact_path(
+ current_base_folder,
+ chart_name,
+ vca_type,
+ current_vnf_revision,
+ ),
+ "target_artifact_path": get_charm_artifact_path(
+ latest_base_folder,
+ chart_name,
+ vca_type,
+ latest_vnfd_revision,
+ ),
+ "ee_id": ee_id,
+ "vca_index": vca_index,
+ "vdu_index": vdu_count_index,
+ }
+ )
charm_artifact_paths = zip(
current_charm_artifact_path, target_charm_artifact_path
# helm base EE
for item in helm_artifacts:
- if (
- not (
- item["current_artifact_path"]
- and item["target_artifact_path"]
- and self.check_charm_hash_changed(
- item["current_artifact_path"], item["target_artifact_path"]
- )
- )
+ if not (
+ item["current_artifact_path"]
+ and item["target_artifact_path"]
+ and self.check_charm_hash_changed(
+ item["current_artifact_path"],
+ item["target_artifact_path"],
+ )
):
continue
- db_update_entry = "_admin.deployed.VCA.{}.".format(item["vca_index"])
+ db_update_entry = "_admin.deployed.VCA.{}.".format(
+ item["vca_index"]
+ )
vnfr_id = db_vnfr["_id"]
osm_config = {"osm": {"ns_id": nsr_id, "vnf_id": vnfr_id}}
db_dict = {
"path": db_update_entry,
}
vca_type, namespace, helm_id = get_ee_id_parts(item["ee_id"])
- await self.vca_map[
- vca_type
- ].upgrade_execution_environment(
+ await self.vca_map[vca_type].upgrade_execution_environment(
namespace=namespace,
helm_id=helm_id,
db_dict=db_dict,
self.logger.debug("get ssh key block")
rw_mgmt_ip = None
if deep_get(
- config_descriptor, ("config-access", "ssh-access", "required")
+ config_descriptor,
+ ("config-access", "ssh-access", "required"),
):
# Needed to inject a ssh key
user = deep_get(
config_descriptor,
("config-access", "ssh-access", "default-user"),
)
- step = "Install configuration Software, getting public ssh key"
- pub_key = await self.vca_map[vca_type].get_ee_ssh_public__key(
+ step = (
+ "Install configuration Software, getting public ssh key"
+ )
+ pub_key = await self.vca_map[
+ vca_type
+ ].get_ee_ssh_public__key(
ee_id=ee_id, db_dict=db_dict, vca_id=vca_id
)
- step = "Insert public key into VM user={} ssh_key={}".format(
- user, pub_key
+ step = (
+ "Insert public key into VM user={} ssh_key={}".format(
+ user, pub_key
+ )
)
self.logger.debug(logging_text + step)
"initial-config-primitive"
)
config_primitive = next(
- (p for p in initial_config_primitive_list if p["name"] == "config"),
+ (
+ p
+ for p in initial_config_primitive_list
+ if p["name"] == "config"
+ ),
None,
)
if not config_primitive:
deploy_params["rw_mgmt_ip"] = rw_mgmt_ip
if db_vnfr.get("additionalParamsForVnf"):
deploy_params.update(
- parse_yaml_strings(db_vnfr["additionalParamsForVnf"].copy())
+ parse_yaml_strings(
+ db_vnfr["additionalParamsForVnf"].copy()
+ )
)
primitive_params_ = self._map_primitive_params(
config_primitive, {}, deploy_params