##
import asyncio
+import checksumdir
from collections import OrderedDict
+import os
from time import time
from osm_lcm.data_utils.database.database import Database
from osm_lcm.data_utils.filesystem.filesystem import Filesystem
return in_dict[in_key]
+def check_juju_bundle_existence(vnfd: dict) -> str:
+ """Checks the existence of juju-bundle in the descriptor
+
+ Args:
+ vnfd: Descriptor as a dictionary
+
+ Returns:
+ Juju bundle if dictionary has juju-bundle else None
+
+ """
+ if vnfd.get("vnfd"):
+ vnfd = vnfd["vnfd"]
+
+ for kdu in vnfd.get("kdu", []):
+ return kdu.get("juju-bundle", None)
+
+
+def get_charm_artifact_path(base_folder, charm_name, charm_type, revision=str()) -> str:
+ """Finds the charm artifact paths
+
+ Args:
+ base_folder: Main folder which will be looked up for charm
+ charm_name: Charm name
+ charm_type: Type of charm native_charm, lxc_proxy_charm or k8s_proxy_charm
+ revision: vnf package revision number if there is
+
+ Returns:
+ artifact_path: (str)
+
+ """
+ extension = ""
+ if revision:
+ extension = ":" + str(revision)
+
+ if base_folder.get("pkg-dir"):
+ artifact_path = "{}/{}/{}/{}".format(
+ base_folder["folder"].split(":")[0] + extension,
+ base_folder["pkg-dir"],
+ "charms"
+ if charm_type in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm")
+ else "helm-charts",
+ charm_name,
+ )
+
+ else:
+ # For SOL004 packages
+ artifact_path = "{}/Scripts/{}/{}".format(
+ base_folder["folder"].split(":")[0] + extension,
+ "charms"
+ if charm_type in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm")
+ else "helm-charts",
+ charm_name,
+ )
+
+ return artifact_path
+
+
def populate_dict(target_dict, key_list, value):
"""
Update target_dict creating nested dictionaries with the key_list. Last key_list item is asigned the value.
def update_db_2(self, item, _id, _desc):
"""
Updates database with _desc information. If success _desc is cleared
- :param item:
- :param _id:
+ :param item: collection
+ :param _id: the _id to use in the query filter
:param _desc: dictionary with the content to update. Keys are dot separated keys for
:return: None. Exception is raised on error
"""
# except DbException as e:
# self.logger.error("Updating {} _id={} with '{}'. Error: {}".format(item, _id, _desc, e))
+ def check_charm_hash_changed(
+ self, current_charm_path: str, target_charm_path: str
+ ) -> bool:
+ """Find the target charm has changed or not by checking the hash of
+ old and new charm packages
+
+ Args:
+ current_charm_path (str): Existing charm package artifact path
+ target_charm_path (str): Target charm package artifact path
+
+ Returns:
+ True/False (bool): if charm has changed it returns True
+
+ """
+ # Check if the charm artifacts are available
+ if os.path.exists(self.fs.path + current_charm_path) and os.path.exists(
+ self.fs.path + target_charm_path
+ ):
+ # Compare the hash of charm folders
+ if checksumdir.dirhash(
+ self.fs.path + current_charm_path
+ ) != checksumdir.dirhash(self.fs.path + target_charm_path):
+
+ return True
+
+ return False
+
+ else:
+ raise LcmException(
+ "Charm artifact {} does not exist in the VNF Package".format(
+ self.fs.path + target_charm_path
+ )
+ )
+
class TaskRegistry(LcmBase):
"""