summaryrefslogtreecommitdiffstats
path: root/eos_downloader/cvp.py
diff options
context:
space:
mode:
authorDaniel Baumann <daniel.baumann@progress-linux.org>2023-10-10 09:54:13 +0000
committerDaniel Baumann <daniel.baumann@progress-linux.org>2023-10-10 09:54:13 +0000
commit26cac87f1cb90abc2fac2144a39ad88b3b70031d (patch)
treef97c513314a569fb51f1c8580b012d418c05fede /eos_downloader/cvp.py
parentAdding upstream version 0.8.2. (diff)
downloadeos-downloader-26cac87f1cb90abc2fac2144a39ad88b3b70031d.tar.xz
eos-downloader-26cac87f1cb90abc2fac2144a39ad88b3b70031d.zip
Adding upstream version 0.9.0.upstream/0.9.0
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'eos_downloader/cvp.py')
-rw-r--r--eos_downloader/cvp.py85
1 files changed, 52 insertions, 33 deletions
diff --git a/eos_downloader/cvp.py b/eos_downloader/cvp.py
index 6f14eb0..678daa8 100644
--- a/eos_downloader/cvp.py
+++ b/eos_downloader/cvp.py
@@ -6,11 +6,12 @@ CVP Uploader content
"""
import os
-from typing import List, Optional, Any
from dataclasses import dataclass
-from loguru import logger
+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
@@ -20,8 +21,9 @@ from cvprac.cvp_client_errors import CvpLoginError
@dataclass
class CvpAuthenticationItem:
"""
- Data structure to represent Cloudvision Authentication
+ Data structure to represent Cloudvision Authentication
"""
+
server: str
port: int = 443
token: Optional[str] = None
@@ -29,15 +31,16 @@ class CvpAuthenticationItem:
validate_cert: bool = False
-class Filer():
+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.filename = ""
+ self.absolute_path = ""
self.relative_path = path
if os.path.exists(path):
self.file_exist = True
@@ -45,13 +48,14 @@ class Filer():
self.absolute_path = os.path.realpath(path)
def __repr__(self) -> str:
- return self.absolute_path if self.file_exist else ''
+ return self.absolute_path if self.file_exist else ""
-class CvFeatureManager():
+class CvFeatureManager:
"""
CvFeatureManager Object to interect with Cloudvision
"""
+
def __init__(self, authentication: CvpAuthenticationItem) -> None:
"""
__init__ Class Creator
@@ -86,19 +90,21 @@ class CvFeatureManager():
try:
client.connect(
nodes=[authentication.server],
- username='',
- password='',
+ username="",
+ password="",
api_token=authentication.token,
is_cvaas=True,
port=authentication.port,
cert=authentication.validate_cert,
- request_timeout=authentication.timeout
+ 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}')
+ 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]:
@@ -111,8 +117,8 @@ class CvFeatureManager():
Fact returned by Cloudvision
"""
images = []
- logger.debug(' -> Collecting images')
- images = self._cv_instance.api.get_images()['data']
+ 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):
@@ -161,7 +167,11 @@ class CvFeatureManager():
bool
True if present
"""
- return any(image_name == image['name'] for image in self._cv_images) if isinstance(self._cv_images, list) else False
+ 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
@@ -192,19 +202,23 @@ class CvFeatureManager():
"""
image_item = Filer(path=image_path)
if image_item.file_exist is False:
- logger.error(f'File not found: {image_item.relative_path}')
+ logger.error(f"File not found: {image_item.relative_path}")
return False
- logger.info(f'File path for image: {image_item}')
+ 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")
+ 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)
+ 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}')
+ 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}')
+ logger.debug(f"Upload Result is : {upload_result}")
return True
def build_image_list(self, image_list: List[str]) -> List[Any]:
@@ -252,25 +266,30 @@ class CvFeatureManager():
bool
True if succeeds
"""
- logger.debug(f'Init creation of an image bundle {name} with following images {images_name}')
+ 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
+ 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}')
+ 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}')
+ logger.debug("Images information: {images_data}")
try:
- data = self._cv_instance.api.save_image_bundle(name=name, images=images_data)
+ 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}')
+ logger.critical(f"{e}")
else:
logger.debug(data)
return True
- logger.critical('No data found for images')
+ logger.critical("No data found for images")
return False