diff options
Diffstat (limited to 'src/onapsdk/sdc/vsp.py')
| -rw-r--r-- | src/onapsdk/sdc/vsp.py | 370 |
1 files changed, 370 insertions, 0 deletions
diff --git a/src/onapsdk/sdc/vsp.py b/src/onapsdk/sdc/vsp.py new file mode 100644 index 0000000..a6c4c24 --- /dev/null +++ b/src/onapsdk/sdc/vsp.py @@ -0,0 +1,370 @@ +"""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 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: + 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 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['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() |