Feature 10944 Change naming of charms

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

diff --git a/osm_lcm/lcm_utils.py b/osm_lcm/lcm_utils.py
index 1854da7..6d20318 100644 (file)
--- a/osm_lcm/lcm_utils.py
+++ b/osm_lcm/lcm_utils.py
@@ -20,9 +20,15 @@ import asyncio
 import checksumdir
 from collections import OrderedDict
 import os
+import shutil
+import traceback
 from time import time
+
+from osm_common.fsbase import FsException
 from osm_lcm.data_utils.database.database import Database
 from osm_lcm.data_utils.filesystem.filesystem import Filesystem
+import yaml
+from zipfile import ZipFile, BadZipfile
 
 # from osm_common.dbbase import DbException
 
@@ -217,6 +223,133 @@ class LcmBase:
                 )
             )
 
+    @staticmethod
+    def get_charm_name(charm_metadata_file: str) -> str:
+        """Get the charm name from metadata file.
+
+        Args:
+            charm_metadata_file    (str):  charm metadata file full path
+
+        Returns:
+            charm_name    (str):  charm name
+
+        """
+        # Read charm metadata.yaml to get the charm name
+        with open(charm_metadata_file, "r") as metadata_file:
+            content = yaml.safe_load(metadata_file)
+            charm_name = content["name"]
+            return str(charm_name)
+
+    def _get_charm_path(
+        self, nsd_package_path: str, nsd_package_name: str, charm_folder_name: str
+    ) -> str:
+        """Get the full path of charm folder.
+
+        Args:
+            nsd_package_path    (str):  NSD package full path
+            nsd_package_name    (str):  NSD package name
+            charm_folder_name   (str):  folder name
+
+        Returns:
+            charm_path    (str):  charm folder full path
+        """
+        charm_path = (
+            self.fs.path
+            + nsd_package_path
+            + "/"
+            + nsd_package_name
+            + "/charms/"
+            + charm_folder_name
+        )
+        return charm_path
+
+    def _get_charm_metadata_file(
+        self,
+        charm_folder_name: str,
+        nsd_package_path: str,
+        nsd_package_name: str,
+        charm_path: str = None,
+    ) -> str:
+        """Get the path of charm metadata file.
+
+        Args:
+            charm_folder_name   (str):  folder name
+            nsd_package_path    (str):  NSD package full path
+            nsd_package_name    (str):  NSD package name
+            charm_path  (str):  Charm full path
+
+        Returns:
+            charm_metadata_file_path    (str):  charm metadata file full path
+
+        """
+        # Locate the charm metadata.yaml
+        if charm_folder_name.endswith(".charm"):
+            extract_path = (
+                self.fs.path
+                + nsd_package_path
+                + "/"
+                + nsd_package_name
+                + "/charms/"
+                + charm_folder_name.split(".")[0]
+            )
+            # Extract .charm to extract path
+            with ZipFile(charm_path, "r") as zipfile:
+                zipfile.extractall(extract_path)
+            return extract_path + "/metadata.yaml"
+        else:
+            return charm_path + "/metadata.yaml"
+
+    def find_charm_name(self, db_nsr: dict, charm_folder_name: str) -> str:
+        """Get the charm name from metadata.yaml of charm package.
+
+        Args:
+            db_nsr  (dict): NS record as a dictionary
+            charm_folder_name   (str): charm folder name
+
+        Returns:
+             charm_name (str):  charm name
+        """
+        try:
+            if not charm_folder_name:
+                raise LcmException("charm_folder_name should be provided.")
+
+            # Find nsd_package details: path, name
+            revision = db_nsr.get("revision", "")
+            nsd_package_path = (
+                db_nsr["nsd-id"] + ":" + str(revision) if revision else db_nsr["nsd-id"]
+            )
+            nsd_package_name = os.listdir(self.fs.path + nsd_package_path)[0]
+
+            # Remove the existing nsd package and sync from FsMongo
+            shutil.rmtree(self.fs.path + nsd_package_path, ignore_errors=True)
+            self.fs.sync(from_path=nsd_package_path)
+
+            # Get the charm path
+            charm_path = self._get_charm_path(
+                nsd_package_path, nsd_package_name, charm_folder_name
+            )
+
+            # Find charm metadata file full path
+            charm_metadata_file = self._get_charm_metadata_file(
+                charm_folder_name, nsd_package_path, nsd_package_name, charm_path
+            )
+
+            # Return charm name
+            return self.get_charm_name(charm_metadata_file)
+
+        except (
+            yaml.YAMLError,
+            IOError,
+            FsException,
+            KeyError,
+            TypeError,
+            FileNotFoundError,
+            BadZipfile,
+        ) as error:
+            self.logger.debug(traceback.format_exc())
+            self.logger.error(f"{error} occured while getting the charm name")
+            raise LcmException(error)
+
 
 class TaskRegistry(LcmBase):
     """