import asyncio
import checksumdir
from collections import OrderedDict
+import hashlib
import os
import shutil
import traceback
pass
-class LcmExceptionNoMgmtIP(LcmException):
- pass
-
-
class LcmExceptionExit(LcmException):
pass
target_dict[key_list[-1]] = value
+def get_ee_id_parts(ee_id):
+ """
+ Parses ee_id stored at database that can be either 'version:namespace.helm_id' or only
+ namespace.helm_id for backward compatibility
+ If exists helm version can be helm-v3 or helm (helm-v2 old version)
+ """
+ version, _, part_id = ee_id.rpartition(":")
+ namespace, _, helm_id = part_id.rpartition(".")
+ return version, namespace, helm_id
+
+
+def vld_to_ro_ip_profile(source_data):
+ if source_data:
+ return {
+ "ip_version": "IPv4"
+ if "v4" in source_data.get("ip-version", "ipv4")
+ else "IPv6",
+ "subnet_address": source_data.get("cidr")
+ or source_data.get("subnet-address"),
+ "gateway_address": source_data.get("gateway-ip")
+ or source_data.get("gateway-address"),
+ "dns_address": ";".join(
+ [v["address"] for v in source_data["dns-server"] if v.get("address")]
+ )
+ if source_data.get("dns-server")
+ else None,
+ "dhcp_enabled": source_data.get("dhcp-params", {}).get("enabled", False)
+ or source_data.get("dhcp-enabled", False),
+ "dhcp_start_address": source_data["dhcp-params"].get("start-address")
+ if source_data.get("dhcp-params")
+ else None,
+ "dhcp_count": source_data["dhcp-params"].get("count")
+ if source_data.get("dhcp-params")
+ else None,
+ "ipv6_address_mode": source_data["ipv6-address-mode"]
+ if "ipv6-address-mode" in source_data
+ else None,
+ }
+
+
class LcmBase:
def __init__(self, msg, logger):
"""
# except DbException as e:
# self.logger.error("Updating {} _id={} with '{}'. Error: {}".format(item, _id, _desc, e))
+ @staticmethod
+ def calculate_charm_hash(zipped_file):
+ """Calculate the hash of charm files which ends with .charm
+
+ Args:
+ zipped_file (str): Existing charm package full path
+
+ Returns:
+ hex digest (str): The hash of the charm file
+ """
+ filehash = hashlib.sha256()
+ with open(zipped_file, mode="rb") as file:
+ contents = file.read()
+ filehash.update(contents)
+ return filehash.hexdigest()
+
+ @staticmethod
+ def compare_charm_hash(current_charm, target_charm):
+ """Compare the existing charm and the target charm if the charms
+ are given as zip files ends with .charm
+
+ Args:
+ current_charm (str): Existing charm package full path
+ target_charm (str): Target charm package full path
+
+ Returns:
+ True/False (bool): if charm has changed it returns True
+ """
+ return LcmBase.calculate_charm_hash(
+ current_charm
+ ) != LcmBase.calculate_charm_hash(target_charm)
+
+ @staticmethod
+ def compare_charmdir_hash(current_charm_dir, target_charm_dir):
+ """Compare the existing charm and the target charm if the charms
+ are given as directories
+
+ Args:
+ current_charm_dir (str): Existing charm package directory path
+ target_charm_dir (str): Target charm package directory path
+
+ Returns:
+ True/False (bool): if charm has changed it returns True
+ """
+ return checksumdir.dirhash(current_charm_dir) != checksumdir.dirhash(
+ target_charm_dir
+ )
+
def check_charm_hash_changed(
self, current_charm_path: str, target_charm_path: str
) -> bool:
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):
+ try:
+ # Check if the charm artifacts are available
+ current_charm = self.fs.path + current_charm_path
+ target_charm = self.fs.path + target_charm_path
- return True
+ if os.path.exists(current_charm) and os.path.exists(target_charm):
+ # Compare the hash of .charm files
+ if current_charm.endswith(".charm"):
+ return LcmBase.compare_charm_hash(current_charm, target_charm)
- return False
+ # Compare the hash of charm folders
+ return LcmBase.compare_charmdir_hash(current_charm, target_charm)
- else:
- raise LcmException(
- "Charm artifact {} does not exist in the VNF Package".format(
- self.fs.path + target_charm_path
+ else:
+ raise LcmException(
+ "Charm artifact {} does not exist in the VNF Package".format(
+ self.fs.path + target_charm_path
+ )
)
- )
+ except (IOError, OSError, TypeError) as error:
+ self.logger.debug(traceback.format_exc())
+ self.logger.error(f"{error} occured while checking the charm hashes")
+ raise LcmException(error)
@staticmethod
def get_charm_name(charm_metadata_file: str) -> str:
# Get the NSD package path
if revision:
-
- nsd_package_path = (
- db_nsr["nsd-id"] + ":" + str(revision)
- )
+ nsd_package_path = db_nsr["nsd-id"] + ":" + str(revision)
db_nsd = self.db.get_one("nsds_revisions", {"_id": nsd_package_path})
else:
projects / osm / LCM.git / blobdiff