diff options
author | Michal Jagiello <michal.jagiello@t-mobile.pl> | 2024-01-16 12:56:15 +0100 |
---|---|---|
committer | Michal Jagiello <michal.jagiello@t-mobile.pl> | 2024-01-31 15:25:22 +0100 |
commit | bfb53be9956b27b20fcefd54504b7d26db052053 (patch) | |
tree | c8056a96a8fa54e8f13e19c0accaf796134b6ccb /src | |
parent | 0a5d6dbc73d21d597eba1069d343b6e7684e91f9 (diff) |
SDC refactor to use sdc v2 API
It's faster and usage of archive/deletion of resources is possible
Issue-ID: TEST-404
Change-Id: I362c1c282afcd8b4ce3e540bc23a4751861e3b4f
Signed-off-by: Michal Jagiello <michal.jagiello@t-mobile.pl>
Diffstat (limited to 'src')
21 files changed, 2165 insertions, 3 deletions
diff --git a/src/onapsdk/configuration/global_settings.py b/src/onapsdk/configuration/global_settings.py index 1412203..7c09187 100644 --- a/src/onapsdk/configuration/global_settings.py +++ b/src/onapsdk/configuration/global_settings.py @@ -62,7 +62,6 @@ SO_MONITOR_GUI_SERVICE = f"{SO_URL}/" SDC_GUI_SERVICE = f"{SDC_FE_URL}/sdc1/portal" SDNC_DG_GUI_SERVICE = f"{SDNC_URL}/nifi/" SDNC_ODL_GUI_SERVICE = f"{SDNC_URL}/odlux/index.html" - DCAEMOD_GUI_SERVICE = f"{DCAEMOD_URL}/" HOLMES_GUI_SERVICE = f"{HOLMES_URL}/iui/holmes/default.html" POLICY_GUI_SERVICE = f"{POLICY_URL}/onap/login.html" @@ -74,3 +73,12 @@ LOB = "Onapsdk_lob" PLATFORM = "Onapsdk_platform" DEFAULT_REQUEST_TIMEOUT = 60 + +# SDC DISTRIBUTION +SDC_SERVICE_DISTRIBUTION_COMPONENTS = [ + "SO-sdc-controller", + "aai-model-loader", + "sdnc-sdc-listener", + "policy-distribution-id", + "multicloud-k8s" +] diff --git a/src/onapsdk/sdc2/__init__.py b/src/onapsdk/sdc2/__init__.py new file mode 100644 index 0000000..39c2a8e --- /dev/null +++ b/src/onapsdk/sdc2/__init__.py @@ -0,0 +1 @@ +"""SDC v2 package.""" diff --git a/src/onapsdk/sdc2/component_instance.py b/src/onapsdk/sdc2/component_instance.py new file mode 100644 index 0000000..aba1afa --- /dev/null +++ b/src/onapsdk/sdc2/component_instance.py @@ -0,0 +1,282 @@ +"""Component instance module.""" +# Copyright 2024 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 TYPE_CHECKING, Any, Dict, Optional, Iterable +from urllib.parse import urljoin + +from onapsdk.sdc2.sdc import SDC +from onapsdk.utils.jinja import jinja_env # type: ignore + +if TYPE_CHECKING: + from onapsdk.sdc2.sdc_resource import SDCResource + + +class ComponentInstanceInput(SDC): # pylint: disable=too-many-instance-attributes + """Component instance input class.""" + + SET_INPUT_VALUE_TEMPLATE = "sdc2_component_instance_input_set_value.json.j2" + + def __init__(self, # pylint: disable=too-many-locals too-many-arguments + component_instance: "ComponentInstance", + definition: bool, + hidden: bool, + unique_id: str, + input_type: str, + required: bool, + password: bool, + name: str, + immutable: bool, + mapped_to_component_property: bool, + is_declared_list_input: bool, + user_created: bool, + get_input_property: bool, + empty: bool, + label: Optional[str] = None, + description: Optional[str] = None, + value: Optional[Any] = None) -> None: + """Component instance input initialisation. + + Args: + component_instance (ComponentInstance): Component instance + definition (bool): Input definiton + hidden (bool): Flag which determines if input is hidden + unique_id (str): Input unique ID + input_type (str): Input type + required (bool): Flag which determies if input is required + password (bool): Flag which determines if input is password type + name (str): Input's name + immutable (bool): Flag which determines if input is immutable + mapped_to_component_property (bool): Flag which determines if input is + mapped to component property + is_declared_list_input (bool): Flag which determines if input is declared list + user_created (bool): Flag which determines if was created by user + get_input_property (bool): Flag which determines if it's get input property + empty (bool): Flag which determines if input is empty + label (Optional[str], optional): Input's label. Defaults to None. + description (Optional[str], optional): Input's description. Defaults to None. + value (Optional[Any], optional): Input's value. Defaults to None. + + """ + super().__init__(name=name) + self.component_instance: "ComponentInstance" = component_instance + self.definition: bool = definition + self.hidden: bool = hidden + self.unique_id: str = unique_id + self.input_type: str = input_type + self.required: bool = required + self.password: bool = password + self.immutable: bool = immutable + self.mapped_to_component_property: bool = mapped_to_component_property + self.is_declared_list_input: bool = is_declared_list_input + self.user_created: bool = user_created + self.get_input_property: bool = get_input_property + self.empty: bool = empty + self.description: Optional[str] = description + self.label: Optional[str] = label + self._value: Optional[Any] = value + + @classmethod + def create_from_api_response(cls, + api_response: Dict[str, Any], + component_instance: "ComponentInstance" + ) -> "ComponentInstanceInput": + """Create instance input using values dict returned by SDC API. + + Args: + api_response (Dict[str, Any]): Values dictionary + component_instance (ComponentInstance): Component instance related with an input + + Returns: + ComponentInstanceInput: Component instance input object + + """ + return cls( + component_instance=component_instance, + definition=api_response["definition"], + hidden=api_response["hidden"], + unique_id=api_response["uniqueId"], + input_type=api_response["type"], + required=api_response["required"], + password=api_response["password"], + name=api_response["name"], + immutable=api_response["immutable"], + mapped_to_component_property=api_response["mappedToComponentProperty"], + is_declared_list_input=api_response["isDeclaredListInput"], + user_created=api_response["userCreated"], + get_input_property=api_response["getInputProperty"], + empty=api_response["empty"], + value=api_response.get("value"), + label=api_response.get("label"), + description=api_response.get("description") + ) + + @property + def value(self) -> Optional[Any]: + """Component instance input value. + + Returns: + Optional[Any]: Value (if any) of input + + """ + return self._value + + @value.setter + def value(self, value: Any) -> None: + """Component instance's input value setter. + + Call an API to set a value of component instances' input + + Args: + value (Any): Any value which is going to be set + + """ + self.send_message_json( + "POST", + f"Set value of {self.component_instance.sdc_resource.name} resource input {self.name}", + urljoin(self.base_back_url, + (f"sdc2/rest/v1/catalog/{self.component_instance.sdc_resource.catalog_type()}/" + f"{self.component_instance.sdc_resource.unique_id}/resourceInstance/" + f"{self.component_instance.unique_id}/inputs")), + data=jinja_env().get_template(self.SET_INPUT_VALUE_TEMPLATE).render( + component_instance_input=self, + value=value + ) + ) + self._value = value + + +class ComponentInstance(SDC): # pylint: disable=too-many-instance-attributes + """Component instance class.""" + + def __init__(self, # pylint: disable=too-many-locals too-many-arguments + actual_component_uid: str, + component_name: str, + component_uid: str, + component_version: str, + creation_time: int, + customization_uuid: str, + icon: str, + invariant_name: str, + is_proxy: bool, + modification_time: int, + name: str, + normalized_name: str, + origin_type: str, + tosca_component_name: str, + unique_id: str, + sdc_resource: "SDCResource") -> None: + """Component instance initialise. + + Args: + actual_component_uid (str): Component actual UID + component_name (str): Component name + component_uid (str): Component UID + component_version (str): Component version + creation_time (int): Creation timestamp + customization_uuid (str): Customization UUID + icon (str): Icon + invariant_name (str): Invariant name + is_proxy (bool): Flag determines if component is proxy + modification_time (int): Modification timestamp + name (str): Component name + normalized_name (str): Component normalized name + origin_type (str): Component origin type + tosca_component_name (str): Component's TOSCA name + unique_id (str): Unique ID + sdc_resource (SDCResource): Components SDC resource + + """ + super().__init__(name=name) + self.actual_component_uid: str = actual_component_uid + self.component_name: str = component_name + self.component_uid: str = component_uid + self.component_version: str = component_version + self.creation_time: int = creation_time + self.customization_uuid: str = customization_uuid + self.icon: str = icon + self.invariant_name: str = invariant_name + self.is_proxy: bool = is_proxy + self.modification_time: int = modification_time + self.normalized_name: str = normalized_name + self.origin_type: str = origin_type + self.tosca_component_name: str = tosca_component_name + self.unique_id: str = unique_id + self.sdc_resource: "SDCResource" = sdc_resource + + @classmethod + def create_from_api_response(cls, + data: Dict[str, Any], + sdc_resource: "SDCResource") -> "ComponentInstance": + """Create components insance from API response. + + Args: + data (Dict[str, Any]): API response values dictionary + sdc_resource (SDCResource): SDC resource with which component instance is related with + + Returns: + ComponentInstance: Component instance object + + """ + return cls( + sdc_resource=sdc_resource, + actual_component_uid=data["actualComponentUid"], + component_name=data["componentName"], + component_uid=data["componentUid"], + component_version=data["componentVersion"], + creation_time=data["creationTime"], + customization_uuid=data["customizationUUID"], + icon=data["icon"], + invariant_name=data["invariantName"], + is_proxy=data["isProxy"], + modification_time=data["modificationTime"], + name=data["name"], + normalized_name=data["normalizedName"], + origin_type=data["originType"], + tosca_component_name=data["toscaComponentName"], + unique_id=data["uniqueId"] + ) + + @property + def inputs(self) -> Iterable[ComponentInstanceInput]: + """Component instance's inputs iterator. + + Yields: + ComponentInstanceInput: Component's instance input object + + """ + for input_data in self.send_message_json( + "GET", + "Get inputs", + urljoin(self.base_back_url, + (f"sdc2/rest/v1/catalog/{self.sdc_resource.catalog_type()}/" + f"{self.sdc_resource.unique_id}/componentInstances/{self.unique_id}/" + f"{self.actual_component_uid}/inputs")) + ): + yield ComponentInstanceInput.create_from_api_response(input_data, self) + + def get_input_by_name(self, input_name: str) -> Optional[ComponentInstanceInput]: + """Get component's input by it's name. + + Args: + input_name (str): Input name + + Returns: + Optional[ComponentInstanceInput]: Input with given name, + None if no input with given name found + + """ + for component_instance_input in self.inputs: + if component_instance_input.name == input_name: + return component_instance_input + return None diff --git a/src/onapsdk/sdc2/pnf.py b/src/onapsdk/sdc2/pnf.py new file mode 100644 index 0000000..9e1ff31 --- /dev/null +++ b/src/onapsdk/sdc2/pnf.py @@ -0,0 +1,40 @@ +"""SDC PNF module.""" +# Copyright 2024 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 onapsdk.sdc2.sdc import ResoureTypeEnum +from onapsdk.sdc2.sdc_resource import SDCResourceTypeObject, SDCResourceTypeObjectCreateMixin + + +class Pnf(SDCResourceTypeObject, SDCResourceTypeObjectCreateMixin): # pylint: disable=too-many-ancestors + """PNF class.""" + + @classmethod + def resource_type(cls) -> ResoureTypeEnum: + """PNF resource type. + + Returns: + ResoureTypeEnum: PNF resource type enum value + + """ + return ResoureTypeEnum.PNF + + @classmethod + def create_payload_template(cls) -> str: + """Get a template to create PNF creation request payload. + + Returns: + str: PNF creation template. + + """ + return "sdc2_create_pnf.json.j2" diff --git a/src/onapsdk/sdc2/sdc.py b/src/onapsdk/sdc2/sdc.py new file mode 100644 index 0000000..e4a4120 --- /dev/null +++ b/src/onapsdk/sdc2/sdc.py @@ -0,0 +1,103 @@ +"""Base SDC module.""" +# Copyright 2024 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 onapsdk.sdc2.sdc import ResoureTypeEnum +from abc import abstractmethod, ABC +from enum import Enum +from typing import Any, Iterator, List + +from onapsdk.configuration import settings # type: ignore +from onapsdk.onap_service import OnapService # type: ignore +from onapsdk.utils.headers_creator import headers_sdc_creator # type: ignore + + +class ResoureTypeEnum(Enum): # pylint: disable=too-few-public-methods + """Resource types enumerator.""" + + PRODUCT = "PRODUCT" + SERVICE = "SERVICE" + VF = "VF" + VFC = "VFC" + CP = "CP" + VL = "VL" + CONFIGURATION = "Configuration" + VFCMT = "VFCMT" + CVFC = "CVFC" + PNF = "PNF" + CR = "CR" + SERVICE_PROXY = "ServiceProxy" + SERVICE_SUBSTITUTION = "ServiceSubstitution" + + @classmethod + def iter_without_resource_type( + cls, + resource_type_to_exclude: "ResoureTypeEnum" + ) -> Iterator["ResoureTypeEnum"]: + """Return an iterator with resource types but one given as a parameter. + + Yields: + ResoureTypeEnum: Resource types without a one given as a parameter + + """ + resources_type_list: List[ResoureTypeEnum] = list(cls) + resources_type_list.pop(resources_type_list.index(resource_type_to_exclude)) + yield from resources_type_list + + +class SDC(OnapService, ABC): + """Base SDC abstracl class.""" + + base_back_url = settings.SDC_BE_URL + SCREEN_ENDPOINT = "sdc2/rest/v1/screen" + ARCHIVE_ENDPOINT = "sdc2/rest/v1/catalog/archive" + headers = headers_sdc_creator(OnapService.headers) + + def __init__(self, name: str) -> None: + """Init SDC object. + + Each SDC object has a name. + + Args: + name (str): Name + + """ + self.name = 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 + + +class SDCCatalog(SDC, ABC): + """SDC Catalog abstract class.""" + + @classmethod + @abstractmethod + def get_all(cls) -> List["SDCCatalog"]: + """Get all SDCCatalog objects. + + That's abstract class for each class which would implement SDC catalog API + (VF, Service etc.) + + """ diff --git a/src/onapsdk/sdc2/sdc_category.py b/src/onapsdk/sdc2/sdc_category.py new file mode 100644 index 0000000..12c6544 --- /dev/null +++ b/src/onapsdk/sdc2/sdc_category.py @@ -0,0 +1,257 @@ +"""SDC category module.""" +# Copyright 2024 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 onapsdk.sdc2.sdc import ResoureTypeEnum +from abc import ABC, abstractmethod +from dataclasses import dataclass +from typing import Any, Dict, Iterable, List, Optional +from urllib.parse import urljoin + +from onapsdk.exceptions import ResourceNotFound # type: ignore +from onapsdk.sdc2.sdc import SDCCatalog + + +@dataclass +class SdcSubCategory: + """SDC subcategory dataclass.""" + + name: str + normalized_name: str + unique_id: str + + +class SdcCategory(SDCCatalog, ABC): # pylint: disable=too-many-instance-attributes + """SDC category class.""" + + def __init__(self, # pylint: disable=too-many-arguments + name: str, + empty: bool, + icons: List[str], + models: List[str], + normalized_name: str, + unique_id: str, + use_service_substitution_for_nested_services: bool, + owner_id: Optional[str] = None, + subcategories: Optional[List[SdcSubCategory]] = None, + category_type: Optional[str] = None, + version: Optional[str] = None, + display_name: Optional[str] = None) -> None: + """Initialise SDC category object. + + Args: + empty (bool): Falg to determine if category is empty. + icons (List[str]): List of icons. + models (List[str]): List of models. + normalized_name (str): Category normalized name + unique_id (str): Category unique ID + use_service_substitution_for_nested_services (bool): Use service substitution + for nested services + owner_id (Optional[str], optional): Owner ID. Defaults to None. + subcategories (Optional[List[SdcSubCategory]], optional): Optional subcategories list. + Defaults to None. + category_type (Optional[str], optional): Category type. Defaults to None. + version (Optional[str], optional): Version. Defaults to None. + display_name (Optional[str], optional): Display name. Defaults to None. + """ + super().__init__(name) + self.empty: bool = empty + self.icons: List[str] = icons + self.models: List[str] = models + self.normalized_name: str = normalized_name + self.unique_id: str = unique_id + self.use_service_substitution_for_nested_services: bool = \ + use_service_substitution_for_nested_services + self.owner_id: Optional[str] = owner_id + self.subcategories: Optional[List[SdcSubCategory]] = subcategories + self.category_type: Optional[str] = category_type + self.version: Optional[str] = version + self.display_name: Optional[str] = display_name + + def __repr__(self) -> str: + """SDC resource description. + + Returns: + str: SDC resource object description + + """ + return f"{self.__class__.__name__.upper()}(name={self.name})" + + @classmethod + def get_all(cls) -> Iterable["SdcCategory"]: + """Get all categories objects. + + Yields: + SdcCategory: SDC category object + + """ + yield from (cls.create_from_api_response(response_obj) \ + for response_obj in cls.send_message_json( + "GET", + f"Get all {cls.__name__}", + cls.get_all_endpoint() + )) + + @classmethod + def get_by_uniqe_id(cls, unique_id: str) -> "SdcCategory": + """Get category by it's unique ID. + + Args: + unique_id (str): Unique ID of a category + + Raises: + ResourceNotFound: Category with given unique ID does not exist. + + Returns: + SdcCategory: SDC category with given ID + + """ + for category in cls.get_all(): + if category.unique_id == unique_id: + return category + raise ResourceNotFound(f"{cls.__name__} with unique id {unique_id} not found") + + @classmethod + def get_by_name(cls, name: str) -> "SdcCategory": + """Get category by name. + + Args: + name (str): Category name + + Raises: + ResourceNotFound: Category with given name does not exist. + + Returns: + SdcCategory: SDC category with given name + + """ + for category in cls.get_all(): + if category.name == name: + return category + raise ResourceNotFound(f"{cls.__name__} with name {name} not found") + + @classmethod + def get_all_endpoint(cls) -> str: + """Get an endpoint which is going to be used to get all categories. + + It's going to be created using `_endpoint_suffix` and common part for all categories + + Returns: + str: Endpoint to be used to get all categories. + + """ + return urljoin(cls.base_back_url, + urljoin("sdc2/rest/v1/categories/", + cls._endpoint_suffix())) + + @classmethod + @abstractmethod + def _endpoint_suffix(cls) -> str: + """Category endpoint suffix. + + Abstract classmethod + + Returns: + str: Category API endpoint suffix. + + """ + + @classmethod + def create_from_api_response(cls, api_response: Dict[str, Any]) -> "SdcCategory": + """Create category object using an API response dictionary. + + Args: + api_response (Dict[str, Any]): API response dictionary with values + + Returns: + SdcCategory: SDC category object + + """ + return cls( + name=api_response["name"], + empty=api_response["empty"], + icons=api_response["icons"], + models=api_response["models"], + normalized_name=api_response["normalizedName"], + unique_id=api_response["uniqueId"], + use_service_substitution_for_nested_services=\ + api_response["useServiceSubstitutionForNestedServices"], + owner_id=api_response["ownerId"], + subcategories=[SdcSubCategory(name=subcategory["name"], + normalized_name=subcategory["normalizedName"], + unique_id=subcategory["uniqueId"]) for subcategory + in api_response["subcategories"]] + if api_response.get("subcategories") else None, + category_type=api_response["type"], + version=api_response["version"], + display_name=api_response["displayName"], + ) + + def get_subcategory(self, subcategory_name: str) -> Optional[SdcSubCategory]: + """Get category's subcategory by it's name. + + Args: + subcategory_name (str): Subcategory name + + Returns: + SdcSubCategory|None: Subcategory object or None if no subcategory + with given name was found + + """ + if not self.subcategories: + return None + for subcategory in self.subcategories: + if subcategory.name == subcategory_name: + return subcategory + return None + + +class ServiceCategory(SdcCategory): + """Service category class.""" + + @classmethod + def _endpoint_suffix(cls) -> str: + """Service category endpoint suffix. + + Returns: + str: Product category endpoint suffix "services". + + """ + return "services" + + +class ResourceCategory(SdcCategory): + """Resource category class.""" + + @classmethod + def _endpoint_suffix(cls) -> str: + """Resource category endpoint suffix. + + Returns: + str: Product category endpoint suffix "resources". + + """ + return "resources" + + +class ProductCategory(SdcCategory): + """Product category class.""" + + @classmethod + def _endpoint_suffix(cls) -> str: + """Product category endpoint suffix. + + Returns: + str: Product category endpoint suffix "products". + + """ + return "products" diff --git a/src/onapsdk/sdc2/sdc_resource.py b/src/onapsdk/sdc2/sdc_resource.py new file mode 100644 index 0000000..16f8ceb --- /dev/null +++ b/src/onapsdk/sdc2/sdc_resource.py @@ -0,0 +1,709 @@ +"""SDC resource module.""" +# Copyright 2024 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 onapsdk.sdc2.sdc import ResoureTypeEnum +from abc import ABC, abstractmethod +from base64 import b64encode +from enum import Enum, auto +from itertools import chain +from typing import Any, Dict, Iterable, Sequence, Optional +from urllib.parse import urljoin + +from onapsdk.exceptions import ResourceNotFound # type: ignore +from onapsdk.sdc2.component_instance import ComponentInstance +from onapsdk.sdc2.sdc import SDC, ResoureTypeEnum, SDCCatalog +from onapsdk.sdc2.sdc_user import SdcUser +from onapsdk.utils.headers_creator import headers_sdc_artifact_upload # type: ignore +from onapsdk.utils.jinja import jinja_env # type: ignore +from onapsdk.sdc2.sdc_category import ResourceCategory, SdcSubCategory +from onapsdk.sdc.vendor import Vendor # type: ignore +from onapsdk.sdc.vsp import Vsp # type: ignore + + +class LifecycleOperation(Enum): # pylint: disable=too-few-public-methods + """Resources lifecycle operations enum.""" + + CHECKOUT = "checkout" + UNDO_CHECKOUT = "undoCheckout" + CHECKIN = "checkin" + CERIFICATION_REQUEST = "certificationRequest" + START_CERTIFICATION = "startCertification" + FAIL_CERTIFICATION = "failCertification" + CANCEL_CERIFICATION = "cancelCertification" + CERTIFY = "certify" + + +class LifecycleState(Enum): # pylint: disable=too-few-public-methods + """Resources lifecycle states enum.""" + + def _generate_next_value_(name, *_, **__): # pylint: disable=no-self-argument + """Return the upper-cased version of the member name.""" + return name.upper() # pylint: disable=no-member + + READY_FOR_CERTIFICATION = auto() + CERTIFICATION_IN_PROGRESS = auto() + CERTIFIED = auto() + NOT_CERTIFIED_CHECKIN = auto() + NOT_CERTIFIED_CHECKOUT = auto() + + +class SDCResource(SDCCatalog): # pylint: disable=too-many-instance-attributes + """SDC resource class.""" + + LIFECYCLE_OPERATION_TEMPLATE = "sdc2_resource_action.json.j2" + + def __init__(self, # pylint: disable=too-many-locals too-many-arguments + *, + name: str, + version: Optional[str] = None, + archived: Optional[bool] = None, + component_type: Optional[str] = None, + icon: Optional[str] = None, + unique_id: Optional[str] = None, + lifecycle_state: Optional[LifecycleState] = None, + last_update_date: Optional[int] = None, + uuid: Optional[str] = None, + invariant_uuid: Optional[str] = None, + system_name: Optional[str] = None, + tags: Optional[Sequence[str]] = None, + last_updater_user_id: Optional[str] = None, + creation_date: Optional[int] = None, + description: Optional[str] = None, + all_versions: Optional[Dict[str, Any]] = None) -> None: + """SDC resource initialisation. + + Args: + name (str): Resource name + version (Optional[str], optional): Resource version.. Defaults to None. + archived (Optional[bool], optional): Flag determines if resource object is archived. + Defaults to None. + component_type (Optional[str], optional): Component type. Defaults to None. + icon (Optional[str], optional): Resource icon. Defaults to None. + unique_id (Optional[str], optional): Resource unique ID. Defaults to None. + lifecycle_state (Optional[LifecycleState], optional): Resoure lifecycle state. + Defaults to None. + last_update_date (Optional[int], optional): Resource last update timestamp. + Defaults to None. + uuid (Optional[str], optional): Resource UUID. Defaults to None. + invariant_uuid (Optional[str], optional): Resource invariant UUID. Defaults to None. + system_name (Optional[str], optional): Resource system name. Defaults to None. + tags (Optional[Sequence[str]], optional): Resource tags. Defaults to None. + last_updater_user_id (Optional[str], optional): Resource last updater user ID. + Defaults to None. + creation_date (Optional[int], optional): Resource creation timestamp. Defaults to None. + description (Optional[str], optional): Resource description. Defaults to None. + all_versions (Optional[Dict[str, Any]], optional): Dictionary with all resources + versions. Defaults to None + + """ + super().__init__(name) + self.version: Optional[str] = version + self.archived: Optional[bool] = archived + self.component_type: Optional[str] = component_type + self.icon: Optional[str] = icon + self.unique_id: Optional[str] = unique_id + self.lifecycle_state: Optional[LifecycleState] = \ + LifecycleState(lifecycle_state) if lifecycle_state else None + self.last_update_date: Optional[int] = last_update_date + self.uuid: Optional[str] = uuid + self.invariant_uuid: Optional[str] = invariant_uuid + self.system_name: Optional[str] = system_name + self.tags: Optional[Sequence[str]] = tags + self.last_updater_user_id: Optional[str] = last_updater_user_id + self.creation_data: Optional[int] = creation_date + self.description: Optional[str] = description + self.all_versions: Optional[Dict[str, Any]] = all_versions + + def __repr__(self) -> str: + """SDC resource description. + + Returns: + str: SDC resource object description + + """ + return f"{self.__class__.__name__.upper()}(name={self.name})" + + def _copy_object(self, obj: 'SDCCatalog') -> None: + """ + Copy relevant properties from object. + + Args: + obj (Sdc): the object to "copy" + + Raises: + NotImplementedError: this is an abstract method. + + """ + self.__dict__ = obj.__dict__.copy() + + @classmethod + def get_by_name(cls, name: str) -> "SDCResource": + """Get resource by name. + + Filter all objects (archived and active) and get a latest one (with highest version) + which name is equal to one we are looking for. + + Args: + name (str): Name of a resource + + Raises: + ResourceNotFound: Resource with given name not found. + + Returns: + SDCResource: Resource with given name + + """ + try: + searched_rough_data: Dict[str, Any] = next( + iter( + sorted( + filter( + lambda obj: obj["name"] == name, + cls._get_all_rough()), + key=lambda obj: obj["version"], + reverse=True) + ) + ) + return cls.get_by_name_and_version( + searched_rough_data["name"], + searched_rough_data["version"] + ) + except StopIteration as exc: + cls._logger.warning("%s %s doesn't exist in SDC", cls.__name__, name) + raise ResourceNotFound from exc + + def delete(self) -> None: + """Delete resource.""" + self.send_message( + "DELETE", + f"Delete {self.name} {self.__class__.__name__}", + urljoin(self.base_back_url, + f"sdc2/rest/v1/catalog/{self.catalog_type()}/{self.unique_id}") + ) + + def archive(self) -> None: + """Archive resource.""" + self.send_message( + "POST", + f"Archive {self.name} {self.__class__.__name__}", + urljoin(self.base_back_url, + f"sdc2/rest/v1/catalog/{self.catalog_type()}/{self.unique_id}/archive") + ) + + @classmethod + def _get_all_rough(cls) -> Iterable[Dict[str, Any]]: + """Get all resources values dictionaries. + + Returns: + Iterable[Dict[str, Any]]: API responses dictionaries of + both active and archived resources + + """ + return chain(cls._get_active_rough(), cls._get_archived_rough()) + + @classmethod + def get_all(cls) -> Iterable["SDCResource"]: + """Get all resources iterator. + + Yields: + SDCResource: SDC resource + + """ + for rough_data in cls._get_all_rough(): + yield cls.get_by_name_and_version(rough_data["name"], rough_data["version"]) + + @classmethod + def _get_active_rough(cls) -> Iterable[Dict[str, Any]]: + """Get all active resources values dictionaries. + + Yields: + Dict[str, Any]: API response dictionary with values of active resource + + """ + yield from cls.filter_response_objects_by_resource_type(cls.send_message_json( + "GET", + f"Get all {cls.__name__.upper()}", + urljoin(cls.base_back_url, + f"{cls.SCREEN_ENDPOINT}?{cls._build_exclude_types_query(cls.resource_type())}"), + )) + + @classmethod + def _get_archived_rough(cls) -> Iterable[Dict[str, Any]]: + """Get all archived resources values dictionaries. + + Yields: + Dict[str, Any]: API response dictionary with values of archived resource + + """ + yield from cls.filter_response_objects_by_resource_type(cls.send_message_json( + "GET", + f"Get archived {cls.__name__.upper()}", + urljoin(cls.base_back_url, cls.ARCHIVE_ENDPOINT), + )) + + @classmethod + def _build_exclude_types_query(cls, type_to_not_exclude: ResoureTypeEnum) -> str: + """Build query to exclude resource types from API request. + + There is no API to get a specific resource type from SDC backend, but it is possible + to exclude some. So that method creates an HTTP query to exclude all types + but not a one passed as a parameter. + + Args: + type_to_not_exclude (ResoureTypeEnum): Type which won't be excluded from API + request + + Returns: + str: HTTP query + + """ + return "&".join([f"excludeTypes={exclude_type}" for exclude_type in + ResoureTypeEnum.iter_without_resource_type(type_to_not_exclude)]) + + @classmethod + @abstractmethod + def filter_response_objects_by_resource_type( + cls, + response: Dict[str, Any] + ) -> Iterable[Dict[str, Any]]: + """Filter API response by resource type. + + An abstract method which has to be implemented by subclass + + Args: + response (Dict[str, Any]): API response to filter + + Yields: + Dict[str, Any]: Filtered API response values dictionary + + """ + + @classmethod + @abstractmethod + def resource_type(cls) -> ResoureTypeEnum: + """Resource object type. + + Abstract classmethod to be implemented. + + Returns: + ResoureTypeEnum: Resource type + + """ + + @classmethod + @abstractmethod + def create_from_api_response(cls, _: Dict[str, Any]) -> SDCCatalog: + """Create resource with values from API response. + + Abstract classmethod to be implemented. + + Returns: + SDCCatalog: Resource object. + + """ + + @classmethod + def catalog_type(cls) -> str: + """Resource catalog type. + + Resources can have two catalog types: + - services + - resources + That method returns a proper one for given resource type object class. + + Returns: + str: Resource catalog type, "services" or "resources" + """ + if cls.resource_type() == ResoureTypeEnum.SERVICE: + return "services" + return "resources" + + @classmethod + def get_by_name_and_version_endpoint(cls, name: str, version: str) -> str: + """Get an endpoint to be used to get resource by name and it's version. + + Args: + name (str): Resource name + version (str): Resource version + + Returns: + str: An endpoint to be used to get resource object by name and version + + """ + return f"sdc2/rest/v1/catalog/resources/resourceName/{name}/resourceVersion/{version}" + + @classmethod + def add_deployment_artifact_endpoint(cls, object_id: str) -> str: + """Get an endpoint to add deployment artifact into object. + + Args: + object_id (str): Object/resource ID + + Returns: + str: An endpoint to be used to send request to add a deployment artifact into resource + + """ + return f"sdc2/rest/v1/catalog/resources/{object_id}/artifacts" + + @classmethod + def get_by_name_and_version(cls, name: str, version: str) -> "SDCResource": + """Get resource by name and version. + + Args: + name (str): Resource name + version (str): Resource version + + Returns: + SDCResource: SDC resource object with given name and version + + Raises: + ResourceNotFoundError: resource with given name and version + does not exist + + """ + return cls.create_from_api_response(cls.send_message_json( + "GET", + f"Get {cls.__name__} by name and version", + urljoin(cls.base_back_url, cls.get_by_name_and_version_endpoint(name, version)) + )) + + def update(self, api_response: Dict[str, Any]) -> None: + """Update resource with values from API response dictionary. + + Args: + api_response (Dict[str, Any]): API response dictionary with values + used for object update + + """ + self.unique_id = api_response["uniqueId"] + self.uuid = api_response["uuid"] + self.invariant_uuid=api_response["invariantUUID"] + self.version = api_response["version"] + self.last_update_date = api_response["lastUpdateDate"] + self.lifecycle_state = api_response["lifecycleState"] + self.last_updater_user_id = api_response["lastUpdaterUserId"] + self.all_versions = api_response["allVersions"] + + def lifecycle_operation(self, lifecycle_operation: LifecycleOperation) -> None: + """Request lifecycle operation on an object. + + Args: + lifecycle_operation (LifecycleOperation): Lifecycle operation to be requested. + + """ + response: Dict[str, Any] = self.send_message_json( + "POST", + (f"Request lifecycle operation {lifecycle_operation} on " + f"{self.__class__.__name__.upper()} object {self.name}"), + urljoin(self.base_back_url, + (f"sdc2/rest/v1/catalog/{self.catalog_type()}/" + f"{self.unique_id}/lifecycleState/{lifecycle_operation}")), + data=jinja_env().get_template( + self.LIFECYCLE_OPERATION_TEMPLATE).render( + lifecycle_operation=lifecycle_operation) + ) + self.update(response) + + def add_deployment_artifact(self, # pylint: disable=too-many-arguments + artifact_type: str, + artifact_label: str, + artifact_name: str, + artifact_file_path: str, + artifact_group_type: str = "DEPLOYMENT", + artifact_description: str = "ONAP SDK ARTIFACT"): + """ + 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 + + """ + self._logger.debug("Add deployment artifact to %s %s", + self.__class__.__name__.upper(), + self.name) + with open(artifact_file_path, 'rb') as artifact_file: + data: bytes = artifact_file.read() + artifact_upload_payload = jinja_env().get_template( + "sdc2_add_deployment_artifact.json.j2").\ + render(artifact_group_type=artifact_group_type, + artifact_description=artifact_description, + artifact_name=artifact_name, + artifact_label=artifact_label, + artifact_type=artifact_type, + artifact_payload=b64encode(data).decode('utf-8')) + + self.send_message_json("POST", + ("Add deployment artifact to " + f"{self.__class__.__name__.upper()} {self.name}"), + urljoin(self.base_back_url, + self.add_deployment_artifact_endpoint(self.unique_id)), + data=artifact_upload_payload, + headers=headers_sdc_artifact_upload(base_header=self.headers, + data=artifact_upload_payload)) + + @property + def component_instances(self) -> Iterable["ComponentInstance"]: + """Iterate through resource's component instances. + + Yields: + ComponentInstance: Resource's component instance + + """ + for component_instance_dict in self.send_message_json( + "GET", + f"Get {self.__class__.__name__} component instances", + urljoin(self.base_back_url, + (f"sdc2/rest/v1/catalog/{self.catalog_type()}/" + f"{self.unique_id}/componentInstances")) + ): + yield ComponentInstance.create_from_api_response(component_instance_dict, self) + + def get_component_by_name(self, name: str) -> Optional[ComponentInstance]: + """Get resource's component instance by it's name. + + Args: + name (str): Component instance's name + + Returns: + Optional[ComponentInstance]: Component instance with given name, + None if no component instance has given name + + """ + for component_instance in self.component_instances: + if component_instance.component_name == name: + return component_instance + return None + + +class SDCResourceCreateMixin(ABC): # pylint: disable=too-few-public-methods + """SDC resource object creation mixin class. + + Not all resource object can be created (VL can't) so that's why that mixin was created. + Object which inherits from that class has to implement: + - send_message_json + - create_from_api_response + - get_create_payload + methods. + """ + + CREATE_ENDPOINT = urljoin(SDC.base_back_url, "sdc2/rest/v1/catalog/resources") + + @classmethod + def create(cls, + name: str, + *, + user: Optional[SdcUser] = None, + description: Optional[str] = None, + **kwargs: Dict[Any, Any]) -> "SDCResource": + """Create object. + + Args: + name (str): Name of an object + user (Optional[SdcUser], optional): Object creator user ID. Defaults to None. + description (Optional[str], optional): Object description. Defaults to None. + + Returns: + SDCResource: Created SDC resource object + + """ + return cls.create_from_api_response(cls.send_message_json( + "POST", + f"Create {cls.__name__.upper()} {name}", + cls.CREATE_ENDPOINT, + data=cls.get_create_payload(name=name, user=user, description=description, **kwargs) + )) + + +class SDCResourceTypeObject(SDCResource, ABC): # pylint: disable=too-few-public-methods + """SDC resource type object class.""" + + def __init__(self, # pylint: disable=too-many-locals too-many-arguments + *, + name: str, + version: Optional[str] = None, + archived: Optional[bool] = None, + component_type: Optional[str] = None, + icon: Optional[str] = None, + unique_id: Optional[str] = None, + lifecycle_state: Optional[LifecycleState] = None, + last_update_date: Optional[int] = None, + category_normalized_name: Optional[str] = None, + sub_category_normalized_name: Optional[str] = None, + uuid: Optional[str] = None, + invariant_uuid: Optional[str] = None, + system_name: Optional[str] = None, + description: Optional[str] = None, + tags: Optional[Sequence[str]] = None, + last_updater_user_id: Optional[str] = None, + creation_date: Optional[int] = None, + all_versions: Optional[Dict[str, Any]] = None): + """Initialise SDC resource type object. + + Args: + name (str): SDC resource object name + version (Optional[str], optional): SDC resource object version. Defaults to None. + archived (Optional[bool], optional): Flag determines if object is archived. + Defaults to None. + component_type (Optional[str], optional): Resource component type. Defaults to None. + icon (Optional[str], optional): Resource icon. Defaults to None. + unique_id (Optional[str], optional): Resource unique ID. Defaults to None. + lifecycle_state (Optional[LifecycleState], optional): SDC resource object lifecycle + state. Defaults to None. + last_update_date (Optional[int], optional): Last update timestamp. Defaults to None. + category_normalized_name (Optional[str], optional): Category normalized name. + Defaults to None. + sub_category_normalized_name (Optional[str], optional): Subcategory normalized name. + Defaults to None. + uuid (Optional[str], optional): Object UUID. Defaults to None. + invariant_uuid (Optional[str], optional): Object invariant UUID. Defaults to None. + system_name (Optional[str], optional): System name. Defaults to None. + description (Optional[str], optional): Resource description. Defaults to None. + tags (Optional[Sequence[str]], optional): Sequence of object tags. Defaults to None. + last_updater_user_id (Optional[str], optional): ID of user who is + the latest object's updater. Defaults to None. + creation_date (Optional[int], optional): Object's creation timestamp. Defaults to None. + all_versions (Optional[Dict[str, Any]], optional): Dictionary with all resources + versions. Defaults to None + """ + super().__init__( + name=name, + version=version, + archived=archived, + component_type=component_type, + icon=icon, + unique_id=unique_id, + lifecycle_state=lifecycle_state, + last_update_date=last_update_date, + uuid=uuid, + invariant_uuid=invariant_uuid, + system_name=system_name, + tags=tags, + last_updater_user_id=last_updater_user_id, + creation_date=creation_date, + description=description, + all_versions=all_versions + ) + self.category_nomalized_name = category_normalized_name + self.sub_category_nomalized_name: Optional[str] = sub_category_normalized_name + + @classmethod + def filter_response_objects_by_resource_type( + cls, + response: Dict[str, Any] + ) -> Iterable[Dict[str, Any]]: + """Filter object from response based on resource type. + + Args: + response (Dict[str, Any]): Response dictionary + + Returns: + Iterable[Dict[str, Any]]: Iterator object values dictionaries + + """ + for resource in response.get("resources", []): + if resource.get("resourceType") == cls.resource_type().value: + yield resource + + @classmethod + def create_from_api_response(cls, api_response: Dict[str, Any]) -> "SDCResourceTypeObject": + """Create SDC resource object using API response dictionary values. + + Args: + api_response (Dict[str, Any]): API responses dictionary + + Returns: + SDCResourceTypeObject: Object created using API response values. + + """ + return cls( + archived=api_response["archived"], + creation_date=api_response["creationDate"], + component_type=api_response["componentType"], + description=api_response["description"], + icon=api_response["icon"], + invariant_uuid=api_response["invariantUUID"], + last_update_date=api_response["lastUpdateDate"], + last_updater_user_id=api_response["lastUpdaterUserId"], + lifecycle_state=api_response["lifecycleState"], + name=api_response["name"], + system_name=api_response["systemName"], + tags=api_response["tags"], + unique_id=api_response["uniqueId"], + uuid=api_response["uuid"], + version=api_response["version"], + ) + + +class SDCResourceTypeObjectCreateMixin(SDCResourceCreateMixin, ABC): + """Mixin class to be used for SDC resource type object creation.""" + + @classmethod + @abstractmethod + def create_payload_template(cls) -> str: + """Get payload template to be used for creation request. + + Abstract classmethod + + Returns: + str: Name of template to be used for payload creation + + """ + + @classmethod + def get_create_payload(cls, # pylint: disable=too-many-arguments + name: str, + *, + vsp: Vsp, + vendor: Vendor, + user: Optional[SdcUser] = None, + description: Optional[str] = None, + category: Optional[ResourceCategory] = None, + subcategory: Optional[SdcSubCategory] = None) -> str: + """Get a payload to create a resource. + + Args: + vsp (Vsp): VSP object + vendor (Vendor): Vendor object + user (Optional[SdcUser], optional): User which be marked as a creator. + If not given "cs0008" is going to be used. Defaults to None. + description (Optional[str], optional): Resource description. Defaults to None. + category (Optional[ResourceCategory], optional): Resource category. + If not given (with subcategory) then "Generic: Network Service" is going to be used. + Defaults to None. + subcategory (Optional[SdcSubCategory], optional): Resource subcategory. + Defaults to None. + + Returns: + str: Resource creation payload + + """ + if not all([category, subcategory]): + category = ResourceCategory.get_by_name("Generic") + subcategory = category.get_subcategory("Network Service") + return jinja_env().get_template(cls.create_payload_template()).render( + name=name, + vsp=vsp, + vendor=vendor, + category=category, + subcategory=subcategory, + user_id=user.user_id if user else "cs0008", + description=description if description else "ONAP SDK Resource" + ) diff --git a/src/onapsdk/sdc2/sdc_user.py b/src/onapsdk/sdc2/sdc_user.py new file mode 100644 index 0000000..bc5a325 --- /dev/null +++ b/src/onapsdk/sdc2/sdc_user.py @@ -0,0 +1,119 @@ +"""SDC user module.""" +# Copyright 2024 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 onapsdk.sdc2.sdc import ResoureTypeEnum +from enum import Enum +from typing import Any, Dict, Iterator +from urllib.parse import urljoin + +from onapsdk.exceptions import ResourceNotFound # type: ignore +from onapsdk.sdc2.sdc import SDCCatalog + + +class SdcUser(SDCCatalog): # pylint: disable=too-many-instance-attributes + """SDC user class.""" + + GET_ALL_ENDPOINT = "sdc2/rest/v1/user/users" + + class SdcUserStatus(Enum): # pylint: disable=too-few-public-methods + """SDC user status enum.""" + + ACTIVE = "ACTIVE" + INACTIVE = "INACTIVE" + + def __init__(self, # pylint: disable=too-many-arguments too-many-instance-attributes + user_id: str, + role: str, + email: str, + first_name: str, + full_name: str, + last_login_time: int, + last_name: str, + status: str) -> None: + """Initialise SDC user class object. + + Args: + user_id (str): User ID. Would be used as name as well (as each SDC object has name). + role (str): User role + email (str): User email + first_name (str): User first name + full_name (str): User full name + last_login_time (int): User last login timestamp + last_name (str): User last name + status (str): User status + """ + super().__init__(name=user_id) + self.user_id: str = user_id + self.role: str = role + self.email: str = email + self.first_name: str = first_name + self.full_name: str = full_name + self.last_login_time: int = last_login_time + self.last_name: str = last_name + self.status: str = status + + @classmethod + def get_all(cls) -> Iterator["SdcUser"]: + """Get all users. + + Returns: + Iterator["SdcUser"]: SDC users iterator + + """ + return (cls.create_from_api_response(response_obj) for + response_obj in cls.send_message_json( + "GET", + f"Get all {cls.__name__}", + urljoin(cls.base_back_url, cls.GET_ALL_ENDPOINT) + )) + + @classmethod + def get_by_user_id(cls, user_id: str) -> "SdcUser": + """Get an user by it's ID. + + Args: + user_id (str): ID of user to get + + Raises: + ResourceNotFound: User with given ID not found + + Returns: + SdcUser: SDC user with given ID. + + """ + for user in cls.get_all(): + if user.user_id == user_id: + return user + raise ResourceNotFound(f"{cls.__name__} with name {user_id} user ID not found") + + @classmethod + def create_from_api_response(cls, api_response: Dict[str, Any]) -> "SdcUser": + """Create sdc user using values returned by API. + + Args: + api_response (Dict[str, Any]): API response values dictionary. + + Returns: + SdcUser: SDC user created using values from API response. + + """ + return cls( + user_id=api_response["userId"], + role=api_response["role"], + email=api_response["email"], + first_name=api_response["firstName"], + full_name=api_response["fullName"], + last_login_time=api_response["lastLoginTime"], + last_name=api_response["lastName"], + status=cls.SdcUserStatus(api_response["status"]), + ) diff --git a/src/onapsdk/sdc2/service.py b/src/onapsdk/sdc2/service.py new file mode 100644 index 0000000..4f7482f --- /dev/null +++ b/src/onapsdk/sdc2/service.py @@ -0,0 +1,504 @@ +"""SDC service module.""" +# Copyright 2024 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 enum import Enum +from typing import Any, Dict, Iterable, Iterator, Sequence, Optional, Set +from urllib.parse import urljoin + +from onapsdk.configuration import settings # type: ignore +from onapsdk.sdc2.sdc import SDC, ResoureTypeEnum +from onapsdk.sdc2.sdc_category import SdcCategory, ServiceCategory +from onapsdk.sdc2.sdc_resource import SDCResource, SDCResourceCreateMixin +from onapsdk.sdc2.sdc_user import SdcUser +from onapsdk.utils.jinja import jinja_env # type: ignore + + +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 ServiceDistribution(SDC): + """Service distribution class.""" + + DISTRIBUTED_DEPLOYMENT_STATUS = "Distributed" + + @dataclass + class DistributionStatus: + """Dataclass of service distribution status. Internal usage only.""" + + component_id: str + timestamp: str + url: str + status: str + error_reason: str + + @property + def failed(self) -> bool: + """Flad to determine if distribution status is failed or not. + + If error reason of distribution status is not empty it doesn't mean + always that distribution failed at all. On some cases that means + that service was already distributed on that component. That's why + we checks also if status is not "ALREADY_DEPLOYED". + + Returns: + bool: True if distribution on component failed or not. + + """ + return self.error_reason != "null" and \ + self.status != "ALREADY_DEPLOYED" + + def __init__(self, # pylint: disable=too-many-arguments + distribution_id: str, + timestamp: str, + user_id: str, + deployment_status: str) -> None: + """Initialise service distribution class. + + Stores information about service distribution and is a source of truth + if service is distributed or not. + + Args: + distribution_id (str): Distribution ID + timestamp (str): Distribution timestamp + user_id (str): ID of user which requested distribution. + deployment_status (str): Status of deployment + + """ + super().__init__(name=distribution_id) + self.distribution_id: str = distribution_id + self.timestamp: str = timestamp + self.user_id: str = user_id + self.deployment_status: str = deployment_status + self._distribution_status_list: Optional[ + Sequence["self.DistributionStatus"]] = None # type: ignore + + @property + def distributed(self) -> bool: + """Distribution status. + + Need to pass 3 tests: + - deployment status of distribution it "Distributed" + - service was distributed on all components listed + on settings.SDC_SERVICE_DISTRIBUTION_COMPONENTS + - there was no distribution error + + An order of tests is fixed to reduce SDC API calls. + + Returns: + bool: True is service can be considered as distributed, False otherwise. + + """ + return all([ + self._deployment_status_test, + self._distribution_components_test, + self._no_distribution_errors_test + ]) + + @property + def _deployment_status_test(self) -> bool: + """Test to check a distribution deployment status. + + Passed if distribution status is equal to "Distributed" + + Returns: + bool: True if distribution deployment status is equal to "Distributed". + False otherwise + + """ + return self.deployment_status == self.DISTRIBUTED_DEPLOYMENT_STATUS + + @property + def _distribution_components_test(self) -> bool: + """Test to check if all required components were notified about distribution. + + List of required components can be configured via SDC_SERVICE_DISTRIBUTION_COMPONENTS + setting value. + + Returns: + bool: True if all required components were notified, False otherwise + + """ + notified_components_set: Set[str] = { + distribution.component_id for distribution in self.distribution_status_list + } + return notified_components_set == set(settings.SDC_SERVICE_DISTRIBUTION_COMPONENTS) + + @property + def _no_distribution_errors_test(self) -> bool: + """Test to check if there is no error on any component distribution. + + Returns: + bool: True if no error occured on any component distribution, False otherwise + + """ + return not list(filter(lambda obj: obj.failed, + self.distribution_status_list)) + + @property + def distribution_status_list(self) -> Sequence[DistributionStatus]: + """List of distribution statuses. + + Returns: + List[DistributionStatus]: List of distribution statuses. + + """ + if not self._distribution_status_list: + self._distribution_status_list = [self.DistributionStatus( + component_id=distribution_status_dict["omfComponentID"], + timestamp=distribution_status_dict["timestamp"], + url=distribution_status_dict["url"], + status=distribution_status_dict["status"], + error_reason=distribution_status_dict["errorReason"] + ) for distribution_status_dict in + self.send_message_json( + "GET", + f"Get status of {self.distribution_id} distribution", + urljoin(self.base_back_url, + f"sdc2/rest/v1/catalog/services/distribution/{self.distribution_id}") + ).get("distributionStatusList", []) + ] + return self._distribution_status_list + + + +class Service(SDCResource, SDCResourceCreateMixin): + """SDC service class.""" + + ADD_RESOURCE_TEMPLATE = "sdc2_add_resource.json.j2" + CREATE_ENDPOINT = urljoin(SDC.base_back_url, "sdc2/rest/v1/catalog/services") + CREATE_SERVICE_TEMPLATE = "sdc2_create_service.json.j2" + + def __init__(self, # pylint: disable=too-many-locals too-many-arguments + *, + name: str, + version: Optional[str] = None, + archived: Optional[bool] = None, + component_type: Optional[str] = None, + icon: Optional[str] = None, + unique_id: Optional[str] = None, + lifecycle_state: Optional[str] = None, + last_update_date: Optional[int] = None, + uuid: Optional[str] = None, + invariant_uuid: Optional[str] = None, + system_name: Optional[str] = None, + tags: Optional[Sequence[str]] = None, + last_updater_user_id: Optional[str] = None, + creation_date: Optional[int] = None, + description: Optional[str] = None, + actual_component_type: Optional[str] = None, + all_versions: Optional[Dict[str, str]] = None, + categories: Optional[Sequence[SdcCategory]] = None, + distribuition_status: Optional[str] = None, + instantiation_type: Optional[ServiceInstantiationType] = None) -> None: + """Initialize service object. + + Args: + name (str): Service name + actual_component_type (Optional[str]): Service actual component type. Defaults to None. + all_versions (Optional[Dict[str, str]]): Dictionary with all versions of service. + Defaults to None. + categories (Optional[List[SdcCategory]]): List with all serivce categories. + Defaults to None. + version (Optional[str], optional): Service version. Defaults to None. + archived (Optional[bool], optional): Flag determines if service is archived or not. + Defaults to None. + component_type (Optional[str], optional): Service component type. Defaults to None. + icon (Optional[str], optional): Service icon. Defaults to None. + unique_id (Optional[str], optional): Service unique ID. Defaults to None. + lifecycle_state (Optional[str], optional): Service lifecycle state. Defaults to None. + last_update_date (Optional[int], optional): Service last update date. Defaults to None. + uuid (Optional[str], optional): Service UUID. Defaults to None. + invariant_uuid (Optional[str], optional): Service invariant UUID. Defaults to None. + system_name (Optional[str], optional): Service system name. Defaults to None. + tags (Optional[List[str]], optional): List with service tags. Defaults to None. + last_updater_user_id (Optional[str], optional): ID of user who was last service + updater. Defaults to None. + creation_date (Optional[int], optional): Timestamp of service creation. + Defaults to None. + description (Optional[str], optional): Service description. + Defaults to None. + distribuition_status (Optional[str], optional): Service distribution status. + Defaults to None. + """ + super().__init__( + name=name, + archived=archived, + version=version, + icon=icon, + component_type=component_type, + unique_id=unique_id, + uuid=uuid, + lifecycle_state=lifecycle_state, + last_update_date=last_update_date, + tags=tags, + invariant_uuid=invariant_uuid, + system_name=system_name, + creation_date=creation_date, + last_updater_user_id=last_updater_user_id, + description=description + ) + self.actual_component_type: Optional[str] = actual_component_type + self.all_versions: Optional[Dict[str, str]] = all_versions + self.distribuition_status: Optional[str] = distribuition_status + self.categories: Optional[Sequence[SdcCategory]] = categories + self.instantiation_type: Optional[ServiceInstantiationType] = instantiation_type + + @classmethod + def resource_type(cls) -> ResoureTypeEnum: + """Service resource type enum value. + + Returns: + ResoureTypeEnum: Service resource type enum value + + """ + return ResoureTypeEnum.SERVICE + + @classmethod + def filter_response_objects_by_resource_type( + cls, + response: Dict[str, Any] + ) -> Iterable[Dict[str, Any]]: + """Filter list of objects returned by API by resource type. + + Return only "services" from API response to reduce objects to iterate. + + Args: + response (Dict[str, Any]): API response dictionary + + Returns: + Iterable[Dict[str, Any]]: Dictionaries containing only services data + + """ + return response.get("services", []) + + @classmethod + def create_from_api_response(cls, api_response: Dict[str, Any]) -> "Service": # type: ignore + """Create Service using values from API response. + + Args: + api_response (Dict[str, Any]): Dictionary with values returned by API. + + Returns: + Service: Service object + + """ + return cls( + actual_component_type=api_response["actualComponentType"], + all_versions=api_response["allVersions"], + creation_date=api_response["creationDate"], + version=api_response["version"], + component_type=api_response["componentType"], + unique_id=api_response["uniqueId"], + icon=api_response["icon"], + lifecycle_state=api_response["lifecycleState"], + last_update_date=api_response["lastUpdateDate"], + name=api_response["name"], + invariant_uuid=api_response["invariantUUID"], + distribuition_status=api_response["distributionStatus"], + description=api_response["description"], + uuid=api_response["uuid"], + system_name=api_response["systemName"], + tags=api_response["tags"], + last_updater_user_id=api_response["lastUpdaterUserId"], + archived=api_response["archived"], + categories=[ServiceCategory.get_by_uniqe_id(response_category["uniqueId"]) + for response_category in api_response["categories"]], + instantiation_type=ServiceInstantiationType(api_response["instantiationType"]) + ) + + def update(self, api_response: Dict[str, Any]) -> None: + """Update service with values from API response. + + Args: + api_response (Dict[str, Any]): API response dictionary which values from are going to + be used to update service object + + """ + super().update(api_response) + self.distribuition_status = api_response["distributionStatus"] + + @classmethod + def get_create_payload(cls, # pylint: disable=arguments-differ too-many-arguments + name: str, + *, + user: Optional[SdcUser] = None, + description: Optional[str] = None, + category: Optional[ServiceCategory] = None, + instantiation_type: ServiceInstantiationType = \ + ServiceInstantiationType.MACRO) -> str: + """Get a payload to be sued for service creation. + + Args: + name (str): Name of the service to be created + user (Optional[SdcUser], optional): User which will be marked as a creaton. If no user + is passed then 'cs0008' user ID is going to be used. Defaults to None. + description (Optional[str], optional): Service description. Defaults to None. + category (Optional[ServiceCategory], optional): Service category. + If no category is given then "Network Service" is going to be used. + Defaults to None. + + Returns: + str: Service creation API payload. + + """ + return jinja_env().get_template(cls.CREATE_SERVICE_TEMPLATE).render( + name=name, + category=category if category else ServiceCategory.get_by_name("Network Service"), + user_id=user.user_id if user else "cs0008", + description=description if description else "ONAP SDK Service", + instantiation_type=instantiation_type) + + def add_resource(self, resource: SDCResource) -> None: + """Add resource into service composition. + + Args: + resource (SDCResource): Resource to be added into service. + + """ + self.send_message( + "POST", + f"Add resource {resource.name} into service {self.name}", + urljoin(self.base_back_url, + f"sdc2/rest/v1/catalog/services/{self.unique_id}/resourceInstance/"), + data=jinja_env().get_template(self.ADD_RESOURCE_TEMPLATE).render(resource=resource) + ) + + def distribute(self, env: str = "PROD") -> None: + """Distribute service. + + Call a request to distribute service. If no error was returned then service is updated + using values returned by API. + SDC allows to distribute services on different environments. By default that method + distribute service on "PROD" environment. + + Args: + env (str, optional): Environment to distribute service on. Defaults to "PROD". + + """ + response: Dict[str, Any] = self.send_message_json( + "POST", + f"Request distribute Service {self.name}", + urljoin(self.base_back_url, + f"sdc2/rest/v1/catalog/services/{self.unique_id}/distribution/{env}/activate") + ) + self.update(response) + + @classmethod + def catalog_type(cls) -> str: + """Service type resource catalog type. + + SDC resources has two catalog types: resources and services. To create + API endpoints which can be used by both that classmethod is overwriten + by Service class. + + Returns: + str: Service catalog type + + """ + return "services" + + @classmethod + def get_by_name_and_version_endpoint(cls, name: str, version: str) -> str: + """Get an endpoint to call a request to get service by it's name and version. + + Service has different endpoint that other resources to send a request + to get an object by it's name and version. + + Args: + name (str): Service name + version (str): Service version + + Returns: + str: Endpoint to call a request to get service by it name and version + + """ + return f"sdc2/rest/v1/catalog/services/serviceName/{name}/serviceVersion/{version}" + + @classmethod + def add_deployment_artifact_endpoint(cls, object_id: str) -> str: + """Get an endpoint to add a deployment artifact into service. + + Service has different endpoint to send a request for adding + a deployment artifact. + + Args: + object_id (str): Service object ID to create an endpoint for + + Returns: + str: Endpoint used to send request to add deployment artifact + + """ + return f"sdc2/rest/v1/catalog/services/{object_id}/artifacts" + + @property + def distributions(self) -> Iterator[ServiceDistribution]: + """Get service distributions. + + Service can be distributed multiple times. That property + returns and iterable object which returns all + distributions in reversed order we get it from API, + so first distribution would be the latest one, not the first + distribution call as it was in API. + + Returns: + Iterable[ServiceDistribution]: Service distributions iterator + + """ + for distribution_status_dict in reversed(self.send_message_json( + "GET", + f"Request Service {self.name} distributions", + urljoin(self.base_back_url, f"sdc2/rest/v1/catalog/services/{self.uuid}/distribution/") + ).get("distributionStatusOfServiceList", [])): + yield ServiceDistribution(distribution_status_dict["distributionID"], + distribution_status_dict["timestamp"], + distribution_status_dict["userId"], + distribution_status_dict["deployementStatus"]) + + @property + def latest_distribution(self) -> Optional[ServiceDistribution]: + """Get the latest distribution of the service. + + Returns: + ServiceDistribution|None: Latest service distribution or + None if service was not distrubuted + """ + try: + return next(self.distributions) + except StopIteration: + return None + + @property + def distributed(self) -> bool: + """Distributed property. + + Return boolean value which determines if serivce was distributed or not. + It checks if latest distribution of service was successfull. + + Returns: + bool: True is service was distributed correctly, False otherwise. + """ + if (latest_distribution := self.latest_distribution) is not None: + return latest_distribution.distributed + return False diff --git a/src/onapsdk/sdc2/templates/sdc2_add_deployment_artifact.json.j2 b/src/onapsdk/sdc2/templates/sdc2_add_deployment_artifact.json.j2 new file mode 100644 index 0000000..dc38011 --- /dev/null +++ b/src/onapsdk/sdc2/templates/sdc2_add_deployment_artifact.json.j2 @@ -0,0 +1,8 @@ +{ + "artifactGroupType": "{{ artifact_group_type }}", + "artifactName": "{{ artifact_name }}", + "artifactLabel": "{{ artifact_label }}", + "artifactType": "{{ artifact_type }}", + "description": "{{ artifact_description }}", + "payloadData": "{{ artifact_payload }}" +} diff --git a/src/onapsdk/sdc2/templates/sdc2_add_resource.json.j2 b/src/onapsdk/sdc2/templates/sdc2_add_resource.json.j2 new file mode 100644 index 0000000..9340909 --- /dev/null +++ b/src/onapsdk/sdc2/templates/sdc2_add_resource.json.j2 @@ -0,0 +1,7 @@ +{ + "name": "{{ resource.name }}", + "originType": "{{ resource.resource_type().value }}", + "componentUid": "{{ resource.unique_id }}", + "componentVersion": "{{ resource.version }}", + "icon":"{{ resource.icon }}" +} diff --git a/src/onapsdk/sdc2/templates/sdc2_component_instance_input_set_value.json.j2 b/src/onapsdk/sdc2/templates/sdc2_component_instance_input_set_value.json.j2 new file mode 100644 index 0000000..fe3f6af --- /dev/null +++ b/src/onapsdk/sdc2/templates/sdc2_component_instance_input_set_value.json.j2 @@ -0,0 +1,13 @@ +[ + { + "name":"{{ component_instance_input.name }}", + "parentUniqueId":"{{ component_instance_input.component_instance.unique_id }}", + "type":"{{ component_instance_input.input_type }}", + "uniqueId":"{{ component_instance_input.unique_id }}", + "value":"{{ value }}", + "definition":{{ component_instance_input.definition | tojson }}, + "toscaPresentation":{ + "ownerId":"{{ component_instance_input.component_instance.unique_id }}" + } + } +]
\ No newline at end of file diff --git a/src/onapsdk/sdc2/templates/sdc2_create_pnf.json.j2 b/src/onapsdk/sdc2/templates/sdc2_create_pnf.json.j2 new file mode 100644 index 0000000..3f8f0ca --- /dev/null +++ b/src/onapsdk/sdc2/templates/sdc2_create_pnf.json.j2 @@ -0,0 +1,2 @@ +{% extends "sdc2_create_resource_base.json.j2" %} +{% block resource_type %}PNF{% endblock %}
\ No newline at end of file diff --git a/src/onapsdk/sdc2/templates/sdc2_create_resource_base.json.j2 b/src/onapsdk/sdc2/templates/sdc2_create_resource_base.json.j2 new file mode 100644 index 0000000..4e8cb46 --- /dev/null +++ b/src/onapsdk/sdc2/templates/sdc2_create_resource_base.json.j2 @@ -0,0 +1,21 @@ +{ + "name": "{{ name }}", + "contactId": "{{ user_id }}", + "componentType": "RESOURCE", + {% if category.name != "Allotted Resource" and vsp is not none %} + "csarUUID": "{{ vsp.csar_uuid }}", + "csarVersion": "1.0", + {% endif %} + "categories": [{ + "name": "{{ category.name }}", + "uniqueId": "{{ category.unique_id }}", + "subcategories": [{ + "name": "{{ subcategory.name }}", + "uniqueId": "{{ subcategory.unique_id }}" + }] + }], + "resourceType": "{% block resource_type %}{% endblock %}", + "description": "{{ description }}", + "vendorName": "{{ vendor.name }}", + "vendorRelease": "1.0" +} diff --git a/src/onapsdk/sdc2/templates/sdc2_create_service.json.j2 b/src/onapsdk/sdc2/templates/sdc2_create_service.json.j2 new file mode 100644 index 0000000..b28a1b1 --- /dev/null +++ b/src/onapsdk/sdc2/templates/sdc2_create_service.json.j2 @@ -0,0 +1,13 @@ +{ + "componentType": "SERVICE", + "name": "{{ name }}", + "contactId": "{{ user_id }}", + "categories": [ + { + "name": "{{ category.name }}", + "uniqueId": "{{ category.unique_id }}" + } + ], + "instantiationType": "{{ instantiation_type.value }}", + "description": "{{ description }}" +}
\ No newline at end of file diff --git a/src/onapsdk/sdc2/templates/sdc2_create_vf.json.j2 b/src/onapsdk/sdc2/templates/sdc2_create_vf.json.j2 new file mode 100644 index 0000000..8f516ea --- /dev/null +++ b/src/onapsdk/sdc2/templates/sdc2_create_vf.json.j2 @@ -0,0 +1,2 @@ +{% extends "sdc2_create_resource_base.json.j2" %} +{% block resource_type %}VF{% endblock %}
\ No newline at end of file diff --git a/src/onapsdk/sdc2/templates/sdc2_resource_action.json.j2 b/src/onapsdk/sdc2/templates/sdc2_resource_action.json.j2 new file mode 100644 index 0000000..8eb06ea --- /dev/null +++ b/src/onapsdk/sdc2/templates/sdc2_resource_action.json.j2 @@ -0,0 +1,3 @@ +{ + "userRemarks": "{{ lifecycle_operation | lower }}" +} diff --git a/src/onapsdk/sdc2/vf.py b/src/onapsdk/sdc2/vf.py new file mode 100644 index 0000000..3e7e59e --- /dev/null +++ b/src/onapsdk/sdc2/vf.py @@ -0,0 +1,40 @@ +"""SDC VF module.""" +# Copyright 2024 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 onapsdk.sdc2.sdc import ResoureTypeEnum +from onapsdk.sdc2.sdc_resource import SDCResourceTypeObject, SDCResourceTypeObjectCreateMixin + + +class Vf(SDCResourceTypeObject, SDCResourceTypeObjectCreateMixin): # pylint: disable=too-many-ancestors + """VF class.""" + + @classmethod + def resource_type(cls) -> ResoureTypeEnum: + """VF resource type. + + Returns: + ResoureTypeEnum: VF resource type enum value + + """ + return ResoureTypeEnum.VF + + @classmethod + def create_payload_template(cls) -> str: + """Get a template to create VF creation request payload. + + Returns: + str: VF creation template. + + """ + return "sdc2_create_vf.json.j2" diff --git a/src/onapsdk/sdc2/vl.py b/src/onapsdk/sdc2/vl.py new file mode 100644 index 0000000..babab68 --- /dev/null +++ b/src/onapsdk/sdc2/vl.py @@ -0,0 +1,30 @@ +"""SDC VL module.""" +# Copyright 2024 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 onapsdk.sdc2.sdc import ResoureTypeEnum +from onapsdk.sdc2.sdc import ResoureTypeEnum +from onapsdk.sdc2.sdc_resource import SDCResourceTypeObject + + +class Vl(SDCResourceTypeObject): + """VL class.""" + + @classmethod + def resource_type(cls) -> ResoureTypeEnum: + """VL resource type. + + Returns: + ResoureTypeEnum: VL resource type enum value + + """ + return ResoureTypeEnum.VL diff --git a/src/onapsdk/utils/jinja.py b/src/onapsdk/utils/jinja.py index 8af130c..10160b2 100644 --- a/src/onapsdk/utils/jinja.py +++ b/src/onapsdk/utils/jinja.py @@ -42,7 +42,7 @@ def jinja_env() -> Environment: PackageLoader("onapsdk.k8s"), PackageLoader("onapsdk.nbi"), PackageLoader("onapsdk.sdc"), - PackageLoader("onapsdk.sdnc"), + PackageLoader("onapsdk.sdc2"), PackageLoader("onapsdk.sdnc"), PackageLoader("onapsdk.so"), PackageLoader("onapsdk.ves"), diff --git a/src/onapsdk/version.py b/src/onapsdk/version.py index c4a77e0..5ff20c2 100644 --- a/src/onapsdk/version.py +++ b/src/onapsdk/version.py @@ -13,4 +13,4 @@ # See the License for the specific language governing permissions and # limitations under the License. -__version__ = "12.10.0" +__version__ = "12.11.0" |