"""VSP module.""" # Copyright 2022 Orange, Deutsche Telekom AG # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import json from typing import Any, Optional from typing import BinaryIO from typing import Callable from typing import Dict from onapsdk.exceptions import APIError, ParameterError from onapsdk.sdc.sdc_element import SdcElement from onapsdk.sdc.vendor import Vendor import onapsdk.constants as const from onapsdk.utils.headers_creator import headers_sdc_creator # Hard to do fewer attributes and still mapping SDC VSP object. class Vsp(SdcElement): # pylint: disable=too-many-instance-attributes """ ONAP VSP Object used for SDC operations. Attributes: name (str): the name of the vsp. Defaults to "ONAP-test-VSP". identifier (str): the unique ID of the VSP from SDC. status (str): the status of the VSP from SDC. version (str): the version ID of the VSP from SDC. csar_uuid (str): the CSAR ID of the VSP from SDC. vendor (Vendor): The VSP Vendor """ VSP_PATH = "vendor-software-products" headers = headers_sdc_creator(SdcElement.headers) def __init__(self, name: str = None, package: BinaryIO = None, vendor: Vendor = None): """ Initialize vsp object. Args: name (optional): the name of the vsp """ super().__init__() self._csar_uuid: str = None self._vendor: Vendor = vendor or None self.name: str = name or "ONAP-test-VSP" self.package = package or None @property def status(self): """Return and load the status.""" self.load_status() return self._status def onboard(self) -> None: """Onboard the VSP in SDC.""" status: Optional[str] = self.status if not status: if not self._vendor: raise ParameterError("No Vendor provided.") self.create() self.onboard() elif status == const.DRAFT: if not self.package: raise ParameterError("No file/package provided.") self.upload_package(self.package) self.onboard() elif status == const.UPLOADED: self.validate() self.onboard() elif status == const.VALIDATED: self.commit() self.onboard() elif status == const.COMMITED: self.submit() self.onboard() elif status == const.CERTIFIED: self.create_csar() def create(self) -> None: """Create the Vsp in SDC if not already existing.""" if self.vendor: self._create("vsp_create.json.j2", name=self.name, vendor=self.vendor) def upload_package(self, package_to_upload: BinaryIO) -> None: """ Upload given zip file into SDC as artifacts for this Vsp. Args: package_to_upload (file): the zip file to upload """ self._action("upload package", const.DRAFT, self._upload_action, package_to_upload=package_to_upload) def update_package(self, package_to_upload: BinaryIO) -> None: """ Upload given zip file into SDC as artifacts for this Vsp. Args: package_to_upload (file): the zip file to upload """ self._action("update package", const.COMMITED, self._upload_action, package_to_upload=package_to_upload) def validate(self) -> None: """Validate the artifacts uploaded.""" self._action("validate", const.UPLOADED, self._validate_action) def commit(self) -> None: """Commit the SDC Vsp.""" self._action("commit", const.VALIDATED, self._generic_action, action=const.COMMIT) def submit(self) -> None: """Submit the SDC Vsp in order to enable it.""" self._action("certify/sumbit", const.COMMITED, self._generic_action, action=const.SUBMIT) def archive(self) -> None: """Archive VSP.""" self._action("archive", const.CERTIFIED, self._generic_action, action=const.ARCHIVE) def create_csar(self) -> None: """Create the CSAR package in the SDC Vsp.""" self._action("create CSAR package", const.CERTIFIED, self._create_csar_action) @property def vendor(self) -> Vendor: """Return and lazy load the vendor.""" if not self._vendor and self.created(): details = self._get_vsp_details() if details: self._vendor = Vendor(name=details['vendorName']) return self._vendor @vendor.setter def vendor(self, vendor: Vendor) -> None: """Set value for Vendor.""" self._vendor = vendor @property def csar_uuid(self) -> str: """Return and lazy load the CSAR UUID.""" if self.created() and not self._csar_uuid: self.create_csar() return self._csar_uuid @csar_uuid.setter def csar_uuid(self, csar_uuid: str) -> None: """Set value for csar uuid.""" self._csar_uuid = csar_uuid def _get_item_version_details(self) -> Dict[Any, Any]: """Get vsp item details.""" if self.created() and self.version: url = "{}/items/{}/versions/{}".format(self._base_url(), self.identifier, self.version) return self.send_message_json('GET', 'get item version', url) return {} def _upload_action(self, package_to_upload: BinaryIO): """Do upload for real.""" url = "{}/{}/{}/orchestration-template-candidate".format( self._base_url(), Vsp._sdc_path(), self._version_path()) headers = self.headers.copy() headers.pop("Content-Type") headers["Accept-Encoding"] = "gzip, deflate" data = {'upload': package_to_upload} upload_result = self.send_message('POST', 'upload ZIP for Vsp', url, headers=headers, files=data) if upload_result: # TODO https://jira.onap.org/browse/SDC-3505 pylint: disable=W0511 response_json = json.loads(upload_result.text) if response_json["status"] != "Success": self._logger.error( "an error occured during file upload for Vsp %s", self.name) raise APIError(response_json) # End TODO https://jira.onap.org/browse/SDC-3505 self._logger.info("Files for Vsp %s have been uploaded", self.name) else: self._logger.error( "an error occured during file upload for Vsp %s", self.name) def _validate_action(self): """Do validate for real.""" url = "{}/{}/{}/orchestration-template-candidate/process".format( self._base_url(), Vsp._sdc_path(), self._version_path()) validate_result = self.send_message_json('PUT', 'Validate artifacts for Vsp', url) if validate_result and validate_result['status'] == 'Success': self._logger.info("Artifacts for Vsp %s have been validated", self.name) else: self._logger.error( "an error occured during artifacts validation for Vsp %s", self.name) def _generic_action(self, action=None): """Do a generic action for real.""" if action: if action == const.ARCHIVE: self._action_to_sdc(action, action_type=const.ARCHIVE) else: self._action_to_sdc(action, action_type="lifecycleState") def _create_csar_action(self): """Create CSAR package for real.""" result = self._action_to_sdc(const.CREATE_PACKAGE, action_type="lifecycleState") if result: self._logger.info("result: %s", result.text) data = result.json() self.csar_uuid = data['packageId'] def _action(self, action_name: str, right_status: str, action_function: Callable[['Vsp'], None], **kwargs) -> None: """ Generate an action on the instance in order to send it to SDC. Args: action_name (str): The name of the action (for the logs) right_status (str): The status that the object must be action_function (function): the function to perform if OK """ self._logger.info("attempting to %s for %s in SDC", action_name, self.name) if self.status == right_status: action_function(**kwargs) else: self._logger.warning( "vsp %s in SDC is not created or not in %s status", self.name, right_status) # VSP: DRAFT --> UPLOADED --> VALIDATED --> COMMITED --> CERTIFIED def load_status(self) -> None: """ Load Vsp status from SDC. rules are following: * DRAFT: status == DRAFT and networkPackageName not present * UPLOADED: status == DRAFT and networkPackageName present and validationData not present * VALIDATED: status == DRAFT and networkPackageName present and validationData present and state.dirty = true * COMMITED: status == DRAFT and networkPackageName present and validationData present and state.dirty = false * CERTIFIED: status == CERTIFIED status is found in sdc items state is found in sdc version from items networkPackageName and validationData is found in SDC vsp show """ item_details = self._get_item_details() if (item_details and item_details['status'] == const.CERTIFIED): self._status = const.CERTIFIED else: self._check_status_not_certified() def _check_status_not_certified(self) -> None: """Check a status when it's not certified.""" vsp_version_details = self._get_item_version_details() vsp_details = self._get_vsp_details() if (vsp_version_details and 'state' in vsp_version_details and not vsp_version_details['state']['dirty'] and vsp_details and 'validationData' in vsp_details): self._status = const.COMMITED else: self._check_status_not_commited() def _check_status_not_commited(self) -> None: """Check a status when it's not certified or commited.""" vsp_details = self._get_vsp_details() if (vsp_details and 'validationData' in vsp_details): self._status = const.VALIDATED elif (vsp_details and 'validationData' not in vsp_details and 'networkPackageName' in vsp_details): self._status = const.UPLOADED elif vsp_details: self._status = const.DRAFT def _get_vsp_details(self) -> Dict[Any, Any]: """Get vsp details.""" if self.created() and self.version: url = "{}/vendor-software-products/{}/versions/{}".format( self._base_url(), self.identifier, self.version) return self.send_message_json('GET', 'get vsp version', url) return {} @classmethod def _get_all_url(cls) -> str: """ Get URL for all elements in SDC. Returns: str: the url """ return f"{cls._base_url()}/items?itemType=vsp" @classmethod def import_from_sdc(cls, values: Dict[str, Any]) -> 'Vsp': """ Import Vsp from SDC. Args: values (Dict[str, Any]): dict to parse returned from SDC. Returns: a Vsp instance with right values """ cls._logger.debug("importing VSP %s from SDC", values['name']) vsp = Vsp(values['name']) vsp.identifier = values['id'] vsp.vendor = Vendor(name=values['properties']['vendorName']) return vsp def _really_submit(self) -> None: """Really submit the SDC Vf in order to enable it. Raises: NotImplementedError """ raise NotImplementedError("VSP don't need _really_submit") @classmethod def _sdc_path(cls) -> None: """Give back the end of SDC path.""" return cls.VSP_PATH def create_new_version(self) -> None: """Create new version of VSP. Create a new major version of VSP so it would be possible to update a package or do some changes in VSP. """ self._logger.debug("Create new version of %s VSP", self.name) self.send_message_json("POST", "Create new VSP version", (f"{self._base_url()}/items/{self.identifier}/" f"versions/{self.version}"), data=json.dumps({ "creationMethod": "major", "description": "New VSP version" })) self.load()