Feature 10908: LCM process NS update request

[osm/LCM.git] / osm_lcm / lcm_utils.py

diff --git a/osm_lcm/lcm_utils.py b/osm_lcm/lcm_utils.py
index 16d5b33..235d485 100644 (file)
--- a/osm_lcm/lcm_utils.py
+++ b/osm_lcm/lcm_utils.py
@@ -17,7 +17,9 @@
 ##
 
 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
@@ -80,6 +82,63 @@ def get_iterable(in_dict, in_key):
     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"] + 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"] + 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.
@@ -124,6 +183,40 @@ class LcmBase:
         # 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):
     """