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 | |
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>
32 files changed, 4674 insertions, 6 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" diff --git a/tests/test_sdc2.py b/tests/test_sdc2.py new file mode 100644 index 0000000..ac4a92c --- /dev/null +++ b/tests/test_sdc2.py @@ -0,0 +1,19 @@ + +from urllib.parse import parse_qs + +from pytest import raises + +from onapsdk.sdc2.sdc import SDC, ResoureTypeEnum +from onapsdk.sdc2.sdc_resource import SDCResource + + +def test_resource_type_enum(): + assert len(list(ResoureTypeEnum)) == 13 + + +def test_sdc_filter_exclude_resource_type(): + for resource_type in ResoureTypeEnum: + resource_types_list_without_one_type = list(ResoureTypeEnum.iter_without_resource_type(resource_type)) + assert len(resource_types_list_without_one_type) == 12 + with raises(ValueError): + resource_types_list_without_one_type.index(resource_type) diff --git a/tests/test_sdc2_category.py b/tests/test_sdc2_category.py new file mode 100644 index 0000000..3988f2d --- /dev/null +++ b/tests/test_sdc2_category.py @@ -0,0 +1,437 @@ + +from collections import namedtuple +from random import randint +from unittest.mock import patch, PropertyMock +from uuid import uuid4 + +from pytest import raises + +from onapsdk.configuration import settings +from onapsdk.exceptions import ResourceNotFound +from onapsdk.sdc2.sdc_category import ServiceCategory, ResourceCategory, ProductCategory, SdcSubCategory + + +# Service categories tests +@patch("onapsdk.sdc2.sdc_category.ServiceCategory.send_message_json") +@patch("onapsdk.sdc2.sdc_category.ServiceCategory.create_from_api_response") +def test_service_category_get_all(mock_create_from_api_response, mock_send_message_json): + mock_send_message_json.return_value = [] + assert len(list(ServiceCategory.get_all())) == 0 + mock_send_message_json.assert_called_once_with( + "GET", + "Get all ServiceCategory", + f"{settings.SDC_BE_URL}/sdc2/rest/v1/categories/services" + ) + + mock_send_message_json.return_value = [{}] + assert len(list(ServiceCategory.get_all())) == 1 + mock_create_from_api_response.assert_called_once_with({}) + + +@patch("onapsdk.sdc2.sdc_category.ServiceCategory.get_all") +def test_service_category_get_by_unique_id(mock_get_all): + mock_get_all.return_value = [] + with raises(ResourceNotFound): + ServiceCategory.get_by_uniqe_id("test_unique_id") + + TestCategory = namedtuple("TestCategory", "unique_id") + mock_get_all.return_value = [TestCategory(f"unique_id_{i}") for i in range(10**2)] + with raises(ResourceNotFound): + ServiceCategory.get_by_uniqe_id("test_unique_id") + for i in range(10**2): + assert ServiceCategory.get_by_uniqe_id(f"unique_id_{i}") is not None + + +@patch("onapsdk.sdc2.sdc_category.ServiceCategory.get_all") +def test_service_category_get_by_name(mock_get_all): + mock_get_all.return_value = [] + with raises(ResourceNotFound): + ServiceCategory.get_by_name("test_name") + + TestCategory = namedtuple("TestCategory", "name") + mock_get_all.return_value = [TestCategory(f"name_{i}") for i in range(10**2)] + with raises(ResourceNotFound): + ServiceCategory.get_by_name("test_name") + for i in range(10**2): + assert ServiceCategory.get_by_name(f"name_{i}") is not None + + +def test_service_category_create_from_api_response(): + api_response = { + "name": str(uuid4()), + "icons": [str(uuid4())], + "models": [str(uuid4())], + "normalizedName": str(uuid4()), + "uniqueId": str(uuid4()), + "ownerId": str(uuid4()), + "type": str(uuid4()), + "version": str(uuid4()), + "displayName": str(uuid4()), + "empty": bool(randint(0, 1)), + "useServiceSubstitutionForNestedServices": bool(randint(0, 1)), + } + si = ServiceCategory.create_from_api_response(api_response) + assert si.name == api_response["name"] + assert si.empty == api_response["empty"] + assert si.icons == api_response["icons"] + assert si.models == api_response["models"] + assert si.normalized_name == api_response["normalizedName"] + assert si.unique_id == api_response["uniqueId"] + assert si.use_service_substitution_for_nested_services == api_response["useServiceSubstitutionForNestedServices"] + assert si.owner_id == api_response["ownerId"] + assert si.subcategories == None + assert si.category_type == api_response["type"] + assert si.version == api_response["version"] + assert si.display_name == api_response["displayName"] + + + api_response = { + "name": str(uuid4()), + "icons": [str(uuid4())], + "models": [str(uuid4())], + "normalizedName": str(uuid4()), + "uniqueId": str(uuid4()), + "ownerId": str(uuid4()), + "type": str(uuid4()), + "version": str(uuid4()), + "displayName": str(uuid4()), + "empty": bool(randint(0, 1)), + "useServiceSubstitutionForNestedServices": bool(randint(0, 1)), + "subcategories": [ + { + "name": str(uuid4()), + "normalizedName": str(uuid4()), + "uniqueId": str(uuid4()), + }, + { + "name": str(uuid4()), + "normalizedName": str(uuid4()), + "uniqueId": str(uuid4()), + } + ] + } + si = ServiceCategory.create_from_api_response(api_response) + assert si.name == api_response["name"] + assert si.empty == api_response["empty"] + assert si.icons == api_response["icons"] + assert si.models == api_response["models"] + assert si.normalized_name == api_response["normalizedName"] + assert si.unique_id == api_response["uniqueId"] + assert si.use_service_substitution_for_nested_services == api_response["useServiceSubstitutionForNestedServices"] + assert si.owner_id == api_response["ownerId"] + assert len(si.subcategories) == 2 + assert si.subcategories[0].name == api_response["subcategories"][0]["name"] + assert si.subcategories[0].normalized_name == api_response["subcategories"][0]["normalizedName"] + assert si.subcategories[0].unique_id == api_response["subcategories"][0]["uniqueId"] + assert si.subcategories[1].name == api_response["subcategories"][1]["name"] + assert si.subcategories[1].normalized_name == api_response["subcategories"][1]["normalizedName"] + assert si.subcategories[1].unique_id == api_response["subcategories"][1]["uniqueId"] + assert si.category_type == api_response["type"] + assert si.version == api_response["version"] + assert si.display_name == api_response["displayName"] + + +def test_service_category_get_subcategory(): + sc = ServiceCategory( + name=str(uuid4()), + empty=bool(randint(0, 1)), + icons=[str(uuid4())], + models=[str(uuid4())], + normalized_name=str(uuid4()), + unique_id=str(uuid4()), + use_service_substitution_for_nested_services=bool(randint(0, 1)), + subcategories=None + ) + assert sc.get_subcategory(subcategory_name="test_subc") is None + sc.subcategories = [SdcSubCategory( + name=f"subcategory_{i}", + normalized_name=f"subcategory_{i}", + unique_id=str(uuid4()) + ) for i in range(10**2)] + assert sc.get_subcategory(subcategory_name="test_subc") is None + for i in range(10**2): + assert sc.get_subcategory(subcategory_name=f"subcategory_{i}") is not None + + +# Resource categories tests +@patch("onapsdk.sdc2.sdc_category.ResourceCategory.send_message_json") +@patch("onapsdk.sdc2.sdc_category.ResourceCategory.create_from_api_response") +def test_resource_category_get_all(mock_create_from_api_response, mock_send_message_json): + mock_send_message_json.return_value = [] + assert len(list(ResourceCategory.get_all())) == 0 + mock_send_message_json.assert_called_once_with( + "GET", + "Get all ResourceCategory", + f"{settings.SDC_BE_URL}/sdc2/rest/v1/categories/resources" + ) + + mock_send_message_json.return_value = [{}] + assert len(list(ResourceCategory.get_all())) == 1 + mock_create_from_api_response.assert_called_once_with({}) + + +@patch("onapsdk.sdc2.sdc_category.ResourceCategory.get_all") +def test_resource_category_get_by_unique_id(mock_get_all): + mock_get_all.return_value = [] + with raises(ResourceNotFound): + ResourceCategory.get_by_uniqe_id("test_unique_id") + + TestCategory = namedtuple("TestCategory", "unique_id") + mock_get_all.return_value = [TestCategory(f"unique_id_{i}") for i in range(10**2)] + with raises(ResourceNotFound): + ResourceCategory.get_by_uniqe_id("test_unique_id") + for i in range(10**2): + assert ResourceCategory.get_by_uniqe_id(f"unique_id_{i}") is not None + + +@patch("onapsdk.sdc2.sdc_category.ResourceCategory.get_all") +def test_resource_category_get_by_name(mock_get_all): + mock_get_all.return_value = [] + with raises(ResourceNotFound): + ResourceCategory.get_by_name("test_name") + + TestCategory = namedtuple("TestCategory", "name") + mock_get_all.return_value = [TestCategory(f"name_{i}") for i in range(10**2)] + with raises(ResourceNotFound): + ResourceCategory.get_by_name("test_name") + for i in range(10**2): + assert ResourceCategory.get_by_name(f"name_{i}") is not None + + +def test_resource_category_create_from_api_response(): + api_response = { + "name": str(uuid4()), + "icons": [str(uuid4())], + "models": [str(uuid4())], + "normalizedName": str(uuid4()), + "uniqueId": str(uuid4()), + "ownerId": str(uuid4()), + "type": str(uuid4()), + "version": str(uuid4()), + "displayName": str(uuid4()), + "empty": bool(randint(0, 1)), + "useServiceSubstitutionForNestedServices": bool(randint(0, 1)), + } + rc = ResourceCategory.create_from_api_response(api_response) + assert rc.name == api_response["name"] + assert rc.empty == api_response["empty"] + assert rc.icons == api_response["icons"] + assert rc.models == api_response["models"] + assert rc.normalized_name == api_response["normalizedName"] + assert rc.unique_id == api_response["uniqueId"] + assert rc.use_service_substitution_for_nested_services == api_response["useServiceSubstitutionForNestedServices"] + assert rc.owner_id == api_response["ownerId"] + assert rc.subcategories == None + assert rc.category_type == api_response["type"] + assert rc.version == api_response["version"] + assert rc.display_name == api_response["displayName"] + + + api_response = { + "name": str(uuid4()), + "icons": [str(uuid4())], + "models": [str(uuid4())], + "normalizedName": str(uuid4()), + "uniqueId": str(uuid4()), + "ownerId": str(uuid4()), + "type": str(uuid4()), + "version": str(uuid4()), + "displayName": str(uuid4()), + "empty": bool(randint(0, 1)), + "useServiceSubstitutionForNestedServices": bool(randint(0, 1)), + "subcategories": [ + { + "name": str(uuid4()), + "normalizedName": str(uuid4()), + "uniqueId": str(uuid4()), + }, + { + "name": str(uuid4()), + "normalizedName": str(uuid4()), + "uniqueId": str(uuid4()), + } + ] + } + rc = ResourceCategory.create_from_api_response(api_response) + assert rc.name == api_response["name"] + assert rc.empty == api_response["empty"] + assert rc.icons == api_response["icons"] + assert rc.models == api_response["models"] + assert rc.normalized_name == api_response["normalizedName"] + assert rc.unique_id == api_response["uniqueId"] + assert rc.use_service_substitution_for_nested_services == api_response["useServiceSubstitutionForNestedServices"] + assert rc.owner_id == api_response["ownerId"] + assert len(rc.subcategories) == 2 + assert rc.subcategories[0].name == api_response["subcategories"][0]["name"] + assert rc.subcategories[0].normalized_name == api_response["subcategories"][0]["normalizedName"] + assert rc.subcategories[0].unique_id == api_response["subcategories"][0]["uniqueId"] + assert rc.subcategories[1].name == api_response["subcategories"][1]["name"] + assert rc.subcategories[1].normalized_name == api_response["subcategories"][1]["normalizedName"] + assert rc.subcategories[1].unique_id == api_response["subcategories"][1]["uniqueId"] + assert rc.category_type == api_response["type"] + assert rc.version == api_response["version"] + assert rc.display_name == api_response["displayName"] + + +def test_resource_category_get_subcategory(): + rc = ResourceCategory( + name=str(uuid4()), + empty=bool(randint(0, 1)), + icons=[str(uuid4())], + models=[str(uuid4())], + normalized_name=str(uuid4()), + unique_id=str(uuid4()), + use_service_substitution_for_nested_services=bool(randint(0, 1)), + subcategories=None + ) + assert rc.get_subcategory(subcategory_name="test_subc") is None + rc.subcategories = [SdcSubCategory( + name=f"subcategory_{i}", + normalized_name=f"subcategory_{i}", + unique_id=str(uuid4()) + ) for i in range(10**2)] + assert rc.get_subcategory(subcategory_name="test_subc") is None + for i in range(10**2): + assert rc.get_subcategory(subcategory_name=f"subcategory_{i}") is not None + + +# Product categories tests +@patch("onapsdk.sdc2.sdc_category.ProductCategory.send_message_json") +@patch("onapsdk.sdc2.sdc_category.ProductCategory.create_from_api_response") +def test_product_category_get_all(mock_create_from_api_response, mock_send_message_json): + mock_send_message_json.return_value = [] + assert len(list(ProductCategory.get_all())) == 0 + mock_send_message_json.assert_called_once_with( + "GET", + "Get all ProductCategory", + f"{settings.SDC_BE_URL}/sdc2/rest/v1/categories/products" + ) + + mock_send_message_json.return_value = [{}] + assert len(list(ProductCategory.get_all())) == 1 + mock_create_from_api_response.assert_called_once_with({}) + + +@patch("onapsdk.sdc2.sdc_category.ProductCategory.get_all") +def test_product_category_get_by_unique_id(mock_get_all): + mock_get_all.return_value = [] + with raises(ResourceNotFound): + ProductCategory.get_by_uniqe_id("test_unique_id") + + TestCategory = namedtuple("TestCategory", "unique_id") + mock_get_all.return_value = [TestCategory(f"unique_id_{i}") for i in range(10**2)] + with raises(ResourceNotFound): + ProductCategory.get_by_uniqe_id("test_unique_id") + for i in range(10**2): + assert ProductCategory.get_by_uniqe_id(f"unique_id_{i}") is not None + + +@patch("onapsdk.sdc2.sdc_category.ProductCategory.get_all") +def test_product_category_get_by_name(mock_get_all): + mock_get_all.return_value = [] + with raises(ResourceNotFound): + ProductCategory.get_by_name("test_name") + + TestCategory = namedtuple("TestCategory", "name") + mock_get_all.return_value = [TestCategory(f"name_{i}") for i in range(10**2)] + with raises(ResourceNotFound): + ProductCategory.get_by_name("test_name") + for i in range(10**2): + assert ProductCategory.get_by_name(f"name_{i}") is not None + + +def test_product_category_create_from_api_response(): + api_response = { + "name": str(uuid4()), + "icons": [str(uuid4())], + "models": [str(uuid4())], + "normalizedName": str(uuid4()), + "uniqueId": str(uuid4()), + "ownerId": str(uuid4()), + "type": str(uuid4()), + "version": str(uuid4()), + "displayName": str(uuid4()), + "empty": bool(randint(0, 1)), + "useServiceSubstitutionForNestedServices": bool(randint(0, 1)), + } + pc = ProductCategory.create_from_api_response(api_response) + assert pc.name == api_response["name"] + assert pc.empty == api_response["empty"] + assert pc.icons == api_response["icons"] + assert pc.models == api_response["models"] + assert pc.normalized_name == api_response["normalizedName"] + assert pc.unique_id == api_response["uniqueId"] + assert pc.use_service_substitution_for_nested_services == api_response["useServiceSubstitutionForNestedServices"] + assert pc.owner_id == api_response["ownerId"] + assert pc.subcategories == None + assert pc.category_type == api_response["type"] + assert pc.version == api_response["version"] + assert pc.display_name == api_response["displayName"] + + + api_response = { + "name": str(uuid4()), + "icons": [str(uuid4())], + "models": [str(uuid4())], + "normalizedName": str(uuid4()), + "uniqueId": str(uuid4()), + "ownerId": str(uuid4()), + "type": str(uuid4()), + "version": str(uuid4()), + "displayName": str(uuid4()), + "empty": bool(randint(0, 1)), + "useServiceSubstitutionForNestedServices": bool(randint(0, 1)), + "subcategories": [ + { + "name": str(uuid4()), + "normalizedName": str(uuid4()), + "uniqueId": str(uuid4()), + }, + { + "name": str(uuid4()), + "normalizedName": str(uuid4()), + "uniqueId": str(uuid4()), + } + ] + } + pc = ProductCategory.create_from_api_response(api_response) + assert pc.name == api_response["name"] + assert pc.empty == api_response["empty"] + assert pc.icons == api_response["icons"] + assert pc.models == api_response["models"] + assert pc.normalized_name == api_response["normalizedName"] + assert pc.unique_id == api_response["uniqueId"] + assert pc.use_service_substitution_for_nested_services == api_response["useServiceSubstitutionForNestedServices"] + assert pc.owner_id == api_response["ownerId"] + assert len(pc.subcategories) == 2 + assert pc.subcategories[0].name == api_response["subcategories"][0]["name"] + assert pc.subcategories[0].normalized_name == api_response["subcategories"][0]["normalizedName"] + assert pc.subcategories[0].unique_id == api_response["subcategories"][0]["uniqueId"] + assert pc.subcategories[1].name == api_response["subcategories"][1]["name"] + assert pc.subcategories[1].normalized_name == api_response["subcategories"][1]["normalizedName"] + assert pc.subcategories[1].unique_id == api_response["subcategories"][1]["uniqueId"] + assert pc.category_type == api_response["type"] + assert pc.version == api_response["version"] + assert pc.display_name == api_response["displayName"] + + +def test_product_category_get_subcategory(): + pc = ProductCategory( + name=str(uuid4()), + empty=bool(randint(0, 1)), + icons=[str(uuid4())], + models=[str(uuid4())], + normalized_name=str(uuid4()), + unique_id=str(uuid4()), + use_service_substitution_for_nested_services=bool(randint(0, 1)), + subcategories=None + ) + assert pc.get_subcategory(subcategory_name="test_subc") is None + pc.subcategories = [SdcSubCategory( + name=f"subcategory_{i}", + normalized_name=f"subcategory_{i}", + unique_id=str(uuid4()) + ) for i in range(10**2)] + assert pc.get_subcategory(subcategory_name="test_subc") is None + for i in range(10**2): + assert pc.get_subcategory(subcategory_name=f"subcategory_{i}") is not None diff --git a/tests/test_sdc2_component_instance.py b/tests/test_sdc2_component_instance.py new file mode 100644 index 0000000..e5fac5a --- /dev/null +++ b/tests/test_sdc2_component_instance.py @@ -0,0 +1,203 @@ + +from collections import namedtuple +from json import loads +from random import randint +from sys import maxsize +from unittest.mock import MagicMock, patch, PropertyMock +from uuid import uuid4 + +from onapsdk.configuration import settings +from onapsdk.sdc2.component_instance import ComponentInstance, ComponentInstanceInput + + +def test_component_instance_create_from_api_response(): + data = { + "actualComponentUid": str(uuid4()), + "componentName": str(uuid4()), + "componentUid": str(uuid4()), + "componentVersion": str(uuid4()), + "creationTime": randint(0, maxsize), + "customizationUUID": str(uuid4()), + "icon": str(uuid4()), + "invariantName": str(uuid4()), + "isProxy": bool(randint(0, 1)), + "modificationTime": randint(0, maxsize), + "name": str(uuid4()), + "normalizedName": str(uuid4()), + "originType": str(uuid4()), + "toscaComponentName": str(uuid4()), + "uniqueId": str(uuid4()), + } + ci = ComponentInstance.create_from_api_response( + data, sdc_resource=MagicMock() + ) + assert ci.name == data["name"] + assert ci.actual_component_uid == data["actualComponentUid"] + assert ci.component_name == data["componentName"] + assert ci.component_uid == data["componentUid"] + assert ci.component_version == data["componentVersion"] + assert ci.creation_time == data["creationTime"] + assert ci.customization_uuid == data["customizationUUID"] + assert ci.icon == data["icon"] + assert ci.invariant_name == data["invariantName"] + assert ci.is_proxy == data["isProxy"] + assert ci.modification_time == data["modificationTime"] + assert ci.normalized_name == data["normalizedName"] + assert ci.origin_type == data["originType"] + assert ci.tosca_component_name == data["toscaComponentName"] + assert ci.unique_id == data["uniqueId"] + + +@patch("onapsdk.sdc2.component_instance.ComponentInstanceInput.create_from_api_response") +@patch("onapsdk.sdc2.component_instance.ComponentInstance.send_message_json") +def test_component_instance_inputs(mock_send_message_json, mock_input_create_from_api_response): + sdc_resource_mock = MagicMock(unique_id=str(uuid4())) + sdc_resource_mock.catalog_type.return_value = "mocked" + ci = ComponentInstance( + actual_component_uid=str(uuid4()), + component_name=str(uuid4()), + component_uid=str(uuid4()), + component_version=str(uuid4()), + creation_time=randint(0, maxsize), + customization_uuid=str(uuid4()), + icon=str(uuid4()), + invariant_name=str(uuid4()), + is_proxy=bool(randint(0, 1)), + modification_time=randint(0, maxsize), + name=str(uuid4()), + normalized_name=str(uuid4()), + origin_type=str(uuid4()), + sdc_resource=sdc_resource_mock, + tosca_component_name=str(uuid4()), + unique_id=str(uuid4()) + ) + mock_send_message_json.return_value = [] + assert len(list(ci.inputs)) == 0 + mock_send_message_json.assert_called_once_with( + "GET", + "Get inputs", + f"{settings.SDC_BE_URL}/sdc2/rest/v1/catalog/mocked/{sdc_resource_mock.unique_id}/componentInstances/{ci.unique_id}/{ci.actual_component_uid}/inputs" + ) + + mock_send_message_json.reset_mock() + mock_send_message_json.return_value = [{}] + assert len(list(ci.inputs)) == 1 + mock_input_create_from_api_response.assert_called_once_with({}, ci) + mock_send_message_json.assert_called_once_with( + "GET", + "Get inputs", + f"{settings.SDC_BE_URL}/sdc2/rest/v1/catalog/mocked/{sdc_resource_mock.unique_id}/componentInstances/{ci.unique_id}/{ci.actual_component_uid}/inputs" + ) + + +@patch("onapsdk.sdc2.component_instance.ComponentInstance.inputs", new_callable=PropertyMock) +def test_get_component_input_by_name(mock_inputs): + ci = ComponentInstance( + actual_component_uid=str(uuid4()), + component_name=str(uuid4()), + component_uid=str(uuid4()), + component_version=str(uuid4()), + creation_time=randint(0, maxsize), + customization_uuid=str(uuid4()), + icon=str(uuid4()), + invariant_name=str(uuid4()), + is_proxy=bool(randint(0, 1)), + modification_time=randint(0, maxsize), + name=str(uuid4()), + normalized_name=str(uuid4()), + origin_type=str(uuid4()), + sdc_resource=MagicMock(), + tosca_component_name=str(uuid4()), + unique_id=str(uuid4()) + ) + + mock_inputs.return_value = [] + assert ci.get_input_by_name("test_name") is None + + Input = namedtuple("Input", ["name"]) + mock_inputs.return_value = [Input("test_name")] + assert ci.get_input_by_name("test_name") is not None + + mock_inputs.return_value = [Input(f"test_name_{i}") for i in range(10**2)] + assert ci.get_input_by_name("test_name") is None + + mock_inputs.return_value = [Input(f"test_name_{i}") for i in range(10**2)] + for i in range(10**2): + assert ci.get_input_by_name(f"test_name_{i}") is not None + + +def test_component_instance_input_create_from_api_response(): + api_response={ + "definition": bool(randint(0, 1)), + "hidden": bool(randint(0, 1)), + "uniqueId": str(uuid4()), + "type": str(uuid4()), + "required": bool(randint(0, 1)), + "password": bool(randint(0, 1)), + "name": str(uuid4()), + "immutable": bool(randint(0, 1)), + "mappedToComponentProperty": bool(randint(0, 1)), + "isDeclaredListInput": bool(randint(0, 1)), + "userCreated": bool(randint(0, 1)), + "getInputProperty": bool(randint(0, 1)), + "empty": bool(randint(0, 1)) + } + cli = ComponentInstanceInput.create_from_api_response( + component_instance=MagicMock(), + api_response=api_response + ) + assert cli.name == api_response["name"] + assert cli.definition == api_response["definition"] + assert cli.hidden == api_response["hidden"] + assert cli.unique_id == api_response["uniqueId"] + assert cli.input_type == api_response["type"] + assert cli.required == api_response["required"] + assert cli.password == api_response["password"] + assert cli.immutable == api_response["immutable"] + assert cli.mapped_to_component_property == api_response["mappedToComponentProperty"] + assert cli.is_declared_list_input == api_response["isDeclaredListInput"] + assert cli.user_created == api_response["userCreated"] + assert cli.get_input_property == api_response["getInputProperty"] + assert cli.empty == api_response["empty"] + assert cli.description is None + assert cli.label is None + assert cli.value is None + + +@patch("onapsdk.sdc2.component_instance.ComponentInstanceInput.send_message_json") +def test_component_instance_input_set_value(mock_send_message_json): + sdc_resource_mock = MagicMock() + sdc_resource_mock.name = "mocked sdc resource" + sdc_resource_mock.catalog_type.return_value = "mocked" + sdc_resource_mock.unique_id = "mockUid" + component_instance_mock = MagicMock(sdc_resource=sdc_resource_mock) + component_instance_mock.unique_id = "mockUid" + cii = ComponentInstanceInput( + component_instance=component_instance_mock, + name=str(uuid4()), + definition=bool(randint(0, 1)), + hidden=bool(randint(0, 1)), + required=bool(randint(0, 1)), + password=bool(randint(0, 1)), + immutable=bool(randint(0, 1)), + mapped_to_component_property=bool(randint(0, 1)), + is_declared_list_input=bool(randint(0, 1)), + user_created=bool(randint(0, 1)), + get_input_property=bool(randint(0, 1)), + empty=bool(randint(0, 1)), + unique_id=str(uuid4()), + input_type=str(uuid4()) + ) + cii.value = "!23" + mock_send_message_json.assert_called_once() + method, log, url = mock_send_message_json.mock_calls[0].args + data = loads(mock_send_message_json.mock_calls[0].kwargs["data"]) + assert method == "POST" + assert log == f"Set value of mocked sdc resource resource input {cii.name}" + assert url == f"{settings.SDC_BE_URL}/sdc2/rest/v1/catalog/mocked/mockUid/resourceInstance/mockUid/inputs" + assert data[0]["name"] == cii.name + assert data[0]["parentUniqueId"] == component_instance_mock.unique_id + assert data[0]["type"] == cii.input_type + assert data[0]["uniqueId"] == cii.unique_id + assert data[0]["value"] == "!23" + assert data[0]["toscaPresentation"]["ownerId"] == component_instance_mock.unique_id diff --git a/tests/test_sdc2_pnf.py b/tests/test_sdc2_pnf.py new file mode 100644 index 0000000..07834a5 --- /dev/null +++ b/tests/test_sdc2_pnf.py @@ -0,0 +1,413 @@ + +import json +from collections import namedtuple +from random import choice, randint +from sys import maxsize +from tempfile import NamedTemporaryFile +from unittest.mock import MagicMock, patch, PropertyMock +from uuid import uuid4 + +from pytest import raises + +from onapsdk.exceptions import ResourceNotFound +from onapsdk.sdc2.sdc_resource import LifecycleState, LifecycleOperation +from onapsdk.sdc2.pnf import Pnf, ResoureTypeEnum + + +def test_pnf_resource_type(): + assert Pnf.resource_type() == ResoureTypeEnum.PNF + + +def test_pnf_create_payload_template(): + assert Pnf.create_payload_template() == "sdc2_create_pnf.json.j2" + + +def test_copy_pnf(): + pnf1 = Pnf(name="test_pnf1") + pnf2 = Pnf(name="test_pnf2") + assert pnf1.name == "test_pnf1" + assert pnf2.name == "test_pnf2" + assert pnf1 != pnf2 + pnf2._copy_object(pnf1) + assert pnf1.name == "test_pnf1" + assert pnf2.name == "test_pnf1" + assert pnf1 == pnf2 + + +@patch("onapsdk.sdc2.pnf.Pnf._get_active_rough") +@patch("onapsdk.sdc2.pnf.Pnf._get_archived_rough") +def test_get_all_rough(mock_get_archived, mock_get_active): + Pnf._get_all_rough() + mock_get_archived.assert_called_once() + mock_get_active.assert_called_once() + + +@patch("onapsdk.sdc2.pnf.Pnf._get_all_rough") +@patch("onapsdk.sdc2.pnf.Pnf.get_by_name_and_version") +def test_get_by_name(mock_get_by_name_and_version, mock_get_all_rough): + mock_get_all_rough.return_value = [] + with raises(ResourceNotFound): + Pnf.get_by_name("test_1") + + mock_get_all_rough.return_value = [ + { + "name": "not_test_1", + "version": "1.0" + } + ] + with raises(ResourceNotFound): + Pnf.get_by_name("test_1") + + mock_get_all_rough.return_value = [ + { + "name": "not_test_1", + "version": "1.0" + }, + { + "name": "test_1", + "version": "1.0" + } + ] + Pnf.get_by_name("test_1") + mock_get_by_name_and_version.assert_called_once_with("test_1", "1.0") + + mock_get_by_name_and_version.reset_mock() + mock_get_all_rough.return_value = [ + { + "name": "not_test_1", + "version": "1.0" + }, + { + "name": "test_1", + "version": "1.0" + }, + { + "name": "test_1", + "version": "2.0" + } + ] + Pnf.get_by_name("test_1") + mock_get_by_name_and_version.assert_called_once_with("test_1", "2.0") + + +@patch("onapsdk.sdc2.pnf.Pnf.send_message") +def test_delete(mock_send_message_json): + pnf = Pnf(name="test_pnf", unique_id="test_pnf_unique_id") + pnf.delete() + mock_send_message_json.assert_called_once() + method, log, url = mock_send_message_json.mock_calls[0].args + assert method == "DELETE" + assert "test_pnf" in log + assert url.endswith("test_pnf_unique_id") + + +@patch("onapsdk.sdc2.pnf.Pnf.send_message") +def test_archive(mock_send_message_json): + pnf = Pnf(name="test_pnf", unique_id="test_pnf_unique_id") + pnf.archive() + mock_send_message_json.assert_called_once() + method, log, url = mock_send_message_json.mock_calls[0].args + assert method == "POST" + assert "test_pnf" in log + assert url.endswith("test_pnf_unique_id/archive") + + +@patch("onapsdk.sdc2.pnf.Pnf._get_all_rough") +@patch("onapsdk.sdc2.pnf.Pnf.get_by_name_and_version") +def test_get_all(mock_get_by_name_and_version, mock_get_all_rough): + mock_get_all_rough.return_value = [] + Pnf.get_all() + mock_get_by_name_and_version.assert_not_called() + + mock_get_all_rough.return_value = [ + { + "name": "test_pnf_1", + "version": "1.0" + } + ] + list(Pnf.get_all()) + mock_get_by_name_and_version.assert_called_once_with("test_pnf_1", "1.0") + + mock_get_by_name_and_version.reset_mock() + mock_get_all_rough.return_value = ( + { + "name": f"test_pnf_{idx}", + "version": f"{idx}.0" + } for idx in range(100) + ) + list(Pnf.get_all()) + assert len(mock_get_by_name_and_version.mock_calls) == 100 + + +@patch("onapsdk.sdc2.pnf.Pnf.send_message_json") +def test_get_active_rough(mock_send_message_json): + mock_send_message_json.return_value = {"resources": []} + assert(len(list(Pnf._get_active_rough()))) == 0 + + mock_send_message_json.return_value = {"resources": [ + { + "resourceType": "VF" + }, + { + "resourceType": "PNF" + }, + { + "resourceType": "VL" + } + ]} + assert(len(list(Pnf._get_active_rough()))) == 1 + + +@patch("onapsdk.sdc2.pnf.Pnf.send_message_json") +def test_get_archive_rough(mock_send_message_json): + mock_send_message_json.return_value = {"resources": []} + assert(len(list(Pnf._get_archived_rough()))) == 0 + + mock_send_message_json.return_value = {"resources": [ + { + "resourceType": "VF" + }, + { + "resourceType": "PNF" + }, + { + "resourceType": "VL" + } + ]} + assert(len(list(Pnf._get_archived_rough()))) == 1 + + +def test_get_by_name_and_version_endpoint(): + for i in range(1000): + assert Pnf.get_by_name_and_version_endpoint( + name=f"pnf_{i}", + version=f"v{i}.0").endswith( + f"catalog/resources/resourceName/pnf_{i}/resourceVersion/v{i}.0") + + +def test_add_deployment_artifact_endpoint(): + for _ in range(10**3): + object_id: str = str(uuid4()) + assert Pnf.add_deployment_artifact_endpoint( + object_id=object_id + ) == f"sdc2/rest/v1/catalog/resources/{object_id}/artifacts" + + +def test_update(): + pnf = Pnf( + name=str(uuid4()), + unique_id=str(uuid4()), + uuid=str(uuid4()), + invariant_uuid=str(uuid4()), + version=str(uuid4()), + last_update_date=randint(0, maxsize), + lifecycle_state=choice(list(LifecycleState)), + last_updater_user_id=str(uuid4()), + all_versions=[str(uuid4()) for _ in range(10**3)] + ) + update_dict = { + "uniqueId": str(uuid4()), + "uuid": str(uuid4()), + "invariantUUID": str(uuid4()), + "version": str(uuid4()), + "lastUpdateDate": randint(0, maxsize), + "lifecycleState": choice(list(LifecycleState)), + "lastUpdaterUserId": str(uuid4()), + "allVersions": [str(uuid4()) for _ in range(10**3)] + } + pnf.update(update_dict) + assert pnf.unique_id == update_dict["uniqueId"] + assert pnf.uuid == update_dict["uuid"] + assert pnf.invariant_uuid == update_dict["invariantUUID"] + assert pnf.version == update_dict["version"] + assert pnf.last_update_date == update_dict["lastUpdateDate"] + assert pnf.lifecycle_state == update_dict["lifecycleState"] + assert pnf.last_updater_user_id == update_dict["lastUpdaterUserId"] + assert pnf.all_versions == update_dict["allVersions"] + + +@patch("onapsdk.sdc2.pnf.Pnf.send_message_json") +@patch("onapsdk.sdc2.pnf.Pnf.update") +def test_lifecycle_operation(mock_update, mock_send_message_json): + for lifecycle_operation in LifecycleOperation: + mock_send_message_json.reset_mock() + mock_update.reset_mock() + return_dict = { + "uniqueId": str(uuid4()), + "uuid": str(uuid4()), + "invariantUUID": str(uuid4()), + "version": str(uuid4()), + "lastUpdateDate": randint(0, maxsize), + "lifecycleState": choice(list(LifecycleState)), + "lastUpdaterUserId": str(uuid4()), + "allVersions": [str(uuid4()) for _ in range(10**3)] + } + mock_send_message_json.return_value = return_dict + pnf_unique_id = str(uuid4()) + pnf = Pnf( + name=str(uuid4()), + unique_id=pnf_unique_id + ) + pnf.lifecycle_operation(lifecycle_operation) + mock_send_message_json.assert_called_once() + method, log, url = mock_send_message_json.mock_calls[0].args + data = mock_send_message_json.mock_calls[0].kwargs["data"] + assert method == "POST" + assert log.startswith(f"Request lifecycle operation {lifecycle_operation}") + assert url.endswith(f"sdc2/rest/v1/catalog/resources/{pnf_unique_id}/lifecycleState/{lifecycle_operation}") + assert json.loads(data)["userRemarks"] == str(lifecycle_operation).lower() + mock_update.assert_called_once_with(return_dict) + + +@patch("onapsdk.sdc2.pnf.Pnf.send_message_json") +@patch("onapsdk.sdc2.pnf.Pnf.get_by_name_and_version_endpoint") +def test_get_by_name_and_version(mock_get_endpoint, mock_send_message_json): + for name, version in [(str(uuid4()), str(uuid4())) for _ in range(10**2)]: + mock_send_message_json.reset_mock() + mock_send_message_json.return_value = { + "uniqueId": str(uuid4()), + "uuid": str(uuid4()), + "invariantUUID": str(uuid4()), + "version": str(uuid4()), + "lastUpdateDate": randint(0, maxsize), + "lifecycleState": choice(list(LifecycleState)), + "lastUpdaterUserId": str(uuid4()), + "allVersions": [str(uuid4()) for _ in range(10**2)], + "archived": choice([True, False]), + "creationDate": randint(0, maxsize), + "componentType": str(uuid4()), + "description": str(uuid4()), + "icon": str(uuid4()), + "name": str(uuid4()), + "systemName": str(uuid4()), + "tags": [str(uuid4()) for _ in range(10**2)], + } + mock_get_endpoint.reset_mock() + mock_get_endpoint.return_value = f"{name}/{version}" + Pnf.get_by_name_and_version(name, version) + mock_send_message_json.assert_called_once() + method, log, _ = mock_send_message_json.mock_calls[0].args + assert method == "GET" + assert log == "Get Pnf by name and version" + mock_get_endpoint.assert_called_once_with(name, version) + + +@patch("onapsdk.sdc2.pnf.Pnf.send_message_json") +def test_add_deployment_artifact(mock_send_message_json): + with NamedTemporaryFile() as temp_file: + temp_file.write(b"Hello world!") + temp_file.seek(0) + + pnf = Pnf(name=str(uuid4()), unique_id=str(uuid4())) + pnf.add_deployment_artifact( + str(uuid4()), + str(uuid4()), + str(uuid4()), + temp_file.name + ) + mock_send_message_json.assert_called_once() + + +@patch("onapsdk.sdc2.pnf.Pnf.send_message_json") +@patch("onapsdk.sdc2.sdc_resource.ComponentInstance.create_from_api_response") +def test_add_deployment_artifact(mock_create_component_instance, mock_send_message_json): + pnf = Pnf(name=str(uuid4()), unique_id=str(uuid4())) + mock_send_message_json.return_value = [] + assert len(list(pnf.component_instances)) == 0 + mock_send_message_json.assert_called_once_with( + "GET", + "Get Pnf component instances", + f"{pnf.base_back_url}/sdc2/rest/v1/catalog/resources/{pnf.unique_id}/componentInstances" + ) + + mock_send_message_json.return_value = [{}] + assert len(list(pnf.component_instances)) == 1 + mock_create_component_instance.assert_called_once_with({}, pnf) + + +@patch("onapsdk.sdc2.pnf.Pnf.component_instances", new_callable=PropertyMock) +def test_get_component_by_name(mock_component_instances): + ComponentInstance = namedtuple("ComponentInstance", ["component_name"]) + + pnf = Pnf(name="test_component_instances") + mock_component_instances.return_value = [] + assert pnf.get_component_by_name("test_name") is None + + mock_component_instances.return_value = [ComponentInstance(component_name="test_name")] + assert pnf.get_component_by_name("test_name") is not None + assert pnf.get_component_by_name("test_name").component_name == "test_name" + + mock_component_instances.return_value = [ComponentInstance(component_name=f"test_name_{i}") for i in range(10**2)] + assert pnf.get_component_by_name("test_name") is None + + random_name = f"test_name_{choice(range(10**2))}" + assert pnf.get_component_by_name(random_name) is not None + assert pnf.get_component_by_name(random_name).component_name == random_name + + +@patch("onapsdk.sdc2.pnf.Pnf.send_message_json") +@patch("onapsdk.sdc2.pnf.Pnf.get_create_payload") +@patch("onapsdk.sdc2.pnf.Pnf.create_from_api_response") +def test_create(mock_create_from_api_response, mock_get_create_payload, mock_send_message_json): + mock_get_create_payload.return_value = "test_payload" + Pnf.create(name="test_pnf") + mock_send_message_json.assert_called_once_with( + "POST", + "Create PNF test_pnf", + Pnf.CREATE_ENDPOINT, + data="test_payload" + ) + + +@patch("onapsdk.sdc2.sdc_resource.ResourceCategory.get_by_name") +def test_get_create_payload(mock_resource_category_get_by_name): + mock_vsp = MagicMock(csar_uuid="test_vsp_csar_uuid") + mock_vendor = MagicMock() + mock_vendor.name = "test_vendor_name" + resource_category_mock = MagicMock() + resource_category_mock.name = "test_category" + resource_category_mock.unique_id = "category_unique_id" + resource_subcategory_mock = MagicMock() + resource_subcategory_mock.name = "test_subcategory" + resource_subcategory_mock.unique_id = "subcategory_unique_id" + resource_category_mock.get_subcategory.return_value = resource_subcategory_mock + mock_resource_category_get_by_name.return_value = resource_category_mock + + create_payload = json.loads(Pnf.get_create_payload( + name="test_pnf", + vsp=mock_vsp, + vendor=mock_vendor + )) + assert create_payload["name"] == "test_pnf" + assert create_payload["contactId"] == "cs0008" + assert create_payload["componentType"] == "RESOURCE" + assert create_payload["csarUUID"] == "test_vsp_csar_uuid" + assert create_payload["csarVersion"] == "1.0" + assert create_payload["categories"][0]["name"] == "test_category" + assert create_payload["categories"][0]["uniqueId"] == "category_unique_id" + assert create_payload["categories"][0]["subcategories"][0]["name"] == "test_subcategory" + assert create_payload["categories"][0]["subcategories"][0]["uniqueId"] == "subcategory_unique_id" + assert create_payload["resourceType"] == "PNF" + assert create_payload["description"] == "ONAP SDK Resource" + assert create_payload["vendorName"] == "test_vendor_name" + assert create_payload["vendorRelease"] == "1.0" + + create_payload = json.loads(Pnf.get_create_payload( + name="test_pnf", + vsp=mock_vsp, + vendor=mock_vendor, + description="test description" + )) + assert create_payload["name"] == "test_pnf" + assert create_payload["contactId"] == "cs0008" + assert create_payload["componentType"] == "RESOURCE" + assert create_payload["csarUUID"] == "test_vsp_csar_uuid" + assert create_payload["csarVersion"] == "1.0" + assert create_payload["categories"][0]["name"] == "test_category" + assert create_payload["categories"][0]["uniqueId"] == "category_unique_id" + assert create_payload["categories"][0]["subcategories"][0]["name"] == "test_subcategory" + assert create_payload["categories"][0]["subcategories"][0]["uniqueId"] == "subcategory_unique_id" + assert create_payload["resourceType"] == "PNF" + assert create_payload["description"] == "test description" + assert create_payload["vendorName"] == "test_vendor_name" + assert create_payload["vendorRelease"] == "1.0" diff --git a/tests/test_sdc2_resource.py b/tests/test_sdc2_resource.py new file mode 100644 index 0000000..e4d25c1 --- /dev/null +++ b/tests/test_sdc2_resource.py @@ -0,0 +1,14 @@ + +from urllib.parse import parse_qs + +from pytest import raises + +from onapsdk.sdc2.sdc import ResoureTypeEnum +from onapsdk.sdc2.sdc_resource import SDCResource + +def test_build_exclude_types_query(): + for resource_type in ResoureTypeEnum: + query = SDCResource._build_exclude_types_query(resource_type) + assert query.count("excludeTypes=") == 12 + with raises(ValueError): + parse_qs(query)["excludeTypes"].index(resource_type.value) diff --git a/tests/test_sdc2_service.py b/tests/test_sdc2_service.py new file mode 100644 index 0000000..59e18b5 --- /dev/null +++ b/tests/test_sdc2_service.py @@ -0,0 +1,604 @@ + +import json +from collections import namedtuple +from random import choice, randint +from sys import maxsize +from tempfile import NamedTemporaryFile +from unittest.mock import MagicMock, patch, PropertyMock +from uuid import uuid4 + +from pytest import raises + +from onapsdk.configuration import settings +from onapsdk.exceptions import ResourceNotFound +from onapsdk.sdc2.sdc_resource import LifecycleState, LifecycleOperation +from onapsdk.sdc2.service import ServiceInstantiationType, Service, ResoureTypeEnum, ServiceDistribution + + +def test_service_resource_type(): + assert Service.resource_type() == ResoureTypeEnum.SERVICE + + +def test_copy_service(): + service1 = Service(name="test_service1") + service2 = Service(name="test_service2") + assert service1.name == "test_service1" + assert service2.name == "test_service2" + assert service1 != service2 + service2._copy_object(service1) + assert service1.name == "test_service1" + assert service2.name == "test_service1" + assert service1 == service2 + + +@patch("onapsdk.sdc2.service.Service._get_active_rough") +@patch("onapsdk.sdc2.service.Service._get_archived_rough") +def test_get_all_rough(mock_get_archived, mock_get_active): + Service._get_all_rough() + mock_get_archived.assert_called_once() + mock_get_active.assert_called_once() + + +@patch("onapsdk.sdc2.service.Service._get_all_rough") +@patch("onapsdk.sdc2.service.Service.get_by_name_and_version") +def test_get_by_name(mock_get_by_name_and_version, mock_get_all_rough): + mock_get_all_rough.return_value = [] + with raises(ResourceNotFound): + Service.get_by_name("test_1") + + mock_get_all_rough.return_value = [ + { + "name": "not_test_1", + "version": "1.0" + } + ] + with raises(ResourceNotFound): + Service.get_by_name("test_1") + + mock_get_all_rough.return_value = [ + { + "name": "not_test_1", + "version": "1.0" + }, + { + "name": "test_1", + "version": "1.0" + } + ] + Service.get_by_name("test_1") + mock_get_by_name_and_version.assert_called_once_with("test_1", "1.0") + + mock_get_by_name_and_version.reset_mock() + mock_get_all_rough.return_value = [ + { + "name": "not_test_1", + "version": "1.0" + }, + { + "name": "test_1", + "version": "1.0" + }, + { + "name": "test_1", + "version": "2.0" + } + ] + Service.get_by_name("test_1") + mock_get_by_name_and_version.assert_called_once_with("test_1", "2.0") + + +@patch("onapsdk.sdc2.service.Service.send_message") +def test_delete(mock_send_message_json): + service = Service(name="test_service", unique_id="test_service_unique_id") + service.delete() + mock_send_message_json.assert_called_once() + method, log, url = mock_send_message_json.mock_calls[0].args + assert method == "DELETE" + assert "test_service" in log + assert url.endswith("test_service_unique_id") + + +@patch("onapsdk.sdc2.service.Service.send_message") +def test_archive(mock_send_message_json): + service = Service(name="test_service", unique_id="test_service_unique_id") + service.archive() + mock_send_message_json.assert_called_once() + method, log, url = mock_send_message_json.mock_calls[0].args + assert method == "POST" + assert "test_service" in log + assert url.endswith("test_service_unique_id/archive") + + +@patch("onapsdk.sdc2.service.Service._get_all_rough") +@patch("onapsdk.sdc2.service.Service.get_by_name_and_version") +def test_get_all(mock_get_by_name_and_version, mock_get_all_rough): + mock_get_all_rough.return_value = [] + Service.get_all() + mock_get_by_name_and_version.assert_not_called() + + mock_get_all_rough.return_value = [ + { + "name": "test_service_1", + "version": "1.0" + } + ] + list(Service.get_all()) + mock_get_by_name_and_version.assert_called_once_with("test_service_1", "1.0") + + mock_get_by_name_and_version.reset_mock() + mock_get_all_rough.return_value = ( + { + "name": f"test_service_{idx}", + "version": f"{idx}.0" + } for idx in range(100) + ) + list(Service.get_all()) + assert len(mock_get_by_name_and_version.mock_calls) == 100 + + +@patch("onapsdk.sdc2.service.Service.send_message_json") +def test_get_active_rough(mock_send_message_json): + mock_send_message_json.return_value = {"resources": []} + assert(len(list(Service._get_active_rough()))) == 0 + + mock_send_message_json.return_value = { + "resources": [ + { + "resourceType": "VF" + }, + { + "resourceType": "PNF" + }, + { + "resourceType": "VL" + } + ], + "services": [ + { + "resourceType": "SERVICE" + } + ] + } + assert(len(list(Service._get_active_rough()))) == 1 + + +@patch("onapsdk.sdc2.service.Service.send_message_json") +def test_get_archive_rough(mock_send_message_json): + mock_send_message_json.return_value = {"resources": []} + assert(len(list(Service._get_archived_rough()))) == 0 + + mock_send_message_json.return_value = { + "resources": [ + { + "resourceType": "VF" + }, + { + "resourceType": "PNF" + }, + { + "resourceType": "VL" + } + ], + "services": [ + { + "resourceType": "SERVICE" + } + ] + } + assert(len(list(Service._get_archived_rough()))) == 1 + + +def test_get_by_name_and_version_endpoint(): + for i in range(1000): + assert Service.get_by_name_and_version_endpoint( + name=f"service_{i}", + version=f"v{i}.0").endswith( + f"catalog/services/serviceName/service_{i}/serviceVersion/v{i}.0") + + +def test_add_deployment_artifact_endpoint(): + for _ in range(10**3): + object_id: str = str(uuid4()) + assert Service.add_deployment_artifact_endpoint( + object_id=object_id + ) == f"sdc2/rest/v1/catalog/services/{object_id}/artifacts" + + +def test_update(): + service = Service( + name=str(uuid4()), + unique_id=str(uuid4()), + uuid=str(uuid4()), + invariant_uuid=str(uuid4()), + version=str(uuid4()), + last_update_date=randint(0, maxsize), + lifecycle_state=choice(list(LifecycleState)), + last_updater_user_id=str(uuid4()), + all_versions=[str(uuid4()) for _ in range(10**3)] + ) + update_dict = { + "uniqueId": str(uuid4()), + "uuid": str(uuid4()), + "invariantUUID": str(uuid4()), + "version": str(uuid4()), + "lastUpdateDate": randint(0, maxsize), + "lifecycleState": choice(list(LifecycleState)), + "lastUpdaterUserId": str(uuid4()), + "allVersions": [str(uuid4()) for _ in range(10**3)], + "distributionStatus": str(uuid4()) + } + service.update(update_dict) + assert service.unique_id == update_dict["uniqueId"] + assert service.uuid == update_dict["uuid"] + assert service.invariant_uuid == update_dict["invariantUUID"] + assert service.version == update_dict["version"] + assert service.last_update_date == update_dict["lastUpdateDate"] + assert service.lifecycle_state == update_dict["lifecycleState"] + assert service.last_updater_user_id == update_dict["lastUpdaterUserId"] + assert service.all_versions == update_dict["allVersions"] + assert service.distribuition_status == update_dict["distributionStatus"] + + +@patch("onapsdk.sdc2.service.Service.send_message_json") +@patch("onapsdk.sdc2.service.Service.update") +def test_lifecycle_operation(mock_update, mock_send_message_json): + for lifecycle_operation in LifecycleOperation: + mock_send_message_json.reset_mock() + mock_update.reset_mock() + return_dict = { + "uniqueId": str(uuid4()), + "uuid": str(uuid4()), + "invariantUUID": str(uuid4()), + "version": str(uuid4()), + "lastUpdateDate": randint(0, maxsize), + "lifecycleState": choice(list(LifecycleState)), + "lastUpdaterUserId": str(uuid4()), + "allVersions": [str(uuid4()) for _ in range(10**3)] + } + mock_send_message_json.return_value = return_dict + service_unique_id = str(uuid4()) + service = Service( + name=str(uuid4()), + unique_id=service_unique_id + ) + service.lifecycle_operation(lifecycle_operation) + mock_send_message_json.assert_called_once() + method, log, url = mock_send_message_json.mock_calls[0].args + data = mock_send_message_json.mock_calls[0].kwargs["data"] + assert method == "POST" + assert log.startswith(f"Request lifecycle operation {lifecycle_operation}") + assert url.endswith(f"sdc2/rest/v1/catalog/services/{service_unique_id}/lifecycleState/{lifecycle_operation}") + assert json.loads(data)["userRemarks"] == str(lifecycle_operation).lower() + mock_update.assert_called_once_with(return_dict) + + +@patch("onapsdk.sdc2.service.Service.send_message_json") +@patch("onapsdk.sdc2.service.Service.get_by_name_and_version_endpoint") +@patch("onapsdk.sdc2.service.ServiceCategory.get_by_uniqe_id") +def test_get_by_name_and_version(mock_get_category_by_unique_id, mock_get_endpoint, mock_send_message_json): + for name, version in [(str(uuid4()), str(uuid4())) for _ in range(10**2)]: + mock_send_message_json.reset_mock() + mock_send_message_json.return_value = { + "uniqueId": str(uuid4()), + "uuid": str(uuid4()), + "invariantUUID": str(uuid4()), + "version": str(uuid4()), + "lastUpdateDate": randint(0, maxsize), + "lifecycleState": choice(list(LifecycleState)), + "lastUpdaterUserId": str(uuid4()), + "allVersions": [str(uuid4()) for _ in range(10**2)], + "archived": choice([True, False]), + "creationDate": randint(0, maxsize), + "componentType": str(uuid4()), + "description": str(uuid4()), + "icon": str(uuid4()), + "name": str(uuid4()), + "systemName": str(uuid4()), + "tags": [str(uuid4()) for _ in range(10**2)], + "actualComponentType": str(uuid4()), + "distributionStatus": str(uuid4()), + "categories": [{"uniqueId": str(uuid4())} for _ in range(10**2)], + "instantiationType": choice(list(ServiceInstantiationType)), + } + mock_get_endpoint.reset_mock() + mock_get_endpoint.return_value = f"{name}/{version}" + Service.get_by_name_and_version(name, version) + mock_send_message_json.assert_called_once() + method, log, _ = mock_send_message_json.mock_calls[0].args + assert method == "GET" + assert log == "Get Service by name and version" + mock_get_endpoint.assert_called_once_with(name, version) + + +@patch("onapsdk.sdc2.service.Service.send_message_json") +def test_add_deployment_artifact(mock_send_message_json): + with NamedTemporaryFile() as temp_file: + temp_file.write(b"Hello world!") + temp_file.seek(0) + + service = Service(name=str(uuid4()), unique_id=str(uuid4())) + service.add_deployment_artifact( + str(uuid4()), + str(uuid4()), + str(uuid4()), + temp_file.name + ) + mock_send_message_json.assert_called_once() + + +@patch("onapsdk.sdc2.service.Service.send_message_json") +@patch("onapsdk.sdc2.sdc_resource.ComponentInstance.create_from_api_response") +def test_add_deployment_artifact(mock_create_component_instance, mock_send_message_json): + service = Service(name=str(uuid4()), unique_id=str(uuid4())) + mock_send_message_json.return_value = [] + assert len(list(service.component_instances)) == 0 + mock_send_message_json.assert_called_once_with( + "GET", + "Get Service component instances", + f"{service.base_back_url}/sdc2/rest/v1/catalog/services/{service.unique_id}/componentInstances" + ) + + mock_send_message_json.return_value = [{}] + assert len(list(service.component_instances)) == 1 + mock_create_component_instance.assert_called_once_with({}, service) + + +@patch("onapsdk.sdc2.service.Service.component_instances", new_callable=PropertyMock) +def test_get_component_by_name(mock_component_instances): + ComponentInstance = namedtuple("ComponentInstance", ["component_name"]) + + service = Service(name="test_component_instances") + mock_component_instances.return_value = [] + assert service.get_component_by_name("test_name") is None + + mock_component_instances.return_value = [ComponentInstance(component_name="test_name")] + assert service.get_component_by_name("test_name") is not None + assert service.get_component_by_name("test_name").component_name == "test_name" + + mock_component_instances.return_value = [ComponentInstance(component_name=f"test_name_{i}") for i in range(10**2)] + assert service.get_component_by_name("test_name") is None + + random_name = f"test_name_{choice(range(10**2))}" + assert service.get_component_by_name(random_name) is not None + assert service.get_component_by_name(random_name).component_name == random_name + + +@patch("onapsdk.sdc2.service.Service.send_message_json") +@patch("onapsdk.sdc2.service.Service.get_create_payload") +@patch("onapsdk.sdc2.service.Service.create_from_api_response") +def test_create(mock_create_from_api_response, mock_get_create_payload, mock_send_message_json): + mock_get_create_payload.return_value = "test_payload" + Service.create(name="test_service") + mock_send_message_json.assert_called_once_with( + "POST", + "Create SERVICE test_service", + Service.CREATE_ENDPOINT, + data="test_payload" + ) + + +@patch("onapsdk.sdc2.service.ServiceCategory.get_by_name") +def test_get_create_payload(mock_resource_category_get_by_name): + resource_category_mock = MagicMock() + resource_category_mock.name = "test_category" + resource_category_mock.unique_id = "category_unique_id" + mock_resource_category_get_by_name.return_value = resource_category_mock + + create_payload = json.loads(Service.get_create_payload(name="test_service")) + assert create_payload["componentType"] == "SERVICE" + assert create_payload["name"] == "test_service" + assert create_payload["contactId"] == "cs0008" + assert create_payload["categories"][0]["name"] == "test_category" + assert create_payload["categories"][0]["uniqueId"] == "category_unique_id" + assert create_payload["description"] == "ONAP SDK Service" + assert create_payload["instantiationType"] == "Macro" + + create_payload = json.loads(Service.get_create_payload(name="test_service", description="test description")) + assert create_payload["componentType"] == "SERVICE" + assert create_payload["name"] == "test_service" + assert create_payload["contactId"] == "cs0008" + assert create_payload["categories"][0]["name"] == "test_category" + assert create_payload["categories"][0]["uniqueId"] == "category_unique_id" + assert create_payload["description"] == "test description" + assert create_payload["instantiationType"] == "Macro" + + +@patch("onapsdk.sdc2.service.Service.send_message") +def test_add_resource(mock_send_message): + s = Service(name="test_service", unique_id=str(uuid4())) + mock_resource = MagicMock() + mock_resource.name = str(uuid4()) + mock_resource.resource_type.return_value = choice(list(ResoureTypeEnum)) + mock_resource.unique_id = str(uuid4()) + mock_resource.version = str(uuid4()) + mock_resource.icon = str(uuid4()) + s.add_resource(mock_resource) + mock_send_message.assert_called_once() + method, log, url = mock_send_message.mock_calls[0].args + data = json.loads(mock_send_message.mock_calls[0].kwargs["data"]) + assert method == "POST" + assert log == f"Add resource {mock_resource.name} into service test_service" + assert url.endswith(f"sdc2/rest/v1/catalog/services/{s.unique_id}/resourceInstance/") + assert data["name"] == mock_resource.name + assert data["originType"] == mock_resource.resource_type().value + assert data["componentUid"] == mock_resource.unique_id + assert data["componentVersion"] == mock_resource.version + assert data["icon"] == mock_resource.icon + + +@patch("onapsdk.sdc2.service.Service.send_message_json") +@patch("onapsdk.sdc2.service.Service.update") +def test_distribute(mock_update, mock_send_message_json): + s = Service(name="test_service", unique_id=str(uuid4())) + s.distribute() + mock_send_message_json.assert_called_once() + method, log, url = mock_send_message_json.mock_calls[0].args + assert method == "POST" + assert log == "Request distribute Service test_service" + assert url.endswith(f"sdc2/rest/v1/catalog/services/{s.unique_id}/distribution/PROD/activate") + + mock_send_message_json.reset_mock() + s.distribute(env="TEST") + mock_send_message_json.assert_called_once() + method, log, url = mock_send_message_json.mock_calls[0].args + assert method == "POST" + assert log == "Request distribute Service test_service" + assert url.endswith(f"sdc2/rest/v1/catalog/services/{s.unique_id}/distribution/TEST/activate") + + +@patch("onapsdk.sdc2.service.Service.send_message_json") +def test_distributions(mock_send_message_json): + s = Service(name="test_service", uuid=str(uuid4())) + + mock_send_message_json.return_value = {} + assert len(list(s.distributions)) == 0 + method, log, url = mock_send_message_json.mock_calls[0].args + assert method == "GET" + assert log == "Request Service test_service distributions" + assert url.endswith(f"sdc2/rest/v1/catalog/services/{s.uuid}/distribution/") + + mock_send_message_json.return_value = { + "distributionStatusOfServiceList": [ + { + "distributionID": str(uuid4()), + "timestamp": randint(0, maxsize), + "userId": str(uuid4()), + "deployementStatus": str(uuid4()) + } + ] + } + assert len(list(s.distributions)) == 1 + + mock_send_message_json.return_value = { + "distributionStatusOfServiceList": [ + { + "distributionID": str(uuid4()), + "timestamp": randint(0, maxsize), + "userId": str(uuid4()), + "deployementStatus": str(uuid4()) + } for _ in range(10**2) + ] + } + assert len(list(s.distributions)) == 10**2 + + +@patch("onapsdk.sdc2.service.Service.distributions", new_callable=PropertyMock) +def test_latest_distribution(mock_distributions): + s = Service(name="test_service") + + mock_distributions.return_value = iter([]) + assert s.latest_distribution is None + + mock_distributions.return_value = iter([1]) # Whatever + assert s.latest_distribution is not None + + +@patch("onapsdk.sdc2.service.Service.latest_distribution", new_callable=PropertyMock) +def test_service_distributed(mock_latest_distribution): + s = Service(name="test_service") + + mock_latest_distribution.return_value = None + assert s.distributed is False + + LatestDistribution = namedtuple("LatestDistribution", "distributed") + mock_latest_distribution.return_value = LatestDistribution(False) + assert s.distributed is False + + mock_latest_distribution.return_value = LatestDistribution(True) + assert s.distributed is True + + +def test_service_distribution_deployment_status_test(): + sd = ServiceDistribution( + distribution_id=str(uuid4()), + timestamp=str(randint(0, maxsize)), + user_id=str(uuid4()), + deployment_status=str(uuid4()) + ) + assert sd._deployment_status_test is False + sd.deployment_status = ServiceDistribution.DISTRIBUTED_DEPLOYMENT_STATUS + assert sd._deployment_status_test is True + + +@patch("onapsdk.sdc2.service.ServiceDistribution.distribution_status_list", new_callable=PropertyMock) +def test_service_distribution_distribution_components_test(mock_distribution_status_list): + mock_distribution_status_list.return_value = [] + sd = ServiceDistribution( + distribution_id=str(uuid4()), + timestamp=str(randint(0, maxsize)), + user_id=str(uuid4()), + deployment_status=str(uuid4()) + ) + assert sd._distribution_components_test is False + + mock_distribution_status_list.return_value = [ + ServiceDistribution.DistributionStatus( + component_id=component_id, + timestamp=str(randint(0, maxsize)), + status=str(uuid4()), + url=str(uuid4()), + error_reason=str(uuid4()) + ) for component_id in settings.SDC_SERVICE_DISTRIBUTION_COMPONENTS + ] + assert sd._distribution_components_test is True + + +@patch("onapsdk.sdc2.service.ServiceDistribution.distribution_status_list", new_callable=PropertyMock) +def test_service_distribution_no_distribution_errors_test(mock_distribution_status_list): + mock_distribution_status_list.return_value = [] + sd = ServiceDistribution( + distribution_id=str(uuid4()), + timestamp=str(randint(0, maxsize)), + user_id=str(uuid4()), + deployment_status=str(uuid4()) + ) + assert sd._no_distribution_errors_test is True + + DistributionStatus = namedtuple("DistributionStatus", ["failed"]) + mock_distribution_status_list.return_value = [ + DistributionStatus(failed=True) + ] + assert sd._no_distribution_errors_test is False + + mock_distribution_status_list.return_value = [ + DistributionStatus(failed=True), + DistributionStatus(failed=False) + ] + assert sd._no_distribution_errors_test is False + + mock_distribution_status_list.return_value = [ + DistributionStatus(failed=False), + DistributionStatus(failed=False) + ] + assert sd._no_distribution_errors_test is True + + +@patch("onapsdk.sdc2.service.ServiceDistribution.send_message_json") +def test_service_distribution_distributon_status_list(mock_send_message_json): + mock_send_message_json.return_value = { + "distributionStatusList": [] + } + sd = ServiceDistribution( + distribution_id=str(uuid4()), + timestamp=str(randint(0, maxsize)), + user_id=str(uuid4()), + deployment_status=str(uuid4()) + ) + assert sd.distribution_status_list == [] + + distribution_status_data = { + "omfComponentID": str(uuid4()), + "timestamp": str(randint(0, maxsize)), + "status": str(uuid4()), + "url": str(uuid4()), + "errorReason": str(uuid4()) + } + mock_send_message_json.return_value = { + "distributionStatusList": [distribution_status_data] + } + assert len(sd.distribution_status_list) == 1 + assert sd.distribution_status_list[0].component_id == distribution_status_data["omfComponentID"] + assert sd.distribution_status_list[0].timestamp == distribution_status_data["timestamp"] + assert sd.distribution_status_list[0].status == distribution_status_data["status"] + assert sd.distribution_status_list[0].url == distribution_status_data["url"] + assert sd.distribution_status_list[0].error_reason == distribution_status_data["errorReason"] diff --git a/tests/test_sdc2_user.py b/tests/test_sdc2_user.py new file mode 100644 index 0000000..8b17fc2 --- /dev/null +++ b/tests/test_sdc2_user.py @@ -0,0 +1,60 @@ +from collections import namedtuple +from random import choice, randint +from sys import maxsize +from unittest.mock import patch +from uuid import uuid4 + +from pytest import raises + +from onapsdk.exceptions import ResourceNotFound +from onapsdk.sdc2.sdc_user import SdcUser + + +@patch("onapsdk.sdc2.sdc_user.SdcUser.send_message_json") +@patch("onapsdk.sdc2.sdc_user.SdcUser.create_from_api_response") +def test_get_all(mock_create_from_api_response, mock_send_message_json): + mock_send_message_json.return_value = [] + + assert len(list(SdcUser.get_all())) == 0 + mock_create_from_api_response.assert_not_called() + + mock_send_message_json.return_value = [{}] + assert len(list(SdcUser.get_all())) == 1 + mock_create_from_api_response.assert_called_once_with({}) + + +@patch("onapsdk.sdc2.sdc_user.SdcUser.get_all") +def test_get_by_user_id(mock_get_all): + mock_get_all.return_value = [] + with raises(ResourceNotFound): + SdcUser.get_by_user_id("test_user") + + TestUser = namedtuple("TestUser", ["user_id"]) + mock_get_all.return_value = [TestUser("not_test_user")] + with raises(ResourceNotFound): + SdcUser.get_by_user_id("test_user") + + mock_get_all.return_value = [TestUser("test_user")] + assert SdcUser.get_by_user_id("test_user") is not None + + +def test_create_from_api_response(): + api_response = { + "userId": str(uuid4()), + "role": str(uuid4()), + "email": str(uuid4()), + "firstName": str(uuid4()), + "fullName": str(uuid4()), + "lastLoginTime": randint(0, maxsize), + "lastName": str(uuid4()), + "status": choice(list(SdcUser.SdcUserStatus)) + } + sdc_user = SdcUser.create_from_api_response(api_response) + assert sdc_user.user_id == api_response["userId"] + assert sdc_user.role == api_response["role"] + assert sdc_user.email == api_response["email"] + assert sdc_user.first_name == api_response["firstName"] + assert sdc_user.last_login_time == api_response["lastLoginTime"] + assert sdc_user.last_name == api_response["lastName"] + assert sdc_user.full_name == api_response["fullName"] + assert sdc_user.status == SdcUser.SdcUserStatus(api_response["status"]) diff --git a/tests/test_sdc2_vf.py b/tests/test_sdc2_vf.py new file mode 100644 index 0000000..a5657d5 --- /dev/null +++ b/tests/test_sdc2_vf.py @@ -0,0 +1,414 @@ + +import json +from collections import namedtuple +from random import choice, randint +from sys import maxsize +from tempfile import NamedTemporaryFile +from unittest.mock import MagicMock, patch, PropertyMock +from uuid import uuid4 + +from pytest import raises + +from onapsdk.exceptions import ResourceNotFound +from onapsdk.sdc2.sdc_resource import LifecycleState, LifecycleOperation +from onapsdk.sdc2.vf import Vf, ResoureTypeEnum + + +def test_pnf_resource_type(): + assert Vf.resource_type() == ResoureTypeEnum.VF + + +def test_pnf_create_payload_template(): + assert Vf.create_payload_template() == "sdc2_create_vf.json.j2" + + +def test_copy_vf(): + vf1 = Vf(name="test_vf1") + vf2 = Vf(name="test_vf2") + assert vf1.name == "test_vf1" + assert vf2.name == "test_vf2" + assert vf1 != vf2 + vf2._copy_object(vf1) + assert vf1.name == "test_vf1" + assert vf2.name == "test_vf1" + assert vf1 == vf2 + + +@patch("onapsdk.sdc2.vf.Vf._get_active_rough") +@patch("onapsdk.sdc2.vf.Vf._get_archived_rough") +def test_get_all_rough(mock_get_archived, mock_get_active): + Vf._get_all_rough() + mock_get_archived.assert_called_once() + mock_get_active.assert_called_once() + + +@patch("onapsdk.sdc2.vf.Vf._get_all_rough") +@patch("onapsdk.sdc2.vf.Vf.get_by_name_and_version") +def test_get_by_name(mock_get_by_name_and_version, mock_get_all_rough): + mock_get_all_rough.return_value = [] + with raises(ResourceNotFound): + Vf.get_by_name("test_1") + + mock_get_all_rough.return_value = [ + { + "name": "not_test_1", + "version": "1.0" + } + ] + with raises(ResourceNotFound): + Vf.get_by_name("test_1") + + mock_get_all_rough.return_value = [ + { + "name": "not_test_1", + "version": "1.0" + }, + { + "name": "test_1", + "version": "1.0" + } + ] + Vf.get_by_name("test_1") + mock_get_by_name_and_version.assert_called_once_with("test_1", "1.0") + + mock_get_by_name_and_version.reset_mock() + mock_get_all_rough.return_value = [ + { + "name": "not_test_1", + "version": "1.0" + }, + { + "name": "test_1", + "version": "1.0" + }, + { + "name": "test_1", + "version": "2.0" + } + ] + Vf.get_by_name("test_1") + mock_get_by_name_and_version.assert_called_once_with("test_1", "2.0") + + +@patch("onapsdk.sdc2.vf.Vf.send_message") +def test_delete(mock_send_message_json): + vf = Vf(name="test_vf", unique_id="test_vf_unique_id") + vf.delete() + mock_send_message_json.assert_called_once() + method, log, url = mock_send_message_json.mock_calls[0].args + assert method == "DELETE" + assert "test_vf" in log + assert url.endswith("test_vf_unique_id") + + +@patch("onapsdk.sdc2.vf.Vf.send_message") +def test_archive(mock_send_message_json): + vf = Vf(name="test_vf", unique_id="test_vf_unique_id") + vf.archive() + mock_send_message_json.assert_called_once() + method, log, url = mock_send_message_json.mock_calls[0].args + assert method == "POST" + assert "test_vf" in log + assert url.endswith("test_vf_unique_id/archive") + + +@patch("onapsdk.sdc2.vf.Vf._get_all_rough") +@patch("onapsdk.sdc2.vf.Vf.get_by_name_and_version") +def test_get_all(mock_get_by_name_and_version, mock_get_all_rough): + mock_get_all_rough.return_value = [] + Vf.get_all() + mock_get_by_name_and_version.assert_not_called() + + mock_get_all_rough.return_value = [ + { + "name": "test_vf_1", + "version": "1.0" + } + ] + list(Vf.get_all()) + mock_get_by_name_and_version.assert_called_once_with("test_vf_1", "1.0") + + mock_get_by_name_and_version.reset_mock() + mock_get_all_rough.return_value = ( + { + "name": f"test_vf_{idx}", + "version": f"{idx}.0" + } for idx in range(100) + ) + list(Vf.get_all()) + assert len(mock_get_by_name_and_version.mock_calls) == 100 + + +@patch("onapsdk.sdc2.vf.Vf.send_message_json") +def test_get_active_rough(mock_send_message_json): + mock_send_message_json.return_value = {"resources": []} + assert(len(list(Vf._get_active_rough()))) == 0 + + mock_send_message_json.return_value = {"resources": [ + { + "resourceType": "VF" + }, + { + "resourceType": "PNF" + }, + { + "resourceType": "VL" + } + ]} + assert(len(list(Vf._get_active_rough()))) == 1 + + +@patch("onapsdk.sdc2.vf.Vf.send_message_json") +def test_get_archive_rough(mock_send_message_json): + mock_send_message_json.return_value = {"resources": []} + assert(len(list(Vf._get_archived_rough()))) == 0 + + mock_send_message_json.return_value = {"resources": [ + { + "resourceType": "VF" + }, + { + "resourceType": "PNF" + }, + { + "resourceType": "VL" + } + ]} + assert(len(list(Vf._get_archived_rough()))) == 1 + + +def test_get_by_name_and_version_endpoint(): + for i in range(1000): + assert Vf.get_by_name_and_version_endpoint( + name=f"vf_{i}", + version=f"v{i}.0").endswith( + f"catalog/resources/resourceName/vf_{i}/resourceVersion/v{i}.0") + + +def test_add_deployment_artifact_endpoint(): + for _ in range(10**3): + object_id: str = str(uuid4()) + assert Vf.add_deployment_artifact_endpoint( + object_id=object_id + ) == f"sdc2/rest/v1/catalog/resources/{object_id}/artifacts" + + +def test_update(): + vf = Vf( + name=str(uuid4()), + unique_id=str(uuid4()), + uuid=str(uuid4()), + invariant_uuid=str(uuid4()), + version=str(uuid4()), + last_update_date=randint(0, maxsize), + lifecycle_state=choice(list(LifecycleState)), + last_updater_user_id=str(uuid4()), + all_versions=[str(uuid4()) for _ in range(10**3)] + ) + update_dict = { + "uniqueId": str(uuid4()), + "uuid": str(uuid4()), + "invariantUUID": str(uuid4()), + "version": str(uuid4()), + "lastUpdateDate": randint(0, maxsize), + "lifecycleState": choice(list(LifecycleState)), + "lastUpdaterUserId": str(uuid4()), + "allVersions": [str(uuid4()) for _ in range(10**3)] + } + vf.update(update_dict) + assert vf.unique_id == update_dict["uniqueId"] + assert vf.uuid == update_dict["uuid"] + assert vf.invariant_uuid == update_dict["invariantUUID"] + assert vf.version == update_dict["version"] + assert vf.last_update_date == update_dict["lastUpdateDate"] + assert vf.lifecycle_state == update_dict["lifecycleState"] + assert vf.last_updater_user_id == update_dict["lastUpdaterUserId"] + assert vf.all_versions == update_dict["allVersions"] + + +@patch("onapsdk.sdc2.vf.Vf.send_message_json") +@patch("onapsdk.sdc2.vf.Vf.update") +def test_lifecycle_operation(mock_update, mock_send_message_json): + for lifecycle_operation in LifecycleOperation: + mock_send_message_json.reset_mock() + mock_update.reset_mock() + return_dict = { + "uniqueId": str(uuid4()), + "uuid": str(uuid4()), + "invariantUUID": str(uuid4()), + "version": str(uuid4()), + "lastUpdateDate": randint(0, maxsize), + "lifecycleState": choice(list(LifecycleState)), + "lastUpdaterUserId": str(uuid4()), + "allVersions": [str(uuid4()) for _ in range(10**3)] + } + mock_send_message_json.return_value = return_dict + vf_unique_id = str(uuid4()) + vf = Vf( + name=str(uuid4()), + unique_id=vf_unique_id + ) + vf.lifecycle_operation(lifecycle_operation) + mock_send_message_json.assert_called_once() + method, log, url = mock_send_message_json.mock_calls[0].args + data = mock_send_message_json.mock_calls[0].kwargs["data"] + assert method == "POST" + assert log.startswith(f"Request lifecycle operation {lifecycle_operation}") + assert url.endswith(f"sdc2/rest/v1/catalog/resources/{vf_unique_id}/lifecycleState/{lifecycle_operation}") + assert json.loads(data)["userRemarks"] == str(lifecycle_operation).lower() + mock_update.assert_called_once_with(return_dict) + + +@patch("onapsdk.sdc2.vf.Vf.send_message_json") +@patch("onapsdk.sdc2.vf.Vf.get_by_name_and_version_endpoint") +def test_get_by_name_and_version(mock_get_endpoint, mock_send_message_json): + for name, version in [(str(uuid4()), str(uuid4())) for _ in range(10**2)]: + mock_send_message_json.reset_mock() + mock_send_message_json.return_value = { + "uniqueId": str(uuid4()), + "uuid": str(uuid4()), + "invariantUUID": str(uuid4()), + "version": str(uuid4()), + "lastUpdateDate": randint(0, maxsize), + "lifecycleState": choice(list(LifecycleState)), + "lastUpdaterUserId": str(uuid4()), + "allVersions": [str(uuid4()) for _ in range(10**2)], + "archived": choice([True, False]), + "creationDate": randint(0, maxsize), + "componentType": str(uuid4()), + "description": str(uuid4()), + "icon": str(uuid4()), + "name": str(uuid4()), + "systemName": str(uuid4()), + "tags": [str(uuid4()) for _ in range(10**2)], + } + mock_get_endpoint.reset_mock() + mock_get_endpoint.return_value = f"{name}/{version}" + Vf.get_by_name_and_version(name, version) + mock_send_message_json.assert_called_once() + method, log, _ = mock_send_message_json.mock_calls[0].args + assert method == "GET" + assert log == "Get Vf by name and version" + mock_get_endpoint.assert_called_once_with(name, version) + + +@patch("onapsdk.sdc2.vf.Vf.send_message_json") +def test_add_deployment_artifact(mock_send_message_json): + with NamedTemporaryFile() as temp_file: + temp_file.write(b"Hello world!") + temp_file.seek(0) + + vf = Vf(name=str(uuid4()), unique_id=str(uuid4())) + vf.add_deployment_artifact( + str(uuid4()), + str(uuid4()), + str(uuid4()), + temp_file.name + ) + mock_send_message_json.assert_called_once() + + +@patch("onapsdk.sdc2.vf.Vf.send_message_json") +@patch("onapsdk.sdc2.sdc_resource.ComponentInstance.create_from_api_response") +def test_add_deployment_artifact(mock_create_component_instance, mock_send_message_json): + vf = Vf(name=str(uuid4()), unique_id=str(uuid4())) + mock_send_message_json.return_value = [] + assert len(list(vf.component_instances)) == 0 + mock_send_message_json.assert_called_once_with( + "GET", + "Get Vf component instances", + f"{vf.base_back_url}/sdc2/rest/v1/catalog/resources/{vf.unique_id}/componentInstances" + ) + + mock_send_message_json.return_value = [{}] + assert len(list(vf.component_instances)) == 1 + mock_create_component_instance.assert_called_once_with({}, vf) + + +@patch("onapsdk.sdc2.vf.Vf.component_instances", new_callable=PropertyMock) +def test_get_component_by_name(mock_component_instances): + ComponentInstance = namedtuple("ComponentInstance", ["component_name"]) + + vf = Vf(name="test_component_instances") + mock_component_instances.return_value = [] + assert vf.get_component_by_name("test_name") is None + + mock_component_instances.return_value = [ComponentInstance(component_name="test_name")] + assert vf.get_component_by_name("test_name") is not None + assert vf.get_component_by_name("test_name").component_name == "test_name" + + mock_component_instances.return_value = [ComponentInstance(component_name=f"test_name_{i}") for i in range(10**2)] + assert vf.get_component_by_name("test_name") is None + + random_name = f"test_name_{choice(range(10**2))}" + assert vf.get_component_by_name(random_name) is not None + assert vf.get_component_by_name(random_name).component_name == random_name + + +@patch("onapsdk.sdc2.vf.Vf.send_message_json") +@patch("onapsdk.sdc2.vf.Vf.get_create_payload") +@patch("onapsdk.sdc2.vf.Vf.create_from_api_response") +def test_create(mock_create_from_api_response, mock_get_create_payload, mock_send_message_json): + mock_get_create_payload.return_value = "test_payload" + Vf.create(name="test_vf") + mock_send_message_json.assert_called_once_with( + "POST", + "Create VF test_vf", + Vf.CREATE_ENDPOINT, + data="test_payload" + ) + + +@patch("onapsdk.sdc2.sdc_resource.ResourceCategory.get_by_name") +def test_get_create_payload(mock_resource_category_get_by_name): + mock_vsp = MagicMock(csar_uuid="test_vsp_csar_uuid") + mock_vendor = MagicMock() + mock_vendor.name = "test_vendor_name" + resource_category_mock = MagicMock() + resource_category_mock.name = "test_category" + resource_category_mock.unique_id = "category_unique_id" + resource_subcategory_mock = MagicMock() + resource_subcategory_mock.name = "test_subcategory" + resource_subcategory_mock.unique_id = "subcategory_unique_id" + resource_category_mock.get_subcategory.return_value = resource_subcategory_mock + mock_resource_category_get_by_name.return_value = resource_category_mock + + create_payload = json.loads(Vf.get_create_payload( + name="test_vf", + vsp=mock_vsp, + vendor=mock_vendor + )) + assert create_payload["name"] == "test_vf" + assert create_payload["contactId"] == "cs0008" + assert create_payload["componentType"] == "RESOURCE" + assert create_payload["csarUUID"] == "test_vsp_csar_uuid" + assert create_payload["csarVersion"] == "1.0" + assert create_payload["categories"][0]["name"] == "test_category" + assert create_payload["categories"][0]["uniqueId"] == "category_unique_id" + assert create_payload["categories"][0]["subcategories"][0]["name"] == "test_subcategory" + assert create_payload["categories"][0]["subcategories"][0]["uniqueId"] == "subcategory_unique_id" + assert create_payload["resourceType"] == "VF" + assert create_payload["description"] == "ONAP SDK Resource" + assert create_payload["vendorName"] == "test_vendor_name" + assert create_payload["vendorRelease"] == "1.0" + + create_payload = json.loads(Vf.get_create_payload( + name="test_vf", + vsp=mock_vsp, + vendor=mock_vendor, + description="test description" + )) + assert create_payload["name"] == "test_vf" + assert create_payload["contactId"] == "cs0008" + assert create_payload["componentType"] == "RESOURCE" + assert create_payload["csarUUID"] == "test_vsp_csar_uuid" + assert create_payload["csarVersion"] == "1.0" + assert create_payload["categories"][0]["name"] == "test_category" + assert create_payload["categories"][0]["uniqueId"] == "category_unique_id" + assert create_payload["categories"][0]["subcategories"][0]["name"] == "test_subcategory" + assert create_payload["categories"][0]["subcategories"][0]["uniqueId"] == "subcategory_unique_id" + assert create_payload["resourceType"] == "VF" + assert create_payload["description"] == "test description" + assert create_payload["vendorName"] == "test_vendor_name" + assert create_payload["vendorRelease"] == "1.0" + diff --git a/tests/test_sdc2_vl.py b/tests/test_sdc2_vl.py new file mode 100644 index 0000000..8c7cd6b --- /dev/null +++ b/tests/test_sdc2_vl.py @@ -0,0 +1,342 @@ + +import json +from collections import namedtuple +from random import choice, randint +from sys import maxsize +from tempfile import NamedTemporaryFile +from unittest.mock import patch, PropertyMock +from uuid import uuid4 + +from pytest import raises + +from onapsdk.exceptions import ResourceNotFound +from onapsdk.sdc2.sdc_resource import LifecycleState, LifecycleOperation +from onapsdk.sdc2.vl import Vl, ResoureTypeEnum + + +def test_pnf_resource_type(): + assert Vl.resource_type() == ResoureTypeEnum.VL + + +def test_copy_vl(): + vl1 = Vl(name="test_vl1") + vl2 = Vl(name="test_vl2") + assert vl1.name == "test_vl1" + assert vl2.name == "test_vl2" + assert vl1 != vl2 + vl2._copy_object(vl1) + assert vl1.name == "test_vl1" + assert vl2.name == "test_vl1" + assert vl1 == vl2 + + +@patch("onapsdk.sdc2.vl.Vl._get_active_rough") +@patch("onapsdk.sdc2.vl.Vl._get_archived_rough") +def test_get_all_rough(mock_get_archived, mock_get_active): + Vl._get_all_rough() + mock_get_archived.assert_called_once() + mock_get_active.assert_called_once() + + +@patch("onapsdk.sdc2.vl.Vl._get_all_rough") +@patch("onapsdk.sdc2.vl.Vl.get_by_name_and_version") +def test_get_by_name(mock_get_by_name_and_version, mock_get_all_rough): + mock_get_all_rough.return_value = [] + with raises(ResourceNotFound): + Vl.get_by_name("test_1") + + mock_get_all_rough.return_value = [ + { + "name": "not_test_1", + "version": "1.0" + } + ] + with raises(ResourceNotFound): + Vl.get_by_name("test_1") + + mock_get_all_rough.return_value = [ + { + "name": "not_test_1", + "version": "1.0" + }, + { + "name": "test_1", + "version": "1.0" + } + ] + Vl.get_by_name("test_1") + mock_get_by_name_and_version.assert_called_once_with("test_1", "1.0") + + mock_get_by_name_and_version.reset_mock() + mock_get_all_rough.return_value = [ + { + "name": "not_test_1", + "version": "1.0" + }, + { + "name": "test_1", + "version": "1.0" + }, + { + "name": "test_1", + "version": "2.0" + } + ] + Vl.get_by_name("test_1") + mock_get_by_name_and_version.assert_called_once_with("test_1", "2.0") + + +@patch("onapsdk.sdc2.vl.Vl.send_message") +def test_delete(mock_send_message_json): + vl = Vl(name="test_vl", unique_id="test_vl_unique_id") + vl.delete() + mock_send_message_json.assert_called_once() + method, log, url = mock_send_message_json.mock_calls[0].args + assert method == "DELETE" + assert "test_vl" in log + assert url.endswith("test_vl_unique_id") + + +@patch("onapsdk.sdc2.vl.Vl.send_message") +def test_archive(mock_send_message_json): + vl = Vl(name="test_vl", unique_id="test_vl_unique_id") + vl.archive() + mock_send_message_json.assert_called_once() + method, log, url = mock_send_message_json.mock_calls[0].args + assert method == "POST" + assert "test_vl" in log + assert url.endswith("test_vl_unique_id/archive") + + +@patch("onapsdk.sdc2.vl.Vl._get_all_rough") +@patch("onapsdk.sdc2.vl.Vl.get_by_name_and_version") +def test_get_all(mock_get_by_name_and_version, mock_get_all_rough): + mock_get_all_rough.return_value = [] + Vl.get_all() + mock_get_by_name_and_version.assert_not_called() + + mock_get_all_rough.return_value = [ + { + "name": "test_vl_1", + "version": "1.0" + } + ] + list(Vl.get_all()) + mock_get_by_name_and_version.assert_called_once_with("test_vl_1", "1.0") + + mock_get_by_name_and_version.reset_mock() + mock_get_all_rough.return_value = ( + { + "name": f"test_vl_{idx}", + "version": f"{idx}.0" + } for idx in range(100) + ) + list(Vl.get_all()) + assert len(mock_get_by_name_and_version.mock_calls) == 100 + + +@patch("onapsdk.sdc2.vl.Vl.send_message_json") +def test_get_active_rough(mock_send_message_json): + mock_send_message_json.return_value = {"resources": []} + assert(len(list(Vl._get_active_rough()))) == 0 + + mock_send_message_json.return_value = {"resources": [ + { + "resourceType": "VF" + }, + { + "resourceType": "PNF" + }, + { + "resourceType": "VL" + } + ]} + assert(len(list(Vl._get_active_rough()))) == 1 + + +@patch("onapsdk.sdc2.vl.Vl.send_message_json") +def test_get_archive_rough(mock_send_message_json): + mock_send_message_json.return_value = {"resources": []} + assert(len(list(Vl._get_archived_rough()))) == 0 + + mock_send_message_json.return_value = {"resources": [ + { + "resourceType": "VF" + }, + { + "resourceType": "PNF" + }, + { + "resourceType": "VL" + } + ]} + assert(len(list(Vl._get_archived_rough()))) == 1 + + +def test_get_by_name_and_version_endpoint(): + for i in range(1000): + assert Vl.get_by_name_and_version_endpoint( + name=f"vl_{i}", + version=f"v{i}.0").endswith( + f"catalog/resources/resourceName/vl_{i}/resourceVersion/v{i}.0") + + +def test_add_deployment_artifact_endpoint(): + for _ in range(10**3): + object_id: str = str(uuid4()) + assert Vl.add_deployment_artifact_endpoint( + object_id=object_id + ) == f"sdc2/rest/v1/catalog/resources/{object_id}/artifacts" + + + +def test_update(): + vl = Vl( + name=str(uuid4()), + unique_id=str(uuid4()), + uuid=str(uuid4()), + invariant_uuid=str(uuid4()), + version=str(uuid4()), + last_update_date=randint(0, maxsize), + lifecycle_state=choice(list(LifecycleState)), + last_updater_user_id=str(uuid4()), + all_versions=[str(uuid4()) for _ in range(10**3)] + ) + update_dict = { + "uniqueId": str(uuid4()), + "uuid": str(uuid4()), + "invariantUUID": str(uuid4()), + "version": str(uuid4()), + "lastUpdateDate": randint(0, maxsize), + "lifecycleState": choice(list(LifecycleState)), + "lastUpdaterUserId": str(uuid4()), + "allVersions": [str(uuid4()) for _ in range(10**3)] + } + vl.update(update_dict) + assert vl.unique_id == update_dict["uniqueId"] + assert vl.uuid == update_dict["uuid"] + assert vl.invariant_uuid == update_dict["invariantUUID"] + assert vl.version == update_dict["version"] + assert vl.last_update_date == update_dict["lastUpdateDate"] + assert vl.lifecycle_state == update_dict["lifecycleState"] + assert vl.last_updater_user_id == update_dict["lastUpdaterUserId"] + assert vl.all_versions == update_dict["allVersions"] + + +@patch("onapsdk.sdc2.vl.Vl.send_message_json") +@patch("onapsdk.sdc2.vl.Vl.update") +def test_lifecycle_operation(mock_update, mock_send_message_json): + for lifecycle_operation in LifecycleOperation: + mock_send_message_json.reset_mock() + mock_update.reset_mock() + return_dict = { + "uniqueId": str(uuid4()), + "uuid": str(uuid4()), + "invariantUUID": str(uuid4()), + "version": str(uuid4()), + "lastUpdateDate": randint(0, maxsize), + "lifecycleState": choice(list(LifecycleState)), + "lastUpdaterUserId": str(uuid4()), + "allVersions": [str(uuid4()) for _ in range(10**3)] + } + mock_send_message_json.return_value = return_dict + vl_unique_id = str(uuid4()) + vl = Vl( + name=str(uuid4()), + unique_id=vl_unique_id + ) + vl.lifecycle_operation(lifecycle_operation) + mock_send_message_json.assert_called_once() + method, log, url = mock_send_message_json.mock_calls[0].args + data = mock_send_message_json.mock_calls[0].kwargs["data"] + assert method == "POST" + assert log.startswith(f"Request lifecycle operation {lifecycle_operation}") + assert url.endswith(f"sdc2/rest/v1/catalog/resources/{vl_unique_id}/lifecycleState/{lifecycle_operation}") + assert json.loads(data)["userRemarks"] == str(lifecycle_operation).lower() + mock_update.assert_called_once_with(return_dict) + + +@patch("onapsdk.sdc2.vl.Vl.send_message_json") +@patch("onapsdk.sdc2.vl.Vl.get_by_name_and_version_endpoint") +def test_get_by_name_and_version(mock_get_endpoint, mock_send_message_json): + for name, version in [(str(uuid4()), str(uuid4())) for _ in range(10**2)]: + mock_send_message_json.reset_mock() + mock_send_message_json.return_value = { + "uniqueId": str(uuid4()), + "uuid": str(uuid4()), + "invariantUUID": str(uuid4()), + "version": str(uuid4()), + "lastUpdateDate": randint(0, maxsize), + "lifecycleState": choice(list(LifecycleState)), + "lastUpdaterUserId": str(uuid4()), + "allVersions": [str(uuid4()) for _ in range(10**2)], + "archived": choice([True, False]), + "creationDate": randint(0, maxsize), + "componentType": str(uuid4()), + "description": str(uuid4()), + "icon": str(uuid4()), + "name": str(uuid4()), + "systemName": str(uuid4()), + "tags": [str(uuid4()) for _ in range(10**2)], + } + mock_get_endpoint.reset_mock() + mock_get_endpoint.return_value = f"{name}/{version}" + Vl.get_by_name_and_version(name, version) + mock_send_message_json.assert_called_once() + method, log, _ = mock_send_message_json.mock_calls[0].args + assert method == "GET" + assert log == "Get Vl by name and version" + mock_get_endpoint.assert_called_once_with(name, version) + + +@patch("onapsdk.sdc2.vl.Vl.send_message_json") +def test_add_deployment_artifact(mock_send_message_json): + with NamedTemporaryFile() as temp_file: + temp_file.write(b"Hello world!") + temp_file.seek(0) + + vl = Vl(name=str(uuid4()), unique_id=str(uuid4())) + vl.add_deployment_artifact( + str(uuid4()), + str(uuid4()), + str(uuid4()), + temp_file.name + ) + mock_send_message_json.assert_called_once() + + +@patch("onapsdk.sdc2.vl.Vl.send_message_json") +@patch("onapsdk.sdc2.sdc_resource.ComponentInstance.create_from_api_response") +def test_get_component_instances(mock_create_component_instance, mock_send_message_json): + vl = Vl(name=str(uuid4()), unique_id=str(uuid4())) + mock_send_message_json.return_value = [] + assert len(list(vl.component_instances)) == 0 + mock_send_message_json.assert_called_once_with( + "GET", + "Get Vl component instances", + f"{vl.base_back_url}/sdc2/rest/v1/catalog/resources/{vl.unique_id}/componentInstances" + ) + + mock_send_message_json.return_value = [{}] + assert len(list(vl.component_instances)) == 1 + mock_create_component_instance.assert_called_once_with({}, vl) + + +@patch("onapsdk.sdc2.vl.Vl.component_instances", new_callable=PropertyMock) +def test_get_component_by_name(mock_component_instances): + ComponentInstance = namedtuple("ComponentInstance", ["component_name"]) + + vl = Vl(name="test_component_instances") + mock_component_instances.return_value = [] + assert vl.get_component_by_name("test_name") is None + + mock_component_instances.return_value = [ComponentInstance(component_name="test_name")] + assert vl.get_component_by_name("test_name") is not None + assert vl.get_component_by_name("test_name").component_name == "test_name" + + mock_component_instances.return_value = [ComponentInstance(component_name=f"test_name_{i}") for i in range(10**2)] + assert vl.get_component_by_name("test_name") is None + + random_name = f"test_name_{choice(range(10**2))}" + assert vl.get_component_by_name(random_name) is not None + assert vl.get_component_by_name(random_name).component_name == random_name diff --git a/tests/test_settings.py b/tests/test_settings.py index 20101f6..4977e66 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -24,7 +24,7 @@ from onapsdk.exceptions import ModuleError def test_global_settings(): """Test global settings.""" - assert len(settings._settings) == 47 + assert len(settings._settings) == 48 assert settings.AAI_URL == "https://aai.api.sparky.simpledemo.onap.org:30233" assert settings.CDS_URL == "http://portal.api.simpledemo.onap.org:30449" assert settings.SDNC_URL == "https://sdnc.api.simpledemo.onap.org:30267" @@ -64,7 +64,7 @@ def test_global_settings(): assert hasattr(settings, "CLAMP_AUTH") assert hasattr(settings, "SO_AUTH") assert hasattr(settings, "SO_CAT_DB_AUTH") - + assert hasattr(settings, "SDC_SERVICE_DISTRIBUTION_COMPONENTS") def test_settings_load_custom(): """Test if custom settings is loaded correctly.""" diff --git a/tests/test_version.py b/tests/test_version.py index cfce4ae..9aef6ce 100644 --- a/tests/test_version.py +++ b/tests/test_version.py @@ -17,4 +17,4 @@ import onapsdk.version as version def test_version(): """Check version is the right one.""" - assert version.__version__ == '12.10.0' + assert version.__version__ == '12.11.0' |