aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorMichal Jagiello <michal.jagiello@t-mobile.pl>2024-01-16 12:56:15 +0100
committerMichal Jagiello <michal.jagiello@t-mobile.pl>2024-01-31 15:25:22 +0100
commitbfb53be9956b27b20fcefd54504b7d26db052053 (patch)
treec8056a96a8fa54e8f13e19c0accaf796134b6ccb
parent0a5d6dbc73d21d597eba1069d343b6e7684e91f9 (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>
-rw-r--r--src/onapsdk/configuration/global_settings.py10
-rw-r--r--src/onapsdk/sdc2/__init__.py1
-rw-r--r--src/onapsdk/sdc2/component_instance.py282
-rw-r--r--src/onapsdk/sdc2/pnf.py40
-rw-r--r--src/onapsdk/sdc2/sdc.py103
-rw-r--r--src/onapsdk/sdc2/sdc_category.py257
-rw-r--r--src/onapsdk/sdc2/sdc_resource.py709
-rw-r--r--src/onapsdk/sdc2/sdc_user.py119
-rw-r--r--src/onapsdk/sdc2/service.py504
-rw-r--r--src/onapsdk/sdc2/templates/sdc2_add_deployment_artifact.json.j28
-rw-r--r--src/onapsdk/sdc2/templates/sdc2_add_resource.json.j27
-rw-r--r--src/onapsdk/sdc2/templates/sdc2_component_instance_input_set_value.json.j213
-rw-r--r--src/onapsdk/sdc2/templates/sdc2_create_pnf.json.j22
-rw-r--r--src/onapsdk/sdc2/templates/sdc2_create_resource_base.json.j221
-rw-r--r--src/onapsdk/sdc2/templates/sdc2_create_service.json.j213
-rw-r--r--src/onapsdk/sdc2/templates/sdc2_create_vf.json.j22
-rw-r--r--src/onapsdk/sdc2/templates/sdc2_resource_action.json.j23
-rw-r--r--src/onapsdk/sdc2/vf.py40
-rw-r--r--src/onapsdk/sdc2/vl.py30
-rw-r--r--src/onapsdk/utils/jinja.py2
-rw-r--r--src/onapsdk/version.py2
-rw-r--r--tests/test_sdc2.py19
-rw-r--r--tests/test_sdc2_category.py437
-rw-r--r--tests/test_sdc2_component_instance.py203
-rw-r--r--tests/test_sdc2_pnf.py413
-rw-r--r--tests/test_sdc2_resource.py14
-rw-r--r--tests/test_sdc2_service.py604
-rw-r--r--tests/test_sdc2_user.py60
-rw-r--r--tests/test_sdc2_vf.py414
-rw-r--r--tests/test_sdc2_vl.py342
-rw-r--r--tests/test_settings.py4
-rw-r--r--tests/test_version.py2
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'