diff options
Diffstat (limited to 'src/onapsdk/sdc')
32 files changed, 4435 insertions, 0 deletions
diff --git a/src/onapsdk/sdc/__init__.py b/src/onapsdk/sdc/__init__.py new file mode 100644 index 0000000..15280d9 --- /dev/null +++ b/src/onapsdk/sdc/__init__.py @@ -0,0 +1,486 @@ +"""SDC Element 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. +from typing import Any, Dict, List, Optional, Union +from operator import attrgetter +from abc import ABC, abstractmethod + +from requests import Response + +from onapsdk.configuration import settings +from onapsdk.exceptions import APIError, RequestError +from onapsdk.onap_service import OnapService +import onapsdk.constants as const +from onapsdk.utils.jinja import jinja_env +from onapsdk.utils.gui import GuiItem, GuiList + +class SDC(OnapService, ABC): + """Mother Class of all SDC elements.""" + + server: str = "SDC" + base_front_url = settings.SDC_FE_URL + base_back_url = settings.SDC_BE_URL + + def __init__(self, name: str = None) -> None: + """Initialize SDC.""" + super().__init__() + self.name: str = name + + def __eq__(self, other: Any) -> bool: + """ + Check equality for SDC and children. + + Args: + other: another object + + Returns: + bool: True if same object, False if not + + """ + if isinstance(other, type(self)): + return self.name == other.name + return False + + @classmethod + @abstractmethod + def _get_all_url(cls) -> str: + """ + Get URL for all elements in SDC. + + Raises: + NotImplementedError: this is an abstract method. + + """ + + @classmethod + @abstractmethod + def _get_objects_list(cls, + result: List[Dict[str, Any]]) -> List['SdcResource']: + """ + Import objects created in SDC. + + Args: + result (Dict[str, Any]): the result returned by SDC in a Dict + + Raises: + NotImplementedError: this is an abstract method. + + """ + + @classmethod + @abstractmethod + def _base_url(cls) -> str: + """ + Give back the base url of Sdc. + + Raises: + NotImplementedError: this is an abstract method. + + """ + + @classmethod + @abstractmethod + def _base_create_url(cls) -> str: + """ + Give back the base url of Sdc. + + Raises: + NotImplementedError: this is an abstract method. + + """ + + @abstractmethod + def _copy_object(self, obj: 'SDC') -> None: + """ + Copy relevant properties from object. + + Args: + obj (Sdc): the object to "copy" + + Raises: + NotImplementedError: this is an abstract method. + + """ + + @classmethod + @abstractmethod + def import_from_sdc(cls, values: Dict[str, Any]) -> 'SDC': + """ + Import Sdc object from SDC. + + Args: + values (Dict[str, Any]): dict to parse returned from SDC. + + Raises: + NotImplementedError: this is an abstract method. + + """ + + @staticmethod + def _get_mapped_version(item: "SDC") -> Optional[Union[float, str]]: + """Map Sdc objects version to float. + + Mostly we need to get the newest version of the requested objects. To do + so we use the version property of them. In most cases it's string + formatted float value, but in some cases (like VSP objects) it isn't. + That method checks if given object has "version" attribute and if it's not + a None it tries to map it's value to float. If it's not possible it + returns the alrady existing value. + + Args: + item (SDC): SDC item to map version to float + + Returns: + Optional[Union[float, str]]: Float format version if possible, + string otherwise. If object doesn't have "version" + attribut returns None. + + """ + if hasattr(item, "version") and item.version is not None: + try: + return float(item.version) + except ValueError: + return item.version + else: + return None + + @classmethod + def get_all(cls, **kwargs) -> List['SDC']: + """ + Get the objects list created in SDC. + + Returns: + the list of the objects + + """ + cls._logger.info("retrieving all objects of type %s from SDC", + cls.__name__) + url = cls._get_all_url() + objects = [] + + try: + result = \ + cls.send_message_json('GET', "get {}s".format(cls.__name__), + url, **kwargs) + + for obj_info in cls._get_objects_list(result): + objects.append(cls.import_from_sdc(obj_info)) + + except APIError as exc: + cls._logger.debug("Couldn't get %s: %s", cls.__name__, exc) + except KeyError as exc: + cls._logger.debug("Invalid result dictionary: %s", exc) + + cls._logger.debug("number of %s returned: %s", cls.__name__, + len(objects)) + return objects + + def exists(self) -> bool: + """ + Check if object already exists in SDC and update infos. + + Returns: + True if exists, False either + + """ + self._logger.debug("check if %s %s exists in SDC", + type(self).__name__, self.name) + objects = self.get_all() + + self._logger.debug("filtering objects of all versions to be %s", + self.name) + relevant_objects = list(filter(lambda obj: obj == self, objects)) + + if not relevant_objects: + + self._logger.info("%s %s doesn't exist in SDC", + type(self).__name__, self.name) + return False + + if hasattr(self, 'version_filter') and self.version_filter is not None: # pylint: disable=no-member + + self._logger.debug("filtering %s objects by version %s", + self.name, self.version_filter) # pylint: disable=no-member + + all_versioned = filter( + lambda obj: obj.version == self.version_filter, relevant_objects) # pylint: disable=no-member + + try: + versioned_object = next(all_versioned) + except StopIteration: + self._logger.info("Version %s of %s %s, doesn't exist in SDC", + self.version_filter, type(self).__name__, # pylint: disable=no-member + self.name) + return False + + else: + versioned_object = max(relevant_objects, key=self._get_mapped_version) + + self._logger.info("%s found, updating information", type(self).__name__) + self._copy_object(versioned_object) + return True + + @classmethod + def get_guis(cls) -> GuiItem: + """Retrieve the status of the SDC GUIs. + + Only one GUI is referenced for SDC + the SDC Front End + + Return the list of GUIs + """ + gui_url = settings.SDC_GUI_SERVICE + sdc_gui_response = cls.send_message( + "GET", "Get SDC GUI Status", gui_url) + guilist = GuiList([]) + guilist.add(GuiItem( + gui_url, + sdc_gui_response.status_code)) + return guilist + +class SdcOnboardable(SDC, ABC): + """Base class for onboardable SDC resources (Vendors, Services, ...).""" + + ACTION_TEMPLATE: str + ACTION_METHOD: str + + def __init__(self, name: str = None) -> None: + """Initialize the object.""" + super().__init__(name) + self._identifier: str = None + self._status: str = None + self._version: str = None + + @property + def identifier(self) -> str: + """Return and lazy load the identifier.""" + if not self._identifier: + self.load() + return self._identifier + + @property + def status(self) -> str: + """Return and lazy load the status.""" + if self.created() and not self._status: + self.load() + return self._status + + @property + def version(self) -> str: + """Return and lazy load the version.""" + if self.created() and not self._version: + self.load() + return self._version + + @identifier.setter + def identifier(self, value: str) -> None: + """Set value for identifier.""" + self._identifier = value + + @status.setter + def status(self, status: str) -> None: + """Return and lazy load the status.""" + self._status = status + + @version.setter + def version(self, version: str) -> None: + """Return and lazy load the status.""" + self._version = version + + def created(self) -> bool: + """Determine if SDC is created.""" + if self.name and not self._identifier: + return self.exists() + return bool(self._identifier) + + def submit(self) -> None: + """Submit the SDC object in order to enable it.""" + self._logger.info("attempting to certify/sumbit %s %s in SDC", + type(self).__name__, self.name) + if self.status != const.CERTIFIED and self.created(): + self._really_submit() + elif self.status == const.CERTIFIED: + self._logger.warning("%s %s in SDC is already submitted/certified", + type(self).__name__, self.name) + elif not self.created(): + self._logger.warning("%s %s in SDC is not created", + type(self).__name__, self.name) + + def _create(self, template_name: str, **kwargs) -> None: + """Create the object in SDC if not already existing.""" + self._logger.info("attempting to create %s %s in SDC", + type(self).__name__, self.name) + if not self.exists(): + url = "{}/{}".format(self._base_create_url(), self._sdc_path()) + template = jinja_env().get_template(template_name) + data = template.render(**kwargs) + try: + create_result = self.send_message_json('POST', + "create {}".format( + type(self).__name__), + url, + data=data) + except RequestError as exc: + self._logger.error( + "an error occured during creation of %s %s in SDC", + type(self).__name__, self.name) + raise exc + else: + self._logger.info("%s %s is created in SDC", + type(self).__name__, self.name) + self._status = const.DRAFT + self.identifier = self._get_identifier_from_sdc(create_result) + self._version = self._get_version_from_sdc(create_result) + self.update_informations_from_sdc_creation(create_result) + + else: + self._logger.warning("%s %s is already created in SDC", + type(self).__name__, self.name) + + def _action_to_sdc(self, action: str, action_type: str = None, + **kwargs) -> Response: + """ + Really do an action in the SDC. + + Args: + action (str): the action to perform + action_type (str, optional): the type of action + headers (Dict[str, str], optional): headers to use if any + + Returns: + Response: the response + + """ + subpath = self._generate_action_subpath(action) + url = self._action_url(self._base_create_url(), + subpath, + self._version_path(), + action_type=action_type) + template = jinja_env().get_template(self.ACTION_TEMPLATE) + data = template.render(action=action, const=const) + + return self.send_message(self.ACTION_METHOD, + "{} {}".format(action, + type(self).__name__), + url, + data=data, + **kwargs) + + @abstractmethod + def update_informations_from_sdc(self, details: Dict[str, Any]) -> None: + """ + + Update instance with details from SDC. + + Args: + details ([type]): [description] + + """ + @abstractmethod + def update_informations_from_sdc_creation(self, + details: Dict[str, Any]) -> None: + """ + + Update instance with details from SDC after creation. + + Args: + details ([type]): the details from SDC + + """ + + @abstractmethod + def load(self) -> None: + """ + Load Object information from SDC. + + Raises: + NotImplementedError: this is an abstract method. + + """ + + @abstractmethod + def _get_version_from_sdc(self, sdc_infos: Dict[str, Any]) -> str: + """ + Get version from SDC results. + + Args: + sdc_infos (Dict[str, Any]): the result dict from SDC + + Raises: + NotImplementedError: this is an abstract method. + + """ + @abstractmethod + def _get_identifier_from_sdc(self, sdc_infos: Dict[str, Any]) -> str: + """ + Get identifier from SDC results. + + Args: + sdc_infos (Dict[str, Any]): the result dict from SDC + + Raises: + NotImplementedError: this is an abstract method. + + """ + @abstractmethod + def _generate_action_subpath(self, action: str) -> str: + """ + + Generate subpath part of SDC action url. + + Args: + action (str): the action that will be done + + Raises: + NotImplementedError: this is an abstract method. + + """ + @abstractmethod + def _version_path(self) -> str: + """ + Give the end of the path for a version. + + Raises: + NotImplementedError: this is an abstract method. + + """ + @abstractmethod + def _really_submit(self) -> None: + """Really submit the SDC Vf in order to enable it.""" + @staticmethod + @abstractmethod + def _action_url(base: str, + subpath: str, + version_path: str, + action_type: str = None) -> str: + """ + Generate action URL for SDC. + + Raises: + NotImplementedError: this is an abstract method. + + """ + @classmethod + @abstractmethod + def _sdc_path(cls) -> None: + """Give back the end of SDC path.""" + + @abstractmethod + def onboard(self) -> None: + """Onboard resource. + + Onboarding is a full stack of actions which needs to be done to + make SDC resource ready to use. It depends on the type of object + but most of them needs to be created and submitted. + """ diff --git a/src/onapsdk/sdc/category_management.py b/src/onapsdk/sdc/category_management.py new file mode 100644 index 0000000..c2d63b4 --- /dev/null +++ b/src/onapsdk/sdc/category_management.py @@ -0,0 +1,285 @@ +"""SDC category management 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 abc import ABC, abstractmethod +from typing import Any, Dict, List + +from onapsdk.configuration import settings +from onapsdk.exceptions import ResourceNotFound +from onapsdk.sdc import SDC +from onapsdk.utils.headers_creator import headers_sdc_generic + + +class BaseCategory(SDC, ABC): # pylint: disable=too-many-instance-attributes + """Base SDC category class. + + It's SDC admin resource, has no common properties with + SDC resourcer or elements, so SDC class can't be it's + base class. + + """ + + SDC_ADMIN_USER = "demo" + + def __init__(self, name: str) -> None: + """Service category initialization. + + Args: + name (str): Service category name. + + """ + super().__init__(name) + self.normalized_name: str = None + self.unique_id: str = None + self.icons: List[str] = None + self.subcategories: List[Dict[str, str]] = None + self.version: str = None + self.owner_id: str = None + self.empty: bool = None + self.type: str = None + + @classmethod + def _get_all_url(cls) -> str: + """Get URL for all categories in SDC.""" + return f"{cls.base_front_url}/sdc1/feProxy/rest/v1/setup/ui" + + @classmethod + def _base_url(cls) -> str: + """Give back the base url of Sdc.""" + return f"{settings.SDC_FE_URL}/sdc1/feProxy/rest/v1/category" + + @classmethod + def headers(cls) -> Dict[str, str]: + """Headers used for category management. + + It uses SDC admin user. + + Returns: + Dict[str, str]: Headers + + """ + return headers_sdc_generic(super().headers, user=cls.SDC_ADMIN_USER) + + @classmethod + def get_all(cls, **kwargs) -> List['SDC']: + """ + Get the categories list created in SDC. + + Returns: + the list of the categories + + """ + return super().get_all(headers=cls.headers()) + + @classmethod + def import_from_sdc(cls, values: Dict[str, Any]) -> 'BaseCategory': + """ + Import category object from SDC. + + Args: + values (Dict[str, Any]): dict to parse returned from SDC. + + """ + category_obj = cls(name=values["name"]) + category_obj.normalized_name = values["normalizedName"] + category_obj.unique_id = values["uniqueId"] + category_obj.icons = values["icons"] + category_obj.subcategories = values["subcategories"] + category_obj.version = values["version"] + category_obj.owner_id = values["ownerId"] + category_obj.empty = values["empty"] + return category_obj + + @classmethod + @abstractmethod + def category_name(cls) -> str: + """Class category name. + + Used for logs. + + Returns: + str: Category name + + """ + + @classmethod + def get(cls, name: str) -> "BaseCategory": + """Get category with given name. + + Raises: + ResourceNotFound: Category with given name does not exist + + Returns: + BaseCategory: BaseCategory instance + + """ + category_obj: "BaseCategory" = cls(name) + if category_obj.exists(): + return category_obj + msg = f"{cls.category_name()} with \"{name}\" name does not exist." + raise ResourceNotFound(msg) + + @classmethod + def create(cls, name: str) -> "BaseCategory": + """Create category instance. + + Checks if category with given name exists and if it already + exists just returns category with given name. + + Returns: + BaseCategory: Created category instance + + """ + category_obj: "BaseCategory" = cls(name) + if category_obj.exists(): + return category_obj + cls.send_message_json("POST", + f"Create {name} {cls.category_name()}", + cls._base_create_url(), + data=json.dumps({"name": name}), + headers=cls.headers()) + category_obj.exists() + return category_obj + + def _copy_object(self, obj: 'BaseCategory') -> None: + """ + Copy relevant properties from object. + + Args: + obj (BaseCategory): the object to "copy" + + """ + self.name = obj.name + self.normalized_name = obj.normalized_name + self.unique_id = obj.unique_id + self.icons = obj.icons + self.subcategories = obj.subcategories + self.version = obj.version + self.owner_id = obj.owner_id + self.empty = obj.empty + + +class ResourceCategory(BaseCategory): + """Resource category class.""" + + @classmethod + def _get_objects_list(cls, + result: Dict[str, Any]) -> List[Dict[str, Any]]: + """ + Get list of resource categories. + + Args: + result (List[Dict[str, Any]]): the result returned by SDC + in a list of dicts + + Raises: + KeyError: Invalid result dictionary + + """ + return result["categories"]["resourceCategories"] + + @classmethod + def _base_create_url(cls) -> str: + """Url to create resource category. + + Returns: + str: Creation url + + """ + return f"{cls._base_url()}/resources" + + @classmethod + def category_name(cls) -> str: + """Resource category name. + + Used for logging. + + Returns: + str: Resource category name + + """ + return "Resource Category" + + @classmethod + def get(cls, name: str, subcategory: str = None) -> "ResourceCategory": # pylint: disable=arguments-differ + """Get resource category with given name. + + It returns resource category with all subcategories by default. You can + get resource category with only one subcategory if you provide it's + name as `subcategory` parameter. + + Args: + name (str): Resource category name. + subcategory (str, optional): Name of subcategory. Defaults to None. + + Raises: + ResourceNotFound: Subcategory with given name does not exist + + Returns: + BaseCategory: BaseCategory instance + + """ + category_obj: "ResourceCategory" = super().get(name=name) + if not subcategory: + return category_obj + filtered_subcategories: Dict[str, str] = list(filter(lambda x: x["name"] == subcategory, + category_obj.subcategories)) + if not filtered_subcategories: + raise ResourceNotFound(f"Subcategory {subcategory} does not exist.") + category_obj.subcategories = filtered_subcategories + return category_obj + + +class ServiceCategory(BaseCategory): + """Service category class.""" + + @classmethod + def _get_objects_list(cls, + result: Dict[str, Any]) -> List[Dict[str, Any]]: + """ + Get list of service categories. + + Args: + result (List[Dict[str, Any]]): the result returned by SDC + in a list of dicts + + Raises: + KeyError: Invalid result dictionary + + """ + return result["categories"]["serviceCategories"] + + @classmethod + def _base_create_url(cls) -> str: + """Url to create service category. + + Returns: + str: Creation url + + """ + return f"{cls._base_url()}/services" + + @classmethod + def category_name(cls) -> str: + """Service category name. + + Used for logging. + + Returns: + str: Service category name + + """ + return "Service Category" diff --git a/src/onapsdk/sdc/component.py b/src/onapsdk/sdc/component.py new file mode 100644 index 0000000..9b26bef --- /dev/null +++ b/src/onapsdk/sdc/component.py @@ -0,0 +1,162 @@ +"""SDC Component 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. +from dataclasses import dataclass +from typing import Any, Dict, Iterator, List, Optional +from onapsdk.exceptions import ParameterError + +from onapsdk.sdc.properties import ComponentProperty +from onapsdk.utils.jinja import jinja_env + + +@dataclass +class Component: # pylint: disable=too-many-instance-attributes + """Component dataclass.""" + + created_from_csar: bool + actual_component_uid: str + unique_id: str + normalized_name: str + name: str + origin_type: str + customization_uuid: str + component_uid: str + component_version: str + tosca_component_name: str + component_name: str + group_instances: Optional[List[Dict[str, Any]]] + sdc_resource: "SdcResource" + parent_sdc_resource: "SdcResource" + + @classmethod + def create_from_api_response(cls, + api_response: Dict[str, Any], + sdc_resource: "SdcResource", + parent_sdc_resource: "SdcResource") -> "Component": + """Create component from api response. + + Args: + api_response (Dict[str, Any]): component API response + sdc_resource (SdcResource): component's SDC resource + parent_sdc_resource (SdcResource): component's parent SDC resource + + Returns: + Component: Component created using api_response and SDC resource + + """ + return cls(created_from_csar=api_response["createdFromCsar"], + actual_component_uid=api_response["actualComponentUid"], + unique_id=api_response["uniqueId"], + normalized_name=api_response["normalizedName"], + name=api_response["name"], + origin_type=api_response["originType"], + customization_uuid=api_response["customizationUUID"], + component_uid=api_response["componentUid"], + component_version=api_response["componentVersion"], + tosca_component_name=api_response["toscaComponentName"], + component_name=api_response["componentName"], + group_instances=api_response["groupInstances"], + sdc_resource=sdc_resource, + parent_sdc_resource=parent_sdc_resource) + + @property + def properties_url(self) -> str: + """Url to get component's properties. + + Returns: + str: Compoent's properties url + + """ + return self.parent_sdc_resource.get_component_properties_url(self) + + @property + def properties_value_url(self) -> str: + """Url to set component property value. + + Returns: + str: Url to set component property value + + """ + return self.parent_sdc_resource.get_component_properties_value_set_url(self) + + @property + def properties(self) -> Iterator["ComponentProperty"]: + """Component properties. + + In SDC it's named as properties, but we uses "inputs" endpoint to fetch them. + Structure is also input's like, but it's a property. + + Yields: + ComponentProperty: Component property object + + """ + for component_property in self.sdc_resource.send_message_json(\ + "GET", + f"Get {self.name} component properties", + self.properties_url): + yield ComponentProperty(unique_id=component_property["uniqueId"], + name=component_property["name"], + property_type=component_property["type"], + _value=component_property.get("value"), + component=self) + + def get_property(self, property_name: str) -> "ComponentProperty": + """Get component property by it's name. + + Args: + property_name (str): property name + + Raises: + ParameterError: Component has no property with given name + + Returns: + ComponentProperty: Component's property object + + """ + for property_obj in self.properties: + if property_obj.name == property_name: + return property_obj + msg = f"Component has no property with {property_name} name" + raise ParameterError(msg) + + def set_property_value(self, property_obj: "ComponentProperty", value: Any) -> None: + """Set property value. + + Set given value to component property + + Args: + property_obj (ComponentProperty): Component property object + value (Any): Property value to set + + """ + self.sdc_resource.send_message_json( + "POST", + f"Set {self.name} component property {property_obj.name} value", + self.properties_value_url, + data=jinja_env().get_template(\ + "sdc_resource_component_set_property_value.json.j2").\ + render( + component=self, + value=value, + property=property_obj + ) + ) + + def delete(self) -> None: + """Delete component.""" + self.sdc_resource.send_message_json( + "DELETE", + f"Delete {self.name} component", + f"{self.parent_sdc_resource.resource_inputs_url}/resourceInstance/{self.unique_id}" + ) diff --git a/src/onapsdk/sdc/pnf.py b/src/onapsdk/sdc/pnf.py new file mode 100644 index 0000000..3fe4657 --- /dev/null +++ b/src/onapsdk/sdc/pnf.py @@ -0,0 +1,74 @@ +"""Pnf 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. +from typing import Dict, List, Union +from onapsdk.exceptions import ParameterError + +from onapsdk.sdc.sdc_resource import SdcResource +from onapsdk.sdc.properties import NestedInput, Property +import onapsdk.constants as const +from onapsdk.sdc.vendor import Vendor +from onapsdk.sdc.vsp import Vsp + + +class Pnf(SdcResource): + """ + ONAP PNF Object used for SDC operations. + + Attributes: + name (str): the name of the pnf. Defaults to "ONAP-test-PNF". + identifier (str): the unique ID of the pnf from SDC. + status (str): the status of the pnf from SDC. + version (str): the version ID of the vendor from SDC. + uuid (str): the UUID of the PNF (which is different from identifier, + don't ask why...) + unique_identifier (str): Yet Another ID, just to puzzle us... + vendor (optional): the vendor of the PNF + vsp (optional): the vsp related to the PNF + + """ + + def __init__(self, name: str = None, version: str = None, vendor: Vendor = None, # pylint: disable=too-many-arguments + sdc_values: Dict[str, str] = None, vsp: Vsp = None, + properties: List[Property] = None, inputs: Union[Property, NestedInput] = None, + category: str = None, subcategory: str = None): + """ + Initialize pnf object. + + Args: + name (optional): the name of the pnf + version (str, optional): the version of a PNF object + + """ + super().__init__(sdc_values=sdc_values, version=version, properties=properties, + inputs=inputs, category=category, subcategory=subcategory) + self.name: str = name or "ONAP-test-PNF" + self.vendor: Vendor = vendor + self.vsp: Vsp = vsp + + def create(self) -> None: + """Create the PNF in SDC if not already existing.""" + if not self.vsp and not self.vendor: + raise ParameterError("Neither Vsp nor Vendor provided.") + self._create("pnf_create.json.j2", + name=self.name, + vsp=self.vsp, + vendor=self.vendor, + category=self.category) + + def _really_submit(self) -> None: + """Really submit the SDC PNF in order to enable it.""" + result = self._action_to_sdc(const.CERTIFY, "lifecycleState") + if result: + self.load() diff --git a/src/onapsdk/sdc/properties.py b/src/onapsdk/sdc/properties.py new file mode 100644 index 0000000..f7e07a0 --- /dev/null +++ b/src/onapsdk/sdc/properties.py @@ -0,0 +1,202 @@ +"""Service properties 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. +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + +from onapsdk.exceptions import ParameterError + + +@dataclass +class Input: + """Property input dataclass.""" + + unique_id: str + input_type: str + name: str + sdc_resource: "SdcResource" + _default_value: Optional[Any] = field(repr=False, default=None) + + @property + def default_value(self) -> Any: + """Input default value. + + Returns: + Any: Input default value + + """ + return self._default_value + + @default_value.setter + def default_value(self, value: Any) -> None: + """Set input default value. + + Use related sdc_resource "set_input_default_value" + method to set default value in SDC. We use that + because different types of SDC resources has different + urls to call SDC API. + + Args: + value (Any): Default value to set + + """ + self.sdc_resource.set_input_default_value(self, value) + self._default_value = value + +@dataclass +class NestedInput: + """Dataclass used for nested input declaration.""" + + sdc_resource: "SdcResource" + input_obj: Input + + +class Property: # pylint: disable=too-many-instance-attributes, too-few-public-methods + """Service property class.""" + + def __init__(self, # pylint: disable=too-many-arguments + name: str, + property_type: str, + description: Optional[str] = None, + unique_id: Optional[str] = None, + parent_unique_id: Optional[str] = None, + sdc_resource: Optional["SdcResource"] = None, + value: Optional[Any] = None, + get_input_values: Optional[List[Dict[str, str]]] = None) -> None: + """Property class initialization. + + Args: + property_type (str): [description] + description (Optional[str], optional): [description]. Defaults to None. + unique_id (Optional[str], optional): [description]. Defaults to None. + parent_unique_id (Optional[str], optional): [description]. Defaults to None. + sdc_resource (Optional[, optional): [description]. Defaults to None. + value (Optional[Any], optional): [description]. Defaults to None. + get_input_values (Optional[List[Dict[str, str]]], optional): [description]. + Defaults to None. + """ + self.name: str = name + self.property_type: str = property_type + self.description: str = description + self.unique_id: str = unique_id + self.parent_unique_id: str = parent_unique_id + self.sdc_resource: "SdcResource" = sdc_resource + self._value: Any = value + self.get_input_values: List[Dict[str, str]] = get_input_values + + def __repr__(self) -> str: + """Property object human readable representation. + + Returns: + str: Property human readable representation + + """ + return f"Property(name={self.name}, property_type={self.property_type})" + + def __eq__(self, obj: "Property") -> bool: + """Check if two Property object are equal. + + Args: + obj (Property): Object to compare + + Returns: + bool: True if objects are equal, False otherwise + + """ + return self.name == obj.name and self.property_type == obj.property_type + + @property + def input(self) -> Input: + """Property input. + + Returns property Input object. + Returns None if property has no associated input. + + Raises: + ParameterError: Input has no associated SdcResource + + ParameterError: Input for given property does not exits. + It shouldn't ever happen, but it's possible if after you + get property object someone delete input. + + Returns: + Input: Property input object. + + """ + if not self.sdc_resource: + raise ParameterError("Property has no associated SdcResource") + if not self.get_input_values: + return None + try: + return next(filter(lambda x: x.unique_id == self.get_input_values[0].get("inputId"), + self.sdc_resource.inputs)) + except StopIteration: + raise ParameterError("Property input does not exist") + + @property + def value(self) -> Any: + """Value property. + + Get property value. + + Returns: + Any: Property value + + """ + return self._value + + @value.setter + def value(self, val: Any) -> Any: + if self.sdc_resource: + self.sdc_resource.set_property_value(self, val) + self._value = val + + +@dataclass +class ComponentProperty: + """Component property dataclass. + + Component properties are inputs objects in SDC, but in logic + it's a property. + + """ + + unique_id: str + property_type: str + name: str + component: "Component" + _value: Optional[Any] = field(repr=False, default=None) + + @property + def value(self) -> Any: + """Property value getter. + + Returns: + Any: Property value + + """ + return self._value + + @value.setter + def value(self, val: Any) -> None: + """Property value setter. + + Set value both in an object and in SDC using it's HTTP API. + + Args: + val (Any): Property value to set + + """ + self.component.set_property_value(self, val) + self._value = val diff --git a/src/onapsdk/sdc/sdc_element.py b/src/onapsdk/sdc/sdc_element.py new file mode 100644 index 0000000..df513b7 --- /dev/null +++ b/src/onapsdk/sdc/sdc_element.py @@ -0,0 +1,227 @@ +"""SDC Element 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. +from abc import ABC, abstractmethod +from operator import itemgetter +from typing import Any, Dict, List, Optional + +from onapsdk.sdc import SdcOnboardable +import onapsdk.constants as const + + +class SdcElement(SdcOnboardable, ABC): + """Mother Class of all SDC elements.""" + + ACTION_TEMPLATE = 'sdc_element_action.json.j2' + ACTION_METHOD = 'PUT' + + def __init__(self, name: str = None) -> None: + """Initialize the object.""" + super().__init__(name=name) + self.human_readable_version: Optional[str] = None + + def _get_item_details(self) -> Dict[str, Any]: + """ + Get item details. + + Returns: + Dict[str, Any]: the description of the item + + """ + if self.created(): + url = "{}/items/{}/versions".format(self._base_url(), + self.identifier) + results: Dict[str, Any] = self.send_message_json('GET', 'get item', url) + if results["listCount"] > 1: + items: List[Dict[str, Any]] = results["results"] + return sorted(items, key=itemgetter("creationTime"), reverse=True)[0] + return results["results"][0] + return {} + + def load(self) -> None: + """Load Object information from SDC.""" + vsp_details = self._get_item_details() + if vsp_details: + self._logger.debug("details found, updating") + self.version = vsp_details['id'] + self.human_readable_version = vsp_details["name"] + self.update_informations_from_sdc(vsp_details) + else: + # exists() method check if exists AND update identifier + self.exists() + + def update_informations_from_sdc(self, details: Dict[str, Any]) -> None: + """ + + Update instance with details from SDC. + + Args: + details ([type]): [description] + + """ + def update_informations_from_sdc_creation(self, + details: Dict[str, Any]) -> None: + """ + + Update instance with details from SDC after creation. + + Args: + details ([type]): the details from SDC + + """ + @classmethod + def _base_url(cls) -> str: + """ + Give back the base url of Sdc. + + Returns: + str: the base url + + """ + return "{}/sdc1/feProxy/onboarding-api/v1.0".format(cls.base_front_url) + + @classmethod + def _base_create_url(cls) -> str: + """ + Give back the base url of Sdc. + + Returns: + str: the base url + + """ + return "{}/sdc1/feProxy/onboarding-api/v1.0".format(cls.base_front_url) + + def _generate_action_subpath(self, action: str) -> str: + """ + + Generate subpath part of SDC action url. + + Args: + action (str): the action that will be done + + Returns: + str: the subpath part + + """ + subpath = self._sdc_path() + if action == const.COMMIT: + subpath = "items" + return subpath + + def _version_path(self) -> str: + """ + Give the end of the path for a version. + + Returns: + str: the end of the path + + """ + return "{}/versions/{}".format(self.identifier, self.version) + + @staticmethod + def _action_url(base: str, + subpath: str, + version_path: str, + action_type: str = None) -> str: + """ + Generate action URL for SDC. + + Args: + base (str): base part of url + subpath (str): subpath of url + version_path (str): version path of the url + action_type (str, optional): the type of action. UNUSED here + + Returns: + str: the URL to use + + """ + return "{}/{}/{}/actions".format(base, subpath, version_path) + + @classmethod + def _get_objects_list(cls, result: List[Dict[str, Any]] + ) -> List[Dict[str, Any]]: + """ + Import objects created in SDC. + + Args: + result (Dict[str, Any]): the result returned by SDC in a Dict + + Return: + List[Dict[str, Any]]: the list of objects + + """ + return result['results'] + + @classmethod + def _get_all_url(cls) -> str: + """ + Get URL for all elements in SDC. + + Returns: + str: the url + + """ + return "{}/{}".format(cls._base_url(), cls._sdc_path()) + + def _copy_object(self, obj: 'SdcElement') -> None: + """ + Copy relevant properties from object. + + Args: + obj (SdcElement): the object to "copy" + + """ + self.identifier = obj.identifier + + def _get_version_from_sdc(self, sdc_infos: Dict[str, Any]) -> str: + """ + Get version from SDC results. + + Args: + sdc_infos (Dict[str, Any]): the result dict from SDC + + Returns: + str: the version + + """ + return sdc_infos['version']['id'] + + def _get_identifier_from_sdc(self, sdc_infos: Dict[str, Any]) -> str: + """ + Get identifier from SDC results. + + Args: + sdc_infos (Dict[str, Any]): the result dict from SDC + + Returns: + str: the identifier + + """ + return sdc_infos['itemId'] + + @classmethod + @abstractmethod + def import_from_sdc(cls, values: Dict[str, Any]) -> 'SdcElement': + """ + Import SdcElement from SDC. + + Args: + values (Dict[str, Any]): dict to parse returned from SDC. + + Raises: + NotImplementedError: this is an abstract method. + + """ + raise NotImplementedError("SdcElement is an abstract class") diff --git a/src/onapsdk/sdc/sdc_resource.py b/src/onapsdk/sdc/sdc_resource.py new file mode 100644 index 0000000..7e7dbb9 --- /dev/null +++ b/src/onapsdk/sdc/sdc_resource.py @@ -0,0 +1,960 @@ +"""SDC Element 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 logging +from abc import ABC +from typing import Any, Dict, Iterator, List, Union +import base64 +import time + +import onapsdk.constants as const +from onapsdk.exceptions import ParameterError, ResourceNotFound, StatusError +from onapsdk.sdc import SdcOnboardable +from onapsdk.sdc.category_management import ResourceCategory, ServiceCategory +from onapsdk.sdc.component import Component +from onapsdk.sdc.properties import Input, NestedInput, Property +from onapsdk.utils.headers_creator import (headers_sdc_creator, + headers_sdc_tester, + headers_sdc_artifact_upload) +from onapsdk.utils.jinja import jinja_env + + +# For an unknown reason, pylint keeps seeing _unique_uuid and +# _unique_identifier as attributes along with unique_uuid and unique_identifier +class SdcResource(SdcOnboardable, ABC): # pylint: disable=too-many-instance-attributes, too-many-public-methods + """Mother Class of all SDC resources.""" + + RESOURCE_PATH = 'resources' + ACTION_TEMPLATE = 'sdc_resource_action.json.j2' + ACTION_METHOD = 'POST' + headers = headers_sdc_creator(SdcOnboardable.headers) + + def __init__(self, name: str = None, version: str = None, # pylint: disable=too-many-arguments + sdc_values: Dict[str, str] = None, properties: List[Property] = None, + inputs: Union[Property, NestedInput] = None, + category: str = None, subcategory: str = None): + """Initialize the object.""" + super().__init__(name) + self.version_filter: str = version + self._unique_uuid: str = None + self._unique_identifier: str = None + self._resource_type: str = "resources" + self._properties_to_add: List[Property] = properties or [] + self._inputs_to_add: Union[Property, NestedInput] = inputs or [] + self._time_wait: int = 10 + self._category_name: str = category + self._subcategory_name: str = subcategory + if sdc_values: + self._logger.debug("SDC values given, using them") + self.identifier = sdc_values['uuid'] + self.version = sdc_values['version'] + self.unique_uuid = sdc_values['invariantUUID'] + distribitution_state = None + if 'distributionStatus' in sdc_values: + distribitution_state = sdc_values['distributionStatus'] + self.status = self._parse_sdc_status(sdc_values['lifecycleState'], + distribitution_state, + self._logger) + self._logger.debug("SDC resource %s status: %s", self.name, + self.status) + + def __repr__(self) -> str: + """SDC resource description. + + Returns: + str: SDC resource object description + + """ + return f"{self.__class__.__name__.upper()}(name={self.name})" + + @property + def unique_uuid(self) -> str: + """Return and lazy load the unique_uuid.""" + if not self._unique_uuid: + self.load() + return self._unique_uuid + + @property + def unique_identifier(self) -> str: + """Return and lazy load the unique_identifier.""" + if not self._unique_identifier: + self.deep_load() + return self._unique_identifier + + @unique_uuid.setter + def unique_uuid(self, value: str) -> None: + """Set value for unique_uuid.""" + self._unique_uuid = value + + @unique_identifier.setter + def unique_identifier(self, value: str) -> None: + """Set value for unique_identifier.""" + self._unique_identifier = value + + def load(self) -> None: + """Load Object information from SDC.""" + self.exists() + + def deep_load(self) -> None: + """Deep load Object informations from SDC.""" + url = ( + f"{self.base_front_url}/sdc1/feProxy/rest/v1/" + "screen?excludeTypes=VFCMT&excludeTypes=Configuration" + ) + headers = headers_sdc_creator(SdcResource.headers) + if self.status == const.UNDER_CERTIFICATION: + headers = headers_sdc_tester(SdcResource.headers) + + response = self.send_message_json("GET", + "Deep Load {}".format( + type(self).__name__), + url, + headers=headers) + + for resource in response[self._sdc_path()]: + if resource["invariantUUID"] == self.unique_uuid: + if resource["uuid"] == self.identifier: + self._logger.debug("Resource %s found in %s list", + resource["name"], self._sdc_path()) + self.unique_identifier = resource["uniqueId"] + self._category_name = resource["categories"][0]["name"] + subcategories = resource["categories"][0].get("subcategories", [{}]) + self._subcategory_name = None if subcategories is None else \ + subcategories[0].get("name") + return + if self._sdc_path() == "services": + for dependency in self.send_message_json("GET", + "Get service dependecies", + f"{self._base_create_url()}/services/" + f"{resource['uniqueId']}/" + "dependencies"): + if dependency["version"] == self.version: + self.unique_identifier = dependency["uniqueId"] + return + + def _generate_action_subpath(self, action: str) -> str: + """ + + Generate subpath part of SDC action url. + + Args: + action (str): the action that will be done + + Returns: + str: the subpath part + + """ + return action + + def _version_path(self) -> str: + """ + Give the end of the path for a version. + + Returns: + str: the end of the path + + """ + return self.unique_identifier + + def _action_url(self, + base: str, + subpath: str, + version_path: str, + action_type: str = None) -> str: + """ + Generate action URL for SDC. + + Args: + base (str): base part of url + subpath (str): subpath of url + version_path (str): version path of the url + action_type (str, optional): the type of action + ('distribution', 'distribution-state' + or 'lifecycleState'). Default to + 'lifecycleState'). + + Returns: + str: the URL to use + + """ + if not action_type: + action_type = "lifecycleState" + return "{}/{}/{}/{}/{}".format(base, self._resource_type, version_path, + action_type, subpath) + + @classmethod + def _base_create_url(cls) -> str: + """ + Give back the base url of Sdc. + + Returns: + str: the base url + + """ + return "{}/sdc1/feProxy/rest/v1/catalog".format(cls.base_front_url) + + @classmethod + def _base_url(cls) -> str: + """ + Give back the base url of Sdc. + + Returns: + str: the base url + + """ + return "{}/sdc/v1/catalog".format(cls.base_back_url) + + @classmethod + def _get_all_url(cls) -> str: + """ + Get URL for all elements in SDC. + + Returns: + str: the url + + """ + return "{}/{}?resourceType={}".format(cls._base_url(), cls._sdc_path(), + cls.__name__.upper()) + + @classmethod + def _get_objects_list(cls, result: List[Dict[str, Any]] + ) -> List[Dict[str, Any]]: + """ + Import objects created in SDC. + + Args: + result (Dict[str, Any]): the result returned by SDC in a Dict + + Return: + List[Dict[str, Any]]: the list of objects + + """ + return result + + def _get_version_from_sdc(self, sdc_infos: Dict[str, Any]) -> str: + """ + Get version from SDC results. + + Args: + sdc_infos (Dict[str, Any]): the result dict from SDC + + Returns: + str: the version + + """ + return sdc_infos['version'] + + def _get_identifier_from_sdc(self, sdc_infos: Dict[str, Any]) -> str: + """ + Get identifier from SDC results. + + Args: + sdc_infos (Dict[str, Any]): the result dict from SDC + + Returns: + str: the identifier + + """ + return sdc_infos['uuid'] + + @classmethod + def import_from_sdc(cls, values: Dict[str, Any]) -> 'SdcResource': + """ + Import SdcResource from SDC. + + Args: + values (Dict[str, Any]): dict to parse returned from SDC. + + Return: + SdcResource: the created resource + + """ + cls._logger.debug("importing SDC Resource %s from SDC", values['name']) + return cls(name=values['name'], sdc_values=values) + + def _copy_object(self, obj: 'SdcResource') -> None: + """ + Copy relevant properties from object. + + Args: + obj (SdcResource): the object to "copy" + + """ + self.identifier = obj.identifier + self.unique_uuid = obj.unique_uuid + self.status = obj.status + self.version = obj.version + self.unique_identifier = obj.unique_identifier + self._specific_copy(obj) + + def _specific_copy(self, obj: 'SdcResource') -> None: + """ + Copy specific properties from object. + + Args: + obj (SdcResource): the object to "copy" + + """ + + def update_informations_from_sdc(self, details: Dict[str, Any]) -> None: + """ + + Update instance with details from SDC. + + Args: + details ([type]): [description] + + """ + def update_informations_from_sdc_creation(self, + details: Dict[str, Any]) -> None: + """ + + Update instance with details from SDC after creation. + + Args: + details ([type]): the details from SDC + + """ + self.unique_uuid = details['invariantUUID'] + distribution_state = None + + if 'distributionStatus' in details: + distribution_state = details['distributionStatus'] + self.status = self._parse_sdc_status(details['lifecycleState'], + distribution_state, self._logger) + self.version = details['version'] + self.unique_identifier = details['uniqueId'] + + # Not my fault if SDC has so many states... + # pylint: disable=too-many-return-statements + @staticmethod + def _parse_sdc_status(sdc_status: str, distribution_state: str, + logger: logging.Logger) -> str: + """ + Parse SDC status in order to normalize it. + + Args: + sdc_status (str): the status found in SDC + distribution_state (str): the distribution status found in SDC. + Can be None. + + Returns: + str: the normalized status + + """ + logger.debug("Parse status for SDC Resource") + if sdc_status.capitalize() == const.CERTIFIED: + if distribution_state and distribution_state == const.SDC_DISTRIBUTED: + return const.DISTRIBUTED + return const.CERTIFIED + if sdc_status == const.NOT_CERTIFIED_CHECKOUT: + return const.DRAFT + if sdc_status == const.NOT_CERTIFIED_CHECKIN: + return const.CHECKED_IN + if sdc_status == const.READY_FOR_CERTIFICATION: + return const.SUBMITTED + if sdc_status == const.CERTIFICATION_IN_PROGRESS: + return const.UNDER_CERTIFICATION + if sdc_status != "": + return sdc_status + return None + + def _really_submit(self) -> None: + """Really submit the SDC Vf in order to enable it.""" + raise NotImplementedError("SDC is an abstract class") + + def onboard(self) -> None: + """Onboard resource in SDC.""" + if not self.status: + self.create() + time.sleep(self._time_wait) + self.onboard() + elif self.status == const.DRAFT: + for property_to_add in self._properties_to_add: + self.add_property(property_to_add) + for input_to_add in self._inputs_to_add: + self.declare_input(input_to_add) + self.submit() + time.sleep(self._time_wait) + self.onboard() + elif self.status == const.CHECKED_IN: + # Checked in status check added + self.certify() + time.sleep(self._time_wait) + self.onboard() + elif self.status == const.CERTIFIED: + self.load() + + @classmethod + def _sdc_path(cls) -> None: + """Give back the end of SDC path.""" + return cls.RESOURCE_PATH + + @property + def deployment_artifacts_url(self) -> str: + """Deployment artifacts url. + + Returns: + str: SdcResource Deployment artifacts url + + """ + return (f"{self._base_create_url()}/resources/" + f"{self.unique_identifier}/filteredDataByParams?include=deploymentArtifacts") + + @property + def add_deployment_artifacts_url(self) -> str: + """Add deployment artifacts url. + + Returns: + str: Url used to add deployment artifacts + + """ + return (f"{self._base_create_url()}/resources/" + f"{self.unique_identifier}/artifacts") + + @property + def properties_url(self) -> str: + """Properties url. + + Returns: + str: SdcResource properties url + + """ + return (f"{self._base_create_url()}/resources/" + f"{self.unique_identifier}/filteredDataByParams?include=properties") + + @property + def add_property_url(self) -> str: + """Add property url. + + Returns: + str: Url used to add property + + """ + return (f"{self._base_create_url()}/services/" + f"{self.unique_identifier}/properties") + + @property + def set_input_default_value_url(self) -> str: + """Url to set input default value. + + Returns: + str: SDC API url used to set input default value + + """ + return (f"{self._base_create_url()}/resources/" + f"{self.unique_identifier}/update/inputs") + + @property + def origin_type(self) -> str: + """Resource origin type. + + Value needed for composition. It's used for adding SDC resource + as an another SDC resource component. + + Returns: + str: SDC resource origin type + + """ + return type(self).__name__.upper() + + @property + def properties(self) -> Iterator[Property]: + """SDC resource properties. + + Iterate resource properties. + + Yields: + Property: Resource property + + """ + for property_data in self.send_message_json(\ + "GET", + f"Get {self.name} resource properties", + self.properties_url).get("properties", []): + yield Property( + sdc_resource=self, + unique_id=property_data["uniqueId"], + name=property_data["name"], + property_type=property_data["type"], + parent_unique_id=property_data["parentUniqueId"], + value=property_data.get("value"), + description=property_data.get("description"), + get_input_values=property_data.get("getInputValues"), + ) + + def get_property(self, property_name: str) -> Property: + """Get resource property by it's name. + + Args: + property_name (str): property name + + Raises: + ResourceNotFound: Resource has no property with given name + + Returns: + Property: Resource's property object + + """ + for property_obj in self.properties: + if property_obj.name == property_name: + return property_obj + + msg = f"Resource has no property with {property_name} name" + raise ResourceNotFound(msg) + + @property + def resource_inputs_url(self) -> str: + """Resource inputs url. + + Method which returns url which point to resource inputs. + + Returns: + str: Resource inputs url + + """ + return (f"{self._base_create_url()}/resources/" + f"{self.unique_identifier}") + + + def create(self) -> None: + """Create resource. + + Abstract method which should be implemented by subclasses and creates resource in SDC. + + Raises: + NotImplementedError: Method not implemented by subclasses. + + """ + raise NotImplementedError + + @property + def inputs(self) -> Iterator[Input]: + """SDC resource inputs. + + Iterate resource inputs. + + Yields: + Iterator[Input]: Resource input + + """ + url = f"{self.resource_inputs_url}/filteredDataByParams?include=inputs" + for input_data in self.send_message_json(\ + "GET", f"Get {self.name} resource inputs", + url).get("inputs", []): + + yield Input( + unique_id=input_data["uniqueId"], + input_type=input_data["type"], + name=input_data["name"], + sdc_resource=self, + _default_value=input_data.get("defaultValue") + ) + + def get_input(self, input_name: str) -> Input: + """Get input by it's name. + + Args: + input_name (str): Input name + + Raises: + ResourceNotFound: Resource doesn't have input with given name + + Returns: + Input: Found input object + + """ + for input_obj in self.inputs: + if input_obj.name == input_name: + return input_obj + raise ResourceNotFound(f"SDC resource has no {input_name} input") + + def add_deployment_artifact(self, artifact_type: str, artifact_label: str, + artifact_name: str, artifact: str): + """ + Add deployment artifact to resource. + + Add deployment artifact to resource using payload data. + + Args: + artifact_type (str): all SDC artifact types are supported (DCAE_*, HEAT_*, ...) + artifact_name (str): the artifact file name including its extension + artifact (str): artifact file to upload + artifact_label (str): Unique Identifier of the artifact within the VF / Service. + + Raises: + StatusError: Resource has not DRAFT status + + """ + data = open(artifact, 'rb').read() + artifact_string = base64.b64encode(data).decode('utf-8') + if self.status != const.DRAFT: + msg = "Can't add artifact to resource which is not in DRAFT status" + raise StatusError(msg) + self._logger.debug("Add deployment artifact to sdc resource") + my_data = jinja_env().get_template( + "sdc_resource_add_deployment_artifact.json.j2").\ + render(artifact_name=artifact_name, + artifact_label=artifact_label, + artifact_type=artifact_type, + b64_artifact=artifact_string) + my_header = headers_sdc_artifact_upload(base_header=self.headers, data=my_data) + + self.send_message_json("POST", + f"Add deployment artifact for {self.name} sdc resource", + self.add_deployment_artifacts_url, + data=my_data, + headers=my_header) + + @property + def components(self) -> Iterator[Component]: + """Resource components. + + Iterate resource components. + + Yields: + Component: Resource component object + + """ + for component_instance in self.send_message_json(\ + "GET", + f"Get {self.name} resource inputs", + f"{self.resource_inputs_url}/filteredDataByParams?include=componentInstances" + ).get("componentInstances", []): + sdc_resource: "SdcResource" = SdcResource.import_from_sdc(self.send_message_json(\ + "GET", + f"Get {self.name} component's SDC resource metadata", + (f"{self.base_front_url}/sdc1/feProxy/rest/v1/catalog/resources/" + f"{component_instance['actualComponentUid']}/" + "filteredDataByParams?include=metadata"))["metadata"]) + yield Component.create_from_api_response(api_response=component_instance, + sdc_resource=sdc_resource, + parent_sdc_resource=self) + + @property + def category(self) -> Union[ResourceCategory, ServiceCategory]: + """Sdc resource category. + + Depends on the resource type returns ResourceCategory or ServiceCategory. + + Returns: + Uniton[ResourceCategory, ServiceCategory]: resource category + + """ + if self.created(): + if not any([self._category_name, self._subcategory_name]): + self.deep_load() + if all([self._category_name, self._subcategory_name]): + return ResourceCategory.get(name=self._category_name, + subcategory=self._subcategory_name) + return ServiceCategory.get(name=self._category_name) + return self.get_category_for_new_resource() + + def get_category_for_new_resource(self) -> ResourceCategory: + """Get category for resource not created in SDC yet. + + If no category values are provided default category is going to be used. + + Returns: + ResourceCategory: Category of the new resource + + """ + if not all([self._category_name, self._subcategory_name]): + return ResourceCategory.get(name="Generic", subcategory="Abstract") + return ResourceCategory.get(name=self._category_name, subcategory=self._subcategory_name) + + def get_component_properties_url(self, component: "Component") -> str: + """Url to get component's properties. + + This method is here because component can have different url when + it's a component of another SDC resource type, eg. for service and + for VF components have different urls. + + Args: + component (Component): Component object to prepare url for + + Returns: + str: Component's properties url + + """ + return (f"{self.resource_inputs_url}/" + f"componentInstances/{component.unique_id}/properties") + + def get_component_properties_value_set_url(self, component: "Component") -> str: + """Url to set component property value. + + This method is here because component can have different url when + it's a component of another SDC resource type, eg. for service and + for VF components have different urls. + + Args: + component (Component): Component object to prepare url for + + Returns: + str: Component's properties url + + """ + return (f"{self.resource_inputs_url}/" + f"resourceInstance/{component.unique_id}/properties") + + def is_own_property(self, property_to_check: Property) -> bool: + """Check if given property is one of the resource's properties. + + Args: + property_to_check (Property): Property to check + + Returns: + bool: True if resource has given property, False otherwise + + """ + return any(( + prop == property_to_check for prop in self.properties + )) + + def get_component(self, sdc_resource: "SdcResource") -> Component: + """Get resource's component. + + Get component by SdcResource object. + + Args: + sdc_resource (SdcResource): Component's SdcResource + + Raises: + ResourceNotFound: Component with given SdcResource does not exist + + Returns: + Component: Component object + + """ + for component in self.components: + if component.sdc_resource.name == sdc_resource.name: + return component + msg = f"SDC resource {sdc_resource.name} is not a component" + raise ResourceNotFound(msg) + + def get_component_by_name(self, component_name: str) -> Component: + """Get resource's component by it's name. + + Get component by name. + + Args: + component_name (str): Component's name + + Raises: + ResourceNotFound: Component with given name does not exist + + Returns: + Component: Component object + + """ + for component in self.components: + if component.sdc_resource.name == component_name: + return component + msg = f"SDC resource {component_name} is not a component" + raise ResourceNotFound(msg) + + def declare_input_for_own_property(self, property_obj: Property) -> None: + """Declare input for resource's property. + + For each property input can be declared. + + Args: + property_obj (Property): Property to declare input + + """ + self._logger.debug("Declare input for SDC resource property") + self.send_message_json("POST", + f"Declare new input for {property_obj.name} property", + f"{self.resource_inputs_url}/create/inputs", + data=jinja_env().get_template(\ + "sdc_resource_add_input.json.j2").\ + render(\ + sdc_resource=self, + property=property_obj)) + + def declare_nested_input(self, + nested_input: NestedInput) -> None: + """Declare nested input for SDC resource. + + Nested input is an input of one of the components. + + Args: + nested_input (NestedInput): Nested input object + + """ + self._logger.debug("Declare input for SDC resource's component property") + component: Component = self.get_component(nested_input.sdc_resource) + self.send_message_json("POST", + f"Declare new input for {nested_input.input_obj.name} input", + f"{self.resource_inputs_url}/create/inputs", + data=jinja_env().get_template(\ + "sdc_resource_add_nested_input.json.j2").\ + render(\ + sdc_resource=self, + component=component, + input=nested_input.input_obj)) + + def declare_input(self, input_to_declare: Union[Property, NestedInput]) -> None: + """Declare input for given property or nested input object. + + Call SDC FE API to declare input for given property. + + Args: + input_declaration (Union[Property, NestedInput]): Property to declare input + or NestedInput object + + Raises: + ParameterError: if the given property is not SDC resource property + + """ + self._logger.debug("Declare input") + if isinstance(input_to_declare, Property): + if self.is_own_property(input_to_declare): + self.declare_input_for_own_property(input_to_declare) + else: + msg = "Given property is not SDC resource property" + raise ParameterError(msg) + else: + self.declare_nested_input(input_to_declare) + + def add_property(self, property_to_add: Property) -> None: + """Add property to resource. + + Call SDC FE API to add property to resource. + + Args: + property_to_add (Property): Property object to add to resource. + + Raises: + StatusError: Resource has not DRAFT status + + """ + if self.status != const.DRAFT: + msg = "Can't add property to resource which is not in DRAFT status" + raise StatusError(msg) + self._logger.debug("Add property to sdc resource") + self.send_message_json("POST", + f"Declare new property for {self.name} sdc resource", + self.add_property_url, + data=jinja_env().get_template( + "sdc_resource_add_property.json.j2").\ + render( + property=property_to_add + )) + + def set_property_value(self, property_obj: Property, value: Any) -> None: + """Set property value. + + Set given value to resource property + + Args: + property_obj (Property): Property object + value (Any): Property value to set + + Raises: + ParameterError: if the given property is not the resource's property + + """ + if not self.is_own_property(property_obj): + raise ParameterError("Given property is not a resource's property") + self._logger.debug("Set %s property value", property_obj.name) + self.send_message_json("PUT", + f"Set {property_obj.name} value to {value}", + self.add_property_url, + data=jinja_env().get_template( + "sdc_resource_set_property_value.json.j2").\ + render( + sdc_resource=self, + property=property_obj, + value=value + ) + ) + + def set_input_default_value(self, input_obj: Input, default_value: Any) -> None: + """Set input default value. + + Set given value as input default value + + Args: + input_obj (Input): Input object + value (Any): Default value to set + + """ + self._logger.debug("Set %s input default value", input_obj.name) + self.send_message_json("POST", + f"Set {input_obj.name} default value to {default_value}", + self.set_input_default_value_url, + data=jinja_env().get_template( + "sdc_resource_set_input_default_value.json.j2").\ + render( + sdc_resource=self, + input=input_obj, + default_value=default_value + ) + ) + + def checkout(self) -> None: + """Checkout SDC resource.""" + self._logger.debug("Checkout %s SDC resource", self.name) + result = self._action_to_sdc(const.CHECKOUT, "lifecycleState") + if result: + self.load() + + def undo_checkout(self) -> None: + """Undo Checkout SDC resource.""" + self._logger.debug("Undo Checkout %s SDC resource", self.name) + result = self._action_to_sdc(const.UNDOCHECKOUT, "lifecycleState") + if result: + self.load() + + def certify(self) -> None: + """Certify SDC resource.""" + self._logger.debug("Certify %s SDC resource", self.name) + result = self._action_to_sdc(const.CERTIFY, "lifecycleState") + if result: + self.load() + + def add_resource(self, resource: 'SdcResource') -> None: + """ + Add a Resource. + + Args: + resource (SdcResource): the resource to add + + """ + if self.status == const.DRAFT: + url = "{}/{}/{}/resourceInstance".format(self._base_create_url(), + self._sdc_path(), + self.unique_identifier) + + template = jinja_env().get_template( + "add_resource_to_service.json.j2") + data = template.render(resource=resource, + resource_type=resource.origin_type) + result = self.send_message("POST", + f"Add {resource.origin_type} to {self.origin_type}", + url, + data=data) + if result: + self._logger.info("Resource %s %s has been added on %s %s", + resource.origin_type, resource.name, + self.origin_type, self.name) + return result + self._logger.error(("an error occured during adding resource %s %s" + " on %s %s in SDC"), + resource.origin_type, resource.name, + self.origin_type, self.name) + return None + msg = f"Can't add resource to {self.origin_type} which is not in DRAFT status" + raise StatusError(msg) diff --git a/src/onapsdk/sdc/service.py b/src/onapsdk/sdc/service.py new file mode 100644 index 0000000..c866fa4 --- /dev/null +++ b/src/onapsdk/sdc/service.py @@ -0,0 +1,932 @@ +"""Service 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 base64 +import pathlib as Path +import time +from dataclasses import dataclass, field +from enum import Enum +from io import BytesIO, TextIOWrapper +from os import makedirs +from typing import Dict, List, Callable, Iterator, Optional, Type, Union, Any, BinaryIO +from zipfile import ZipFile, BadZipFile + +import oyaml as yaml +from requests import Response + +import onapsdk.constants as const +from onapsdk.exceptions import (ParameterError, RequestError, ResourceNotFound, + StatusError, ValidationError) +from onapsdk.sdc.category_management import ServiceCategory +from onapsdk.sdc.properties import NestedInput, Property +from onapsdk.sdc.sdc_resource import SdcResource +from onapsdk.utils.configuration import (components_needing_distribution, + tosca_path) +from onapsdk.utils.headers_creator import headers_sdc_creator, headers_sdc_artifact_upload +from onapsdk.utils.jinja import jinja_env + + +@dataclass +class VfModule: # pylint: disable=too-many-instance-attributes + """VfModule dataclass.""" + + name: str + group_type: str + model_name: str + model_version_id: str + model_invariant_uuid: str + model_version: str + model_customization_id: str + properties: Iterator[Property] + + +@dataclass +class NodeTemplate: # pylint: disable=too-many-instance-attributes + """Node template dataclass. + + Base class for Vnf, Pnf and Network classes. + """ + + name: str + node_template_type: str + model_name: str + model_version_id: str + model_invariant_id: str + model_version: str + model_customization_id: str + model_instance_name: str + component: "Component" + + @property + def properties(self) -> Iterator["Property"]: + """Node template properties. + + Returns: + Iterator[Property]: Node template properties iterator + + """ + return self.component.properties + + +@dataclass +class Vnf(NodeTemplate): + """Vnf dataclass.""" + + vf_modules: List[VfModule] = field(default_factory=list) + + +@dataclass +class Pnf(NodeTemplate): + """Pnf dataclass.""" + + +class Network(NodeTemplate): # pylint: disable=too-few-public-methods + """Network dataclass.""" + + +class ServiceInstantiationType(Enum): + """Service instantiation type enum class. + + Service can be instantiated using `A-la-carte` or `Macro` flow. + It has to be determined during design time. That class stores these + two values to set during initialization. + + """ + + A_LA_CARTE = "A-la-carte" + MACRO = "Macro" + + +class Service(SdcResource): # pylint: disable=too-many-instance-attributes, too-many-public-methods + """ + ONAP Service Object used for SDC operations. + + Attributes: + name (str): the name of the service. Defaults to "ONAP-test-Service". + identifier (str): the unique ID of the service from SDC. + status (str): the status of the service from SDC. + version (str): the version ID of the service from SDC. + uuid (str): the UUID of the Service (which is different from + identifier, don't ask why...) + distribution_status (str): the status of distribution in the different + ONAP parts. + distribution_id (str): the ID of the distribution when service is + distributed. + distributed (bool): True if the service is distributed + unique_identifier (str): Yet Another ID, just to puzzle us... + + """ + + SERVICE_PATH = "services" + + def __init__(self, name: str = None, version: str = None, sdc_values: Dict[str, str] = None, # pylint: disable=too-many-arguments + resources: List[SdcResource] = None, properties: List[Property] = None, + inputs: List[Union[Property, NestedInput]] = None, + instantiation_type: Optional[ServiceInstantiationType] = \ + None, + category: str = None, role: str = "", function: str = "", service_type: str = ""): + """ + Initialize service object. + + Args: + name (str, optional): the name of the service + version (str, optional): the version of the service + sdc_values (Dict[str, str], optional): dictionary of values + returned by SDC + resources (List[SdcResource], optional): list of SDC resources + properties (List[Property], optional): list of properties to add to service. + None by default. + inputs (List[Union[Property, NestedInput]], optional): list of inputs + to declare for service. It can be both Property or NestedInput object. + None by default. + instantiation_type (ServiceInstantiationType, optional): service instantiation + type. ServiceInstantiationType.A_LA_CARTE by default + category (str, optional): service category name + role (str, optional): service role + function (str, optional): service function. Empty by default + service_type (str, optional): service type. Empty by default + + """ + super().__init__(sdc_values=sdc_values, version=version, properties=properties, + inputs=inputs, category=category) + self.name: str = name or "ONAP-test-Service" + self.distribution_status = None + self.category_name: str = category + self.role: str = role + self.function: str = function + self.service_type: str = service_type + if sdc_values: + self.distribution_status = sdc_values['distributionStatus'] + self.category_name = sdc_values["category"] + self.resources = resources or [] + self._instantiation_type: Optional[ServiceInstantiationType] = instantiation_type + self._distribution_id: str = None + self._distributed: bool = False + self._resource_type: str = "services" + self._tosca_model: bytes = None + self._tosca_template: str = None + self._vnfs: list = None + self._pnfs: list = None + self._networks: list = None + self._vf_modules: list = None + + @classmethod + def get_by_unique_uuid(cls, unique_uuid: str) -> "Service": + """Get the service model using unique uuid. + + Returns: + Service: object with provided unique_uuid + + Raises: + ResourceNotFound: No service with given unique_uuid exists + + """ + services: List["Service"] = cls.get_all() + for service in services: + if service.unique_uuid == unique_uuid: + return service + raise ResourceNotFound("Service with given unique uuid doesn't exist") + + def onboard(self) -> None: + """Onboard the Service in SDC. + + Raises: + StatusError: service has an invalid status + ParameterError: no resources, no properties for service + in DRAFT status + + """ + # first Lines are equivalent for all onboard functions but it's more + # readable + if not self.status: + # equivalent step as in onboard-function in sdc_resource + self.create() + time.sleep(self._time_wait) + self.onboard() + elif self.status == const.DRAFT: + if not any([self.resources, self._properties_to_add]): + raise ParameterError("No resources nor properties were given") + self.declare_resources_and_properties() + self.checkin() + time.sleep(self._time_wait) + self.onboard() + elif self.status == const.CHECKED_IN: + self.certify() + time.sleep(self._time_wait) + self.onboard() + elif self.status == const.CERTIFIED: + self.distribute() + self.onboard() + elif self.status == const.DISTRIBUTED: + self._logger.info("Service %s onboarded", self.name) + else: + self._logger.error("Service has invalid status: %s", self.status) + raise StatusError(self.status) + + @property + def distribution_id(self) -> str: + """Return and lazy load the distribution_id.""" + if not self._distribution_id: + self.load_metadata() + return self._distribution_id + + @distribution_id.setter + def distribution_id(self, value: str) -> None: + """Set value for distribution_id.""" + self._distribution_id = value + + @property + def distributed(self) -> bool: + """Return and lazy load the distributed state.""" + if not self._distributed: + self._check_distributed() + return self._distributed + + @property + def tosca_template(self) -> str: + """Service tosca template file. + + Get tosca template from service tosca model bytes. + + Returns: + str: Tosca template file + + """ + if not self._tosca_template and self.tosca_model: + self._unzip_csar_file(BytesIO(self.tosca_model), + self._load_tosca_template) + return self._tosca_template + + @property + def tosca_model(self) -> bytes: + """Service's tosca model file. + + Send request to get service TOSCA model, + + Returns: + bytes: TOSCA model file bytes + + """ + if not self._tosca_model: + url = "{}/services/{}/toscaModel".format(self._base_url(), + self.identifier) + headers = self.headers.copy() + headers["Accept"] = "application/octet-stream" + self._tosca_model = self.send_message( + "GET", + "Download Tosca Model for {}".format(self.name), + url, + headers=headers).content + return self._tosca_model + + def create_node_template(self, + node_template_type: Type[NodeTemplate], + component: "Component") -> NodeTemplate: + """Create a node template type object. + + The base of the all node template types objects (Vnf, Pnf, Network) is the + same. The difference is only for the Vnf which can have vf modules associated with. + Vf modules could have "vf_module_label" property with"base_template_dummy_ignore" + value. These vf modules should be ignored/ + + Args: + node_template_type (Type[NodeTemplate]): Node template class type + component (Component): Component on which base node template object should be created + + Returns: + NodeTemplate: Node template object created from component + + """ + node_template: NodeTemplate = node_template_type( + name=component.name, + node_template_type=component.tosca_component_name, + model_name=component.component_name, + model_version_id=component.sdc_resource.identifier, + model_invariant_id=component.sdc_resource.unique_uuid, + model_version=component.sdc_resource.version, + model_customization_id=component.customization_uuid, + model_instance_name=self.name, + component=component + ) + if node_template_type is Vnf: + if component.group_instances: + for vf_module in component.group_instances: + if not any([property_def["name"] == "vf_module_label"] and \ + property_def["value"] == "base_template_dummy_ignore" for \ + property_def in vf_module["properties"]): + node_template.vf_modules.append(VfModule( + name=vf_module["name"], + group_type=vf_module["type"], + model_name=vf_module["groupName"], + model_version_id=vf_module["groupUUID"], + model_invariant_uuid=vf_module["invariantUUID"], + model_version=vf_module["version"], + model_customization_id=vf_module["customizationUUID"], + properties=( + Property( + name=property_def["name"], + property_type=property_def["type"], + description=property_def["description"], + value=property_def["value"] + ) for property_def in vf_module["properties"] \ + if property_def["value"] and not ( + property_def["name"] == "vf_module_label" and \ + property_def["value"] == "base_template_dummy_ignore" + ) + ) + )) + return node_template + + def __has_component_type(self, origin_type: str) -> bool: + """Check if any of Service's component type is provided origin type. + + In template generation is checked if Service has some types of components, + based on that blocks are added to the request template. It's not + the best option to get all components to check if at least one with + given type exists for conditional statement. + + Args: + origin_type (str): Type to check if any component exists. + + Returns: + bool: True if service has at least one component with given origin type, + False otherwise + + """ + return any((component.origin_type == origin_type for component in self.components)) + + @property + def has_vnfs(self) -> bool: + """Check if service has at least one VF component.""" + return self.__has_component_type("VF") + + @property + def has_pnfs(self) -> bool: + """Check if service has at least one PNF component.""" + return self.__has_component_type("PNF") + + @property + def has_vls(self) -> bool: + """Check if service has at least one VL component.""" + return self.__has_component_type("VL") + + @property + def vnfs(self) -> Iterator[Vnf]: + """Service Vnfs. + + Load VNFs from components generator. + It creates a generator of the vf modules as well, but without + vf modules which has "vf_module_label" property value equal + to "base_template_dummy_ignore". + + Returns: + Iterator[Vnf]: Vnf objects iterator + + """ + for component in self.components: + if component.origin_type == "VF": + yield self.create_node_template(Vnf, component) + + @property + def pnfs(self) -> Iterator[Pnf]: + """Service Pnfs. + + Load PNFS from components generator. + + Returns: + Iterator[Pnf]: Pnf objects generator + + """ + for component in self.components: + if component.origin_type == "PNF": + yield self.create_node_template(Pnf, component) + + @property + def networks(self) -> Iterator[Network]: + """Service networks. + + Load networks from service's components generator. + + Returns: + Iterator[Network]: Network objects generator + + """ + for component in self.components: + if component.origin_type == "VL": + yield self.create_node_template(Network, component) + + @property + def deployment_artifacts_url(self) -> str: + """Deployment artifacts url. + + Returns: + str: SdcResource Deployment artifacts url + + """ + return (f"{self._base_create_url()}/services/" + f"{self.unique_identifier}/filteredDataByParams?include=deploymentArtifacts") + + @property + def add_deployment_artifacts_url(self) -> str: + """Add deployment artifacts url. + + Returns: + str: Url used to add deployment artifacts + + """ + return (f"{self._base_create_url()}/services/" + f"{self.unique_identifier}/artifacts") + + @property + def properties_url(self) -> str: + """Properties url. + + Returns: + str: SdcResource properties url + + """ + return (f"{self._base_create_url()}/services/" + f"{self.unique_identifier}/filteredDataByParams?include=properties") + + @property + def metadata_url(self) -> str: + """Metadata url. + + Returns: + str: Service metadata url + + """ + return (f"{self._base_create_url()}/services/" + f"{self.unique_identifier}/filteredDataByParams?include=metadata") + + @property + def resource_inputs_url(self) -> str: + """Service inputs url. + + Returns: + str: Service inputs url + + """ + return (f"{self._base_create_url()}/services/" + f"{self.unique_identifier}") + + @property + def set_input_default_value_url(self) -> str: + """Url to set input default value. + + Returns: + str: SDC API url used to set input default value + + """ + return (f"{self._base_create_url()}/services/" + f"{self.unique_identifier}/update/inputs") + + @property + def origin_type(self) -> str: + """Service origin type. + + Value needed for composition. It's used for adding SDC resource + as an another SDC resource component. + For Service that value has to be set to "ServiceProxy". + + Returns: + str: Service resource origin type + + """ + return "ServiceProxy" + + @property + def instantiation_type(self) -> ServiceInstantiationType: + """Service instantiation type. + + One of `ServiceInstantiationType` enum value. + + Returns: + ServiceInstantiationType: Service instantiation type + + """ + if not self._instantiation_type: + if not self.created(): + self._instantiation_type = ServiceInstantiationType.A_LA_CARTE + else: + response: str = self.send_message_json("GET", + f"Get service {self.name} metadata", + self.metadata_url)["metadata"]\ + ["instantiationType"] + self._instantiation_type = ServiceInstantiationType(response) + return self._instantiation_type + + def create(self) -> None: + """Create the Service in SDC if not already existing.""" + self._create("service_create.json.j2", + name=self.name, + instantiation_type=self.instantiation_type.value, + category=self.category, + role=self.role, service_type=self.service_type, function=self.function) + + def declare_resources_and_properties(self) -> None: + """Delcare resources and properties. + + It declares also inputs. + + """ + for resource in self.resources: + self.add_resource(resource) + for property_to_add in self._properties_to_add: + self.add_property(property_to_add) + for input_to_add in self._inputs_to_add: + self.declare_input(input_to_add) + + def checkin(self) -> None: + """Checkin Service.""" + self._verify_lcm_to_sdc(const.DRAFT, const.CHECKIN) + + def submit(self) -> None: + """Really submit the SDC Service.""" + self._verify_lcm_to_sdc(const.CHECKED_IN, const.SUBMIT_FOR_TESTING) + + def start_certification(self) -> None: + """Start Certification on Service.""" + headers = headers_sdc_creator(SdcResource.headers) + self._verify_lcm_to_sdc(const.CHECKED_IN, + const.START_CERTIFICATION, + headers=headers) + + def certify(self) -> None: + """Certify Service in SDC.""" + headers = headers_sdc_creator(SdcResource.headers) + self._verify_lcm_to_sdc(const.CHECKED_IN, + const.CERTIFY, + headers=headers) + + def approve(self) -> None: + """Approve Service in SDC.""" + headers = headers_sdc_creator(SdcResource.headers) + self._verify_approve_to_sdc(const.CERTIFIED, + const.APPROVE, + headers=headers) + + def distribute(self) -> None: + """Apptove Service in SDC.""" + headers = headers_sdc_creator(SdcResource.headers) + self._verify_distribute_to_sdc(const.CERTIFIED, + const.DISTRIBUTE, + headers=headers) + + def redistribute(self) -> None: + """Apptove Service in SDC.""" + headers = headers_sdc_creator(SdcResource.headers) + self._verify_distribute_to_sdc(const.DISTRIBUTED, + const.DISTRIBUTE, + headers=headers) + + def get_tosca(self) -> None: + """Get Service tosca files and save it.""" + url = "{}/services/{}/toscaModel".format(self._base_url(), + self.identifier) + headers = self.headers.copy() + headers["Accept"] = "application/octet-stream" + result = self.send_message("GET", + "Download Tosca Model for {}".format( + self.name), + url, + headers=headers) + if result: + self._create_tosca_file(result) + + def _create_tosca_file(self, result: Response) -> None: + """Create Service Tosca files from HTTP response.""" + csar_filename = "service-{}-csar.csar".format(self.name) + makedirs(tosca_path(), exist_ok=True) + with open((tosca_path() + csar_filename), 'wb') as csar_file: + for chunk in result.iter_content(chunk_size=128): + csar_file.write(chunk) + try: + self._unzip_csar_file(tosca_path() + csar_filename, + self._write_csar_file) + except BadZipFile as exc: + self._logger.exception(exc) + + def _check_distributed(self) -> bool: + """Check if service is distributed and update status accordingly.""" + url = "{}/services/distribution/{}".format(self._base_create_url(), + self.distribution_id) + headers = headers_sdc_creator(SdcResource.headers) + + status = {} + for component in components_needing_distribution(): + status[component] = False + + try: + result = self.send_message_json("GET", + "Check distribution for {}".format( + self.name), + url, + headers=headers) + except ResourceNotFound: + msg = f"No distributions found for {self.name} of {self.__class__.__name__}." + self._logger.debug(msg) + else: + status = self._update_components_status(status, result) + + for state in status.values(): + if not state: + self._distributed = False + return + self._distributed = True + + def _update_components_status(self, status: Dict[str, bool], + result: Response) -> Dict[str, bool]: + """Update components distribution status.""" + distrib_list = result['distributionStatusList'] + self._logger.debug("[SDC][Get Distribution] distrib_list = %s", + distrib_list) + for elt in distrib_list: + status = self._parse_components_status(status, elt) + return status + + def _parse_components_status(self, status: Dict[str, bool], + element: Dict[str, Any]) -> Dict[str, bool]: + """Parse components distribution status.""" + for key in status: + if ((key in element['omfComponentID']) + and (const.DOWNLOAD_OK in element['status'])): + status[key] = True + self._logger.info(("[SDC][Get Distribution] Service " + "distributed in %s"), key) + return status + + def load_metadata(self) -> None: + """Load Metada of Service and retrieve informations.""" + url = "{}/services/{}/distribution".format(self._base_create_url(), + self.identifier) + headers = headers_sdc_creator(SdcResource.headers) + result = self.send_message_json("GET", + "Get Metadata for {}".format( + self.name), + url, + headers=headers) + if ('distributionStatusOfServiceList' in result + and len(result['distributionStatusOfServiceList']) > 0): + # API changed and the latest distribution is not added to the end + # of distributions list but inserted as the first one. + dist_status = result['distributionStatusOfServiceList'][0] + self._distribution_id = dist_status['distributionID'] + + @classmethod + def _get_all_url(cls) -> str: + """ + Get URL for all elements in SDC. + + Returns: + str: the url + + """ + return "{}/{}".format(cls._base_url(), cls._sdc_path()) + + def _really_submit(self) -> None: + """Really submit the SDC Service in order to enable it.""" + result = self._action_to_sdc(const.CERTIFY, + action_type="lifecycleState") + if result: + self.load() + + def _specific_copy(self, obj: 'Service') -> None: + """ + Copy specific properties from object. + + Args: + obj (Service): the object to "copy" + + """ + super()._specific_copy(obj) + self.category_name = obj.category_name + self.role = obj.role + + def _verify_distribute_to_sdc(self, desired_status: str, + desired_action: str, **kwargs) -> None: + self._verify_action_to_sdc(desired_status, desired_action, + "distribution", **kwargs) + + def _verify_approve_to_sdc(self, desired_status: str, desired_action: str, + **kwargs) -> None: + self._verify_action_to_sdc(desired_status, desired_action, + "distribution-state", **kwargs) + + def _verify_lcm_to_sdc(self, desired_status: str, desired_action: str, + **kwargs) -> None: + self._verify_action_to_sdc(desired_status, desired_action, + "lifecycleState", **kwargs) + + def _verify_action_to_sdc(self, desired_status: str, desired_action: str, + action_type: str, **kwargs) -> None: + """ + Verify action to SDC. + + Verify that object is in right state before launching the action on + SDC. + + Raises: + StatusError: if current status is not the desired status. + + Args: + desired_status (str): the status the object should be + desired_action (str): the action we want to perform + action_type (str): the type of action ('distribution-state' or + 'lifecycleState') + **kwargs: any specific stuff to give to requests + + """ + self._logger.info("attempting to %s Service %s in SDC", desired_action, + self.name) + if self.status == desired_status and self.created(): + self._action_to_sdc(desired_action, + action_type=action_type, + **kwargs) + self.load() + elif not self.created(): + self._logger.warning("Service %s in SDC is not created", self.name) + elif self.status != desired_status: + msg = (f"Service {self.name} in SDC is in status {self.status} " + f"and it should be in status {desired_status}") + raise StatusError(msg) + + @staticmethod + def _unzip_csar_file(zip_file: Union[str, BytesIO], + function: Callable[[str, + TextIOWrapper], None]) -> None: + """ + Unzip Csar File and perform an action on the file. + + Raises: + ValidationError: CSAR file has no service template + + """ + folder = "Definitions" + prefix = "service-" + suffix = "-template.yml" + with ZipFile(zip_file) as myzip: + service_template = None + for name in myzip.namelist(): + if (name[-13:] == suffix + and name[:20] == f"{folder}/{prefix}"): + service_template = name + + if not service_template: + msg = (f"CSAR file has no service template. " + f"Valid path: {folder}/{prefix}*{suffix}") + raise ValidationError(msg) + + with myzip.open(service_template) as template_file: + function(service_template, template_file) + + @staticmethod + def _write_csar_file(service_template: str, + template_file: TextIOWrapper) -> None: + """Write service temple into a file.""" + with open(tosca_path() + service_template[12:], 'wb') as file: + file.write(template_file.read()) + + # _service_template is not used but function generation is generic + # pylint: disable-unused-argument + def _load_tosca_template(self, _service_template: str, + template_file: TextIOWrapper) -> None: + """Load Tosca template.""" + self._tosca_template = yaml.safe_load(template_file.read()) + + @classmethod + def _sdc_path(cls) -> None: + """Give back the end of SDC path.""" + return cls.SERVICE_PATH + + def get_nf_unique_id(self, nf_name: str) -> str: + """ + Get nf (network function) uniqueID. + + Get nf uniqueID from service nf in sdc. + + Args: + nf_name (str): the nf from which we extract the unique ID + + Returns: + the nf unique ID + + Raises: + ResourceNotFound: Couldn't find NF by name. + + """ + url = f"{self._base_create_url()}/services/{self.unique_identifier}" + request_return = self.send_message_json('GET', + 'Get nf unique ID', + url) + + for instance in filter(lambda x: x["componentName"] == nf_name, + request_return["componentInstances"]): + return instance["uniqueId"] + + raise ResourceNotFound(f"NF '{nf_name}'") + + + def add_artifact_to_vf(self, vnf_name: str, artifact_type: str, + artifact_name: str, artifact: BinaryIO = None): + """ + Add artifact to vf. + + Add artifact to vf using payload data. + + Raises: + RequestError: file upload (POST request) for an artifact fails. + + Args: + vnf_name (str): the vnf which we want to add the artifact + artifact_type (str): all SDC artifact types are supported (DCAE_*, HEAT_*, ...) + artifact_name (str): the artifact file name including its extension + artifact (str): binary data to upload + + """ + missing_identifier = self.get_nf_unique_id(vnf_name) + url = (f"{self._base_create_url()}/services/{self.unique_identifier}/" + f"resourceInstance/{missing_identifier}/artifacts") + template = jinja_env().get_template("add_artifact_to_vf.json.j2") + data = template.render(artifact_name=artifact_name, + artifact_label=f"sdk{Path.PurePosixPath(artifact_name).stem}", + artifact_type=artifact_type, + b64_artifact=base64.b64encode(artifact).decode('utf-8')) + headers = headers_sdc_artifact_upload(base_header=self.headers, + data=data) + try: + self.send_message('POST', + 'Add artifact to vf', + url, + headers=headers, + data=data) + except RequestError as exc: + self._logger.error(("an error occured during file upload for an Artifact" + "to VNF %s"), vnf_name) + raise exc + + def get_component_properties_url(self, component: "Component") -> str: + """Url to get component's properties. + + This method is here because component can have different url when + it's a component of another SDC resource type, eg. for service and + for VF components have different urls. + Also for VL origin type components properties url is different than + for the other types. + + Args: + component (Component): Component object to prepare url for + + Returns: + str: Component's properties url + + """ + if component.origin_type == "VL": + return super().get_component_properties_url(component) + return (f"{self.resource_inputs_url}/" + f"componentInstances/{component.unique_id}/{component.actual_component_uid}/inputs") + + def get_component_properties_value_set_url(self, component: "Component") -> str: + """Url to set component property value. + + This method is here because component can have different url when + it's a component of another SDC resource type, eg. for service and + for VF components have different urls. + Also for VL origin type components properties url is different than + for the other types. + + Args: + component (Component): Component object to prepare url for + + Returns: + str: Component's properties url + + """ + if component.origin_type == "VL": + return super().get_component_properties_value_set_url(component) + return (f"{self.resource_inputs_url}/" + f"resourceInstance/{component.unique_id}/inputs") + + def get_category_for_new_resource(self) -> ServiceCategory: + """Get category for service not created in SDC yet. + + If no category values are provided default category is going to be used. + + Returns: + ServiceCategory: Category of the new service + + """ + if not self._category_name: + return ServiceCategory.get(name="Network Service") + return ServiceCategory.get(name=self._category_name) diff --git a/src/onapsdk/sdc/templates/add_artifact_to_vf.json.j2 b/src/onapsdk/sdc/templates/add_artifact_to_vf.json.j2 new file mode 100644 index 0000000..2b0446c --- /dev/null +++ b/src/onapsdk/sdc/templates/add_artifact_to_vf.json.j2 @@ -0,0 +1,9 @@ +{ + "artifactGroupType": "DEPLOYMENT", + "artifactName": "{{artifact_name}}", + "artifactLabel": "{{artifact_label}}", + "artifactType": "{{artifact_type}}", + "description": "test", + "payloadData": "{{b64_artifact}}", + "heatParameters": [] +} diff --git a/src/onapsdk/sdc/templates/add_resource_to_service.json.j2 b/src/onapsdk/sdc/templates/add_resource_to_service.json.j2 new file mode 100644 index 0000000..d6676e9 --- /dev/null +++ b/src/onapsdk/sdc/templates/add_resource_to_service.json.j2 @@ -0,0 +1,10 @@ +{ + "name": "{{ resource.name }}", + "componentVersion": "{{ resource.version }}", + "posY": {{ posY| default(100) }}, + "posX": {{ posX| default(200) }}, + "uniqueId": "{{ resource.unique_identifier }}", + "originType": "{{ resource_type }}", + "componentUid": "{{ resource.unique_identifier }}", + "icon": "defaulticon" +} diff --git a/src/onapsdk/sdc/templates/component_declare_input.json.j2 b/src/onapsdk/sdc/templates/component_declare_input.json.j2 new file mode 100644 index 0000000..fd0ee03 --- /dev/null +++ b/src/onapsdk/sdc/templates/component_declare_input.json.j2 @@ -0,0 +1,37 @@ +{ + "componentInstanceInputsMap": {}, + "componentInstanceProperties": { + "{{ component.unique_id }}": [ + { + "constraints": null, + "defaultValue": null, + "description": "", + "name": "{{ property.name }}", + "origName": "{{ property.name }}", + "parentUniqueId": null, + "password": false, + "required": true, + "schema": { + "property": {} + }, + "schemaType": null, + "type": "{{ property.property_type }}", + "uniqueId": "{{ property.unique_id }}", + {% if property.value is not none %} + "value":"{{ property.value }}", + {% else %} + "value":null, + {% endif %} + "definition": false, + "getInputValues": null, + "parentPropertyType": null, + "subPropertyInputPath": null, + "getPolicyValues": null, + "inputPath": null, + "metadata": null + } + ] + }, + "groupProperties": {}, + "policyProperties": {} +}
\ No newline at end of file diff --git a/src/onapsdk/sdc/templates/pnf_create.json.j2 b/src/onapsdk/sdc/templates/pnf_create.json.j2 new file mode 100644 index 0000000..fe7e60c --- /dev/null +++ b/src/onapsdk/sdc/templates/pnf_create.json.j2 @@ -0,0 +1,29 @@ +{ + "artifacts": {}, + "attributes": [], + "capabilities": {}, + {% include "sdc_resource_category.json.j2" %}, + "componentInstances": [], + "componentInstancesAttributes": {}, + "componentInstancesProperties": {}, + "componentType": "RESOURCE", + "contactId": "cs0008", + {% if vsp is not none %} + "csarUUID": "{{ vsp.csar_uuid }}", + "csarVersion": "1.0", + "vendorName": "{{ vsp.vendor.name }}", + {% else %} + "vendorName": "{{ vendor.name }}", + {% endif %} + "deploymentArtifacts": {}, + "description": "PNF", + "icon": "defaulticon", + "name": "{{ name }}", + "properties": [], + "groups": [], + "requirements": {}, + "resourceType": "PNF", + "tags": ["{{ name }}"], + "toscaArtifacts": {}, + "vendorRelease": "1.0" +} diff --git a/src/onapsdk/sdc/templates/sdc_element_action.json.j2 b/src/onapsdk/sdc/templates/sdc_element_action.json.j2 new file mode 100644 index 0000000..04fd946 --- /dev/null +++ b/src/onapsdk/sdc/templates/sdc_element_action.json.j2 @@ -0,0 +1,6 @@ +{ +{% if action == const.COMMIT %} + "commitRequest":{"message":"ok"}, +{% endif %} + "action": "{{ action }}" +} diff --git a/src/onapsdk/sdc/templates/sdc_resource_action.json.j2 b/src/onapsdk/sdc/templates/sdc_resource_action.json.j2 new file mode 100644 index 0000000..742d076 --- /dev/null +++ b/src/onapsdk/sdc/templates/sdc_resource_action.json.j2 @@ -0,0 +1,3 @@ +{ + "userRemarks": "{{ action | lower }}" +} diff --git a/src/onapsdk/sdc/templates/sdc_resource_add_deployment_artifact.json.j2 b/src/onapsdk/sdc/templates/sdc_resource_add_deployment_artifact.json.j2 new file mode 100644 index 0000000..290c6d2 --- /dev/null +++ b/src/onapsdk/sdc/templates/sdc_resource_add_deployment_artifact.json.j2 @@ -0,0 +1,8 @@ +{ + "artifactGroupType": "DEPLOYMENT", + "artifactName": "{{artifact_name}}", + "artifactLabel": "{{artifact_label}}", + "artifactType": "{{artifact_type}}", + "description": "test", + "payloadData": "{{b64_artifact}}" +} diff --git a/src/onapsdk/sdc/templates/sdc_resource_add_input.json.j2 b/src/onapsdk/sdc/templates/sdc_resource_add_input.json.j2 new file mode 100644 index 0000000..1964f36 --- /dev/null +++ b/src/onapsdk/sdc/templates/sdc_resource_add_input.json.j2 @@ -0,0 +1,39 @@ +{ + "serviceProperties":{ + "{{ sdc_resource.unique_identifier }}":[ + { + "constraints":null, + "defaultValue":null, + "description":null, + "name":"{{ property.name }}", + "origName":"{{ property.name }}", + "parentUniqueId":"{{ sdc_resource.unique_identifier }}", + "password":false, + "required":false, + "schema":{ + "property":{ + "type":"", + "required":false, + "definition":false, + "description":null, + "password":false + } + }, + "schemaType":"", + "type":"{{ property.property_type }}", + "uniqueId":"{{ sdc_resource.unique_identifier }}.{{ property.name }}", + {% if property.value is not none %} + "value":"{{ property.value }}", + {% else %} + "value":null, + {% endif %} + "definition":false, + "getInputValues":null, + "parentPropertyType":null, + "subPropertyInputPath":null, + "getPolicyValues":null, + "inputPath":null + } + ] + } +}
\ No newline at end of file diff --git a/src/onapsdk/sdc/templates/sdc_resource_add_nested_input.json.j2 b/src/onapsdk/sdc/templates/sdc_resource_add_nested_input.json.j2 new file mode 100644 index 0000000..9dc8261 --- /dev/null +++ b/src/onapsdk/sdc/templates/sdc_resource_add_nested_input.json.j2 @@ -0,0 +1,35 @@ +{ + "componentInstanceInputsMap":{ + "{{ component.unique_id }}":[ + { + {# "defaultValue":null, #} + "name":"{{ input.name }}", + "origName":"{{ input.name }}", + {# "parentUniqueId":"cs0008", #} + "password":false, + "required":false, + "schema":{ + "property":{ + {# "type":"", + "required":false, + "definition":false, + "password":false #} + } + }, + {# "schemaType":"", #} + "type":"{{ input.input_type }}", + "uniqueId":"{{ sdc_resource.unique_identifier }}.{{ input.name }}", + {% if input.default_value is not none %} + "value":"{{ input.default_value }}", + {% endif %} + "definition":false + {# "type":"{{ input.input_type }}", #} + } + ] + }, + "componentInstanceProperties":{ + "{{ component.unique_id }}":[] + }, + "groupProperties":{}, + "policyProperties":{} +}
\ No newline at end of file diff --git a/src/onapsdk/sdc/templates/sdc_resource_add_property.json.j2 b/src/onapsdk/sdc/templates/sdc_resource_add_property.json.j2 new file mode 100644 index 0000000..bed49ca --- /dev/null +++ b/src/onapsdk/sdc/templates/sdc_resource_add_property.json.j2 @@ -0,0 +1,17 @@ +{ + "{{ property.name }}":{ + "schema":{ + "property":{ + "type":"" + } + }, + "name": "{{ property.name }}", + {% if property.description %} + "description": "{{ property.description }}", + {% endif %} + {% if property.value %} + "value": "{{ property.value }}", + {% endif %} + "type": "{{ property.property_type }}" + } +}
\ No newline at end of file diff --git a/src/onapsdk/sdc/templates/sdc_resource_category.json.j2 b/src/onapsdk/sdc/templates/sdc_resource_category.json.j2 new file mode 100644 index 0000000..633aacd --- /dev/null +++ b/src/onapsdk/sdc/templates/sdc_resource_category.json.j2 @@ -0,0 +1,13 @@ + "categories": [ + { + "normalizedName": "{{ category.normalized_name }}", + "name": "{{ category.name }}", + "uniqueId": "{{ category.unique_id }}", + "subcategories": {% if category.subcategories %}{{ category.subcategories|tojson }}{% else %}null{% endif %}, + "version": {% if category.version %}"{{ category.version }}"{% else %}null{% endif %}, + "ownerId": {% if category.owner_id %}"{{ category.owner_id }}"{% else %}null{% endif %}, + "empty": {{ category.empty|tojson }}, + "type": {% if category.type %}"{{ category.type }}"{% else %}null{% endif %}, + "icons": {% if category.icons %}{{ category.icons|tojson }}{% else %}null{% endif %} + } + ]
\ No newline at end of file diff --git a/src/onapsdk/sdc/templates/sdc_resource_component_set_property_value.json.j2 b/src/onapsdk/sdc/templates/sdc_resource_component_set_property_value.json.j2 new file mode 100644 index 0000000..46bd527 --- /dev/null +++ b/src/onapsdk/sdc/templates/sdc_resource_component_set_property_value.json.j2 @@ -0,0 +1,13 @@ +[ + { + "name":"{{ property.name }}", + "parentUniqueId":"{{ component.actual_component_uid }}", + "type":"{{ property.property_type }}", + "uniqueId":"{{ component.actual_component_uid }}.{{ property.name }}", + "value":"{{ value }}", + "definition":false, + "toscaPresentation":{ + "ownerId":"{{ component.actual_component_uid }}" + } + } +]
\ No newline at end of file diff --git a/src/onapsdk/sdc/templates/sdc_resource_set_input_default_value.json.j2 b/src/onapsdk/sdc/templates/sdc_resource_set_input_default_value.json.j2 new file mode 100644 index 0000000..97c2cfd --- /dev/null +++ b/src/onapsdk/sdc/templates/sdc_resource_set_input_default_value.json.j2 @@ -0,0 +1,8 @@ +[ + { + "defaultValue":"{{ default_value }}", + "name":"{{ input.name }}", + "type":"{{ input.input_type }}", + "uniqueId":"{{ input.unique_id }}" + } +]
\ No newline at end of file diff --git a/src/onapsdk/sdc/templates/sdc_resource_set_property_value.json.j2 b/src/onapsdk/sdc/templates/sdc_resource_set_property_value.json.j2 new file mode 100644 index 0000000..d0e73f7 --- /dev/null +++ b/src/onapsdk/sdc/templates/sdc_resource_set_property_value.json.j2 @@ -0,0 +1,13 @@ +[ + { + "name":"{{ property.name }}", + "parentUniqueId":"{{ sdc_resource.unique_identifier }}", + "type":"{{ property.property_type }}", + "uniqueId":"{{ sdc_resource.unique_identifier }}.{{ property.name }}", + "value":"{{ value }}", + "definition":false, + "toscaPresentation":{ + "ownerId":"{{ sdc_resource.unique_identifier }}" + } + } +]
\ No newline at end of file diff --git a/src/onapsdk/sdc/templates/service_create.json.j2 b/src/onapsdk/sdc/templates/service_create.json.j2 new file mode 100644 index 0000000..f247059 --- /dev/null +++ b/src/onapsdk/sdc/templates/service_create.json.j2 @@ -0,0 +1,29 @@ + +{ + "componentType": "SERVICE", + "properties": [], + "requirements": {}, + "toscaArtifacts": {}, + "tags": ["{{ name }}"], + "artifacts": {}, + "description": "service", + "serviceApiArtifacts": {}, + "capabilities": {}, + "name": "{{ name }}", + "componentInstancesProperties": {}, + "componentInstancesAttributes": {}, + "contactId": "cs0008", + "groups": [], + "projectCode": "123456", + "deploymentArtifacts": {}, + "attributes": [], + "componentInstances": [], + "ecompGeneratedNaming": true, + "instantiationType": "{{ instantiation_type }}", + "environmentContext": "General_Revenue-Bearing", + {% include "sdc_resource_category.json.j2" %}, + "icon": "network_l_1-3", + "serviceFunction": "{{ function }}", + "serviceRole": "{{ role }}", + "serviceType": "{{ service_type }}" +} diff --git a/src/onapsdk/sdc/templates/vendor_create.json.j2 b/src/onapsdk/sdc/templates/vendor_create.json.j2 new file mode 100644 index 0000000..858f736 --- /dev/null +++ b/src/onapsdk/sdc/templates/vendor_create.json.j2 @@ -0,0 +1,5 @@ +{ + "iconRef": "icon", + "vendorName": "{{ name }}", + "description": "vendor" +} diff --git a/src/onapsdk/sdc/templates/vf_create.json.j2 b/src/onapsdk/sdc/templates/vf_create.json.j2 new file mode 100644 index 0000000..6f165e5 --- /dev/null +++ b/src/onapsdk/sdc/templates/vf_create.json.j2 @@ -0,0 +1,27 @@ +{ + "artifacts": {}, + "attributes": [], + "capabilities": {}, + {% include "sdc_resource_category.json.j2" %}, + "componentInstances": [], + "componentInstancesAttributes": {}, + "componentInstancesProperties": {}, + "componentType": "RESOURCE", + "contactId": "cs0008", + {% if category.name != "Allotted Resource" %} + "csarUUID": "{{ vsp.csar_uuid }}", + "csarVersion": "1.0", + {% endif %} + "deploymentArtifacts": {}, + "description": "VF", + "icon": "defaulticon", + "name": "{{ name }}", + "properties": [], + "groups": [], + "requirements": {}, + "resourceType": "VF", + "tags": ["{{ name }}"], + "toscaArtifacts": {}, + "vendorName": "{{ vendor.name }}", + "vendorRelease": "1.0" +} diff --git a/src/onapsdk/sdc/templates/vf_vsp_update.json.j2 b/src/onapsdk/sdc/templates/vf_vsp_update.json.j2 new file mode 100644 index 0000000..f862676 --- /dev/null +++ b/src/onapsdk/sdc/templates/vf_vsp_update.json.j2 @@ -0,0 +1,61 @@ +{ + "resourceType": "{{ resource_data['resourceType'] }}", + "componentType": "{{ resource_data['componentType'] }}", + "tags": {{ resource_data['tags'] | tojson }}, + "icon": "{{ resource_data['icon'] }}", + "uniqueId": "{{ resource_data['uniqueId'] }}", + "uuid": "{{ resource_data['uuid'] }}", + "invariantUUID": "{{ resource_data['invariantUUID'] }}", + "contactId": "{{ resource_data['contactId'] }}", + "categories": {{ resource_data['categories'] | tojson }}, + "creatorUserId": "{{ resource_data['creatorUserId'] }}", + "creationDate": {{ resource_data['creationDate'] }}, + "creatorFullName": "{{ resource_data['creatorFullName'] }}", + "description": "{{ resource_data['description'] }}", + "lastUpdateDate": {{ resource_data['lastUpdateDate'] }}, + "lastUpdaterUserId": "{{ resource_data['lastUpdaterUserId'] }}", + "lastUpdaterFullName": "{{ resource_data['lastUpdaterFullName'] }}", + "lifecycleState": "{{ resource_data['lifecycleState'] }}", + "name": "{{ resource_data['name'] }}", + "version": "{{ resource_data['version'] }}", + "allVersions": {{ resource_data['allVersions'] | tojson }}, + "vendorName": "{{ resource_data['vendorName'] }}", + "vendorRelease": "{{ resource_data['vendorRelease'] }}", + "normalizedName": "{{ resource_data['normalizedName'] }}", + "systemName": "{{ resource_data['systemName'] }}", + "archived": {{ resource_data['archived'] | tojson }}, + "componentMetadata": { + "resourceType": "{{ resource_data['resourceType'] }}", + "componentType": "{{ resource_data['componentType'] }}", + "tags": {{ resource_data['tags'] | tojson }}, + "icon": "{{ resource_data['icon'] }}", + "uniqueId": "{{ resource_data['uniqueId'] }}", + "uuid": "{{ resource_data['uuid'] }}", + "invariantUUID": "{{ resource_data['invariantUUID'] }}", + "contactId": "{{ resource_data['contactId'] }}", + "categories": {{ resource_data['categories'] | tojson }}, + "creatorUserId": "{{ resource_data['creatorUserId'] }}", + "creationDate": {{ resource_data['creationDate'] }}, + "creatorFullName": "{{ resource_data['creatorFullName'] }}", + "description": "{{ resource_data['description'] }}", + "lastUpdateDate": {{ resource_data['lastUpdateDate'] }}, + "lastUpdaterUserId": "{{ resource_data['lastUpdaterUserId'] }}", + "lastUpdaterFullName": "{{ resource_data['lastUpdaterFullName'] }}", + "lifecycleState": "{{ resource_data['lifecycleState'] }}", + "name": "{{ resource_data['name'] }}", + "version": "{{ resource_data['version'] }}", + "allVersions": {{ resource_data['allVersions'] | tojson }}, + "vendorName": "{{ resource_data['vendorName'] }}", + "vendorRelease": "{{ resource_data['vendorRelease'] }}", + "normalizedName": "{{ resource_data['normalizedName'] }}", + "systemName": "{{ resource_data['systemName'] }}", + "csarUUID": "{{ resource_data['csarUUID'] }}", + "csarVersion": "{{ resource_data['csarVersion'] }}", + "derivedFrom": null, + "resourceVendorModelNumber": "{{ resource_data['resourceVendorModelNumber'] }}" + }, + "csarUUID": "{{ csarUUID }}", + "csarVersion": "{{ csarVersion }}", + "derivedFrom": null, + "resourceVendorModelNumber": "{{ resource_data['resourceVendorModelNumber'] }}" +}
\ No newline at end of file diff --git a/src/onapsdk/sdc/templates/vsp_create.json.j2 b/src/onapsdk/sdc/templates/vsp_create.json.j2 new file mode 100644 index 0000000..30fa6b9 --- /dev/null +++ b/src/onapsdk/sdc/templates/vsp_create.json.j2 @@ -0,0 +1,11 @@ +{ + "name": "{{ name }}", + "description": "vendor software product", + "icon": "icon", + "category": "resourceNewCategory.generic", + "subCategory": "resourceNewCategory.generic.abstract", + "vendorName": "{{ vendor.name }}", + "vendorId": "{{ vendor.identifier }}", + "licensingData": {}, + "onboardingMethod": "NetworkPackage" +} diff --git a/src/onapsdk/sdc/vendor.py b/src/onapsdk/sdc/vendor.py new file mode 100644 index 0000000..a82e3b9 --- /dev/null +++ b/src/onapsdk/sdc/vendor.py @@ -0,0 +1,108 @@ +"""Vendor 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. +from typing import Any +from typing import Dict + +from onapsdk.sdc.sdc_element import SdcElement +import onapsdk.constants as const +from onapsdk.utils.headers_creator import headers_sdc_creator + + +class Vendor(SdcElement): + """ + ONAP Vendor Object used for SDC operations. + + Attributes: + name (str): the name of the vendor. Defaults to "Generic-Vendor". + identifier (str): the unique ID of the vendor from SDC. + status (str): the status of the vendor from SDC. + version (str): the version ID of the vendor from SDC. + + """ + + VENDOR_PATH = "vendor-license-models" + headers = headers_sdc_creator(SdcElement.headers) + + def __init__(self, name: str = None): + """ + Initialize vendor object. + + Args: + name (optional): the name of the vendor + + """ + super().__init__() + self.name: str = name or "Generic-Vendor" + + def onboard(self) -> None: + """Onboard the vendor in SDC.""" + if not self.status: + self.create() + self.onboard() + elif self.status == const.DRAFT: + self.submit() + + def create(self) -> None: + """Create the vendor in SDC if not already existing.""" + self._create("vendor_create.json.j2", name=self.name) + + def submit(self) -> None: + """Submit the SDC vendor in order to enable it.""" + self._logger.info("attempting to certify/sumbit vendor %s in SDC", + self.name) + if self.status != const.CERTIFIED and self.created(): + self._really_submit() + elif self.status == const.CERTIFIED: + self._logger.warning( + "vendor %s in SDC is already submitted/certified", self.name) + elif not self.created(): + self._logger.warning("vendor %s in SDC is not created", self.name) + + def update_informations_from_sdc(self, details: Dict[str, Any]) -> None: + """ + + Update instance with details from SDC. + + Args: + details (Dict[str, Any]): dict from SDC + + """ + self._status = details['status'] + + @classmethod + def import_from_sdc(cls, values: Dict[str, Any]) -> 'Vendor': + """ + Import Vendor from SDC. + + Args: + values (Dict[str, Any]): dict to parse returned from SDC. + + Returns: + a Vsp instance with right values + + """ + vendor = Vendor(values['name']) + vendor.identifier = values['id'] + return vendor + + def _really_submit(self) -> None: + """Really submit the SDC Vf in order to enable it.""" + self._action_to_sdc(const.SUBMIT) + self._status = const.CERTIFIED + + @classmethod + def _sdc_path(cls) -> None: + """Give back the end of SDC path.""" + return cls.VENDOR_PATH diff --git a/src/onapsdk/sdc/vf.py b/src/onapsdk/sdc/vf.py new file mode 100644 index 0000000..7e4e4a7 --- /dev/null +++ b/src/onapsdk/sdc/vf.py @@ -0,0 +1,164 @@ +"""Vf 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. +from __future__ import annotations +from typing import Any, Dict, List, Optional, TYPE_CHECKING, Union + +from onapsdk.exceptions import ParameterError +from onapsdk.sdc.properties import ComponentProperty, NestedInput, Property +from onapsdk.sdc.sdc_resource import SdcResource +from onapsdk.sdc.vendor import Vendor +from onapsdk.utils.jinja import jinja_env +import onapsdk.constants as const + +if TYPE_CHECKING: + from onapsdk.sdc.vsp import Vsp + + +class Vf(SdcResource): + """ + ONAP Vf Object used for SDC operations. + + Attributes: + name (str): the name of the vendor. Defaults to "Generic-Vendor". + identifier (str): the unique ID of the vendor from SDC. + status (str): the status of the vendor from SDC. + version (str): the version ID of the vendor from SDC. + vsp (Vsp): the Vsp used for VF creation + uuid (str): the UUID of the VF (which is different from identifier, + don't ask why...) + unique_identifier (str): Yet Another ID, just to puzzle us... + + """ + + def __init__(self, name: str = None, version: str = None, sdc_values: Dict[str, str] = None, # pylint: disable=too-many-arguments + vsp: Vsp = None, properties: List[Property] = None, + inputs: Union[Property, NestedInput] = None, + category: str = None, subcategory: str = None, + vendor: Vendor = None): + """ + Initialize vf object. + + Args: + name (optional): the name of the vf + version (str, optional): the version of the vf + vsp (Vsp, optional): VSP object related with the Vf object. + Defaults to None. + vendor (Vendor, optional): Vendor object used if VSP was not provided. + Defaults to None. + + """ + super().__init__(sdc_values=sdc_values, version=version, properties=properties, + inputs=inputs, category=category, subcategory=subcategory) + self.name: str = name or "ONAP-test-VF" + self.vsp: Vsp = vsp + self._vendor: Vendor = vendor + + @property + def vendor(self) -> Optional[Vendor]: + """Vendor related wth Vf. + + If Vf is created vendor is get from it's resource metadata. + Otherwise it's vendor provided by the user or the vendor from vsp. + It's possible that method returns None, but it won't be possible then + to create that Vf resource. + + Returns: + Optional[Vendor]: Vendor object related with Vf + + """ + if self._vendor: + return self._vendor + if self.created(): + resource_data: Dict[str, Any] = self.send_message_json( + "GET", + "Get VF data to update VSP", + self.resource_inputs_url + ) + self._vendor = Vendor(name=resource_data.get("vendorName")) + elif self.vsp: + self._vendor = self.vsp.vendor + return self._vendor + + + def create(self) -> None: + """Create the Vf in SDC if not already existing. + + Raises: + ParameterError: VSP is not provided during VF object initalization + + """ + if not any([self.vsp, self.vendor]): + raise ParameterError("At least vsp or vendor needs to be given") + self._create("vf_create.json.j2", name=self.name, vsp=self.vsp, + category=self.category, vendor=self.vendor) + + def _really_submit(self) -> None: + """Really submit the SDC Vf in order to enable it.""" + self._action_to_sdc(const.CERTIFY, "lifecycleState") + self.load() + + def update_vsp(self, vsp: Vsp) -> None: + """Update Vsp. + + Update VSP UUID and version for Vf object. + + Args: + vsp (Vsp): Object to be used in Vf + + Raises: + ValidationError: Vf object request has invalid structure. + + """ + resource_data: Dict[str, Any] = self.send_message_json( + "GET", + "Get VF data to update VSP", + self.resource_inputs_url + ) + self.send_message_json( + "PUT", + "Update vsp data", + self.resource_inputs_url, + data=jinja_env() + .get_template("vf_vsp_update.json.j2") + .render(resource_data=resource_data, + csarUUID=vsp.csar_uuid, + csarVersion=vsp.human_readable_version) + ) + + def declare_input(self, + input_to_declare: Union[Property, NestedInput, ComponentProperty]) -> None: + """Declare input for given property, nested input or component property object. + + Call SDC FE API to declare input for given property. + + Args: + input_declaration (Union[Property, NestedInput]): Property or ComponentProperty + to declare input or NestedInput object + + Raises: + ParameterError: if the given property is not SDC resource property + + """ + if not isinstance(input_to_declare, ComponentProperty): + super().declare_input(input_to_declare) + else: + self.send_message("POST", + f"Declare new input for {input_to_declare.name} property", + f"{self.resource_inputs_url}/create/inputs", + data=jinja_env().get_template(\ + "component_declare_input.json.j2").\ + render(\ + component=input_to_declare.component, + property=input_to_declare)) diff --git a/src/onapsdk/sdc/vfc.py b/src/onapsdk/sdc/vfc.py new file mode 100644 index 0000000..e2a6f57 --- /dev/null +++ b/src/onapsdk/sdc/vfc.py @@ -0,0 +1,46 @@ +"""VFC 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. +from typing import Dict + +from onapsdk.exceptions import ResourceNotFound + +from .sdc_resource import SdcResource + + +class Vfc(SdcResource): + """ONAP VFC Object used for SDC operations.""" + + def __init__(self, name: str, version: str = None, sdc_values: Dict[str, str] = None): + """Initialize VFC object. + + Vfc has to exist in SDC. + + Args: + name (str): Vfc name + version (str): Vfc version + sdc_values (Dict[str, str], optional): Sdc values of existing Vfc. Defaults to None. + + Raises: + ResourceNotFound: Vfc doesn't exist in SDC + + """ + super().__init__(name=name, version=version, sdc_values=sdc_values) + if not sdc_values and not self.exists(): + raise ResourceNotFound( + "This Vfc has to exist prior to object initialization.") + + def _really_submit(self) -> None: + """Really submit the SDC in order to enable it.""" + raise NotImplementedError("Vfc doesn't need _really_submit") diff --git a/src/onapsdk/sdc/vl.py b/src/onapsdk/sdc/vl.py new file mode 100644 index 0000000..22872f0 --- /dev/null +++ b/src/onapsdk/sdc/vl.py @@ -0,0 +1,46 @@ +"""Vl 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. +from typing import Dict + +from onapsdk.exceptions import ResourceNotFound + +from .sdc_resource import SdcResource + + +class Vl(SdcResource): + """ONAP Vl Object used for SDC operations.""" + + def __init__(self, name: str, version: str = None, sdc_values: Dict[str, str] = None): + """Initialize Vl object. + + Vl has to exists in SDC. + + Args: + name (str): Vl name + version (str): Vl version + sdc_values (Dict[str, str], optional): Sdc values of existing Vl. Defaults to None. + + Raises: + ResourceNotFound: Vl doesn't exist in SDC + + """ + super().__init__(name=name, version=version, sdc_values=sdc_values) + if not sdc_values and not self.exists(): + raise ResourceNotFound( + "This Vl has to exist prior to object initialization.") + + def _really_submit(self) -> None: + """Really submit the SDC in order to enable it.""" + raise NotImplementedError("Vl don't need _really_submit") 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() |