#!/usr/bin/python # coding: utf-8 -*- """ CVP Uploader content """ import os from dataclasses import dataclass from typing import Any, List, Optional from cvprac.cvp_client import CvpClient from cvprac.cvp_client_errors import CvpLoginError from loguru import logger # from eos_downloader.tools import exc_to_str # logger = logging.getLogger(__name__) @dataclass class CvpAuthenticationItem: """ Data structure to represent Cloudvision Authentication """ server: str port: int = 443 token: Optional[str] = None timeout: int = 1200 validate_cert: bool = False class Filer: # pylint: disable=too-few-public-methods """ Filer Helper for file management """ def __init__(self, path: str) -> None: self.file_exist = False self.filename = "" self.absolute_path = "" self.relative_path = path if os.path.exists(path): self.file_exist = True self.filename = os.path.basename(path) self.absolute_path = os.path.realpath(path) def __repr__(self) -> str: return self.absolute_path if self.file_exist else "" class CvFeatureManager: """ CvFeatureManager Object to interect with Cloudvision """ def __init__(self, authentication: CvpAuthenticationItem) -> None: """ __init__ Class Creator Parameters ---------- authentication : CvpAuthenticationItem Authentication information to use to connect to Cloudvision """ self._authentication = authentication # self._cv_instance = CvpClient() self._cv_instance = self._connect(authentication=authentication) self._cv_images = self.__get_images() # self._cv_bundles = self.__get_bundles() def _connect(self, authentication: CvpAuthenticationItem) -> CvpClient: """ _connect Connection management Parameters ---------- authentication : CvpAuthenticationItem Authentication information to use to connect to Cloudvision Returns ------- CvpClient cvprac session to cloudvision """ client = CvpClient() if authentication.token is not None: try: client.connect( nodes=[authentication.server], username="", password="", api_token=authentication.token, is_cvaas=True, port=authentication.port, cert=authentication.validate_cert, request_timeout=authentication.timeout, ) except CvpLoginError as error_data: logger.error( f"Cannot connect to Cloudvision server {authentication.server}" ) logger.debug(f"Error message: {error_data}") logger.info("connected to Cloudvision server") logger.debug(f"Connection info: {authentication}") return client def __get_images(self) -> List[Any]: """ __get_images Collect information about images on Cloudvision Returns ------- dict Fact returned by Cloudvision """ images = [] logger.debug(" -> Collecting images") images = self._cv_instance.api.get_images()["data"] return images if self.__check_api_result(images) else [] # def __get_bundles(self): # """ # __get_bundles [Not In use] Collect information about bundles on Cloudvision # Returns # ------- # dict # Fact returned by Cloudvision # """ # bundles = [] # logger.debug(' -> Collecting images bundles') # bundles = self._cv_instance.api.get_image_bundles()['data'] # # bundles = self._cv_instance.post(url='/cvpservice/image/getImageBundles.do?queryparam=&startIndex=0&endIndex=0')['data'] # return bundles if self.__check_api_result(bundles) else None def __check_api_result(self, arg0: Any) -> bool: """ __check_api_result Check API calls return content Parameters ---------- arg0 : any Element to test Returns ------- bool True if data are correct False in other cases """ logger.debug(arg0) return len(arg0) > 0 def _does_image_exist(self, image_name: str) -> bool: """ _does_image_exist Check if an image is referenced in Cloudvision facts Parameters ---------- image_name : str Name of the image to search for Returns ------- bool True if present """ return ( any(image_name == image["name"] for image in self._cv_images) if isinstance(self._cv_images, list) else False ) def _does_bundle_exist(self, bundle_name: str) -> bool: # pylint: disable=unused-argument """ _does_bundle_exist Check if an image is referenced in Cloudvision facts Returns ------- bool True if present """ # return any(bundle_name == bundle['name'] for bundle in self._cv_bundles) return False def upload_image(self, image_path: str) -> bool: """ upload_image Upload an image to Cloudvision server Parameters ---------- image_path : str Path to the local file to upload Returns ------- bool True if succeeds """ image_item = Filer(path=image_path) if image_item.file_exist is False: logger.error(f"File not found: {image_item.relative_path}") return False logger.info(f"File path for image: {image_item}") if self._does_image_exist(image_name=image_item.filename): logger.error( "Image found in Cloudvision , Please delete it before running this script" ) return False try: upload_result = self._cv_instance.api.add_image( filepath=image_item.absolute_path ) except Exception as e: # pylint: disable=broad-exception-caught logger.error("An error occurred during upload, check CV connection") logger.error(f"Exception message is: {e}") return False logger.debug(f"Upload Result is : {upload_result}") return True def build_image_list(self, image_list: List[str]) -> List[Any]: """ Builds a list of the image data structures, for a given list of image names. Parameters ---------- image_list : list List of software image names Returns ------- List: Returns a list of images, with complete data or None in the event of failure """ internal_image_list = [] image_data = None success = True for entry in image_list: for image in self._cv_images: if image["imageFileName"] == entry: image_data = image if image_data is not None: internal_image_list.append(image_data) image_data = None else: success = False return internal_image_list if success else [] def create_bundle(self, name: str, images_name: List[str]) -> bool: """ create_bundle Create a bundle with a list of images. Parameters ---------- name : str Name of the bundle images_name : List[str] List of images available on Cloudvision Returns ------- bool True if succeeds """ logger.debug( f"Init creation of an image bundle {name} with following images {images_name}" ) all_images_present: List[bool] = [] self._cv_images = self.__get_images() all_images_present.extend( self._does_image_exist(image_name=image_name) for image_name in images_name ) # Bundle Create if self._does_bundle_exist(bundle_name=name) is False: logger.debug( f"Creating image bundle {name} with following images {images_name}" ) images_data = self.build_image_list(image_list=images_name) if images_data is not None: logger.debug("Images information: {images_data}") try: data = self._cv_instance.api.save_image_bundle( name=name, images=images_data ) except Exception as e: # pylint: disable=broad-exception-caught logger.critical(f"{e}") else: logger.debug(data) return True logger.critical("No data found for images") return False