path: root/src/onapsdk/sdc2/service.py

"""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


ARTIFACT_NOT_USED_ERROR_REASON = \
    "The artifact has not been used by the modules defined in the resource"


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 distributed(self) -> bool:
            """Flag to detrmine if status is distributed.

            If status is not DOWNLOAD_OK it means that component is not yet
                distributed and we need to wait for status update.

            Returns:
                bool: True if distribution of component was completed.

            """
            return not self.failed and (
                self.component_id not in settings.SDC_SERVICE_DISTRIBUTION_DESIRED_STATE
                or self.status == settings.SDC_SERVICE_DISTRIBUTION_DESIRED_STATE[
                    self.component_id])

        @property
        def failed(self) -> bool:
            """Flag 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". On SO there is also
                an error "The artifact has not been used..." which also shouldn't
                be treated as an error.

            Returns:
                bool: True if distribution on component failed or not.

            """
            return self.error_reason not in ["null", ARTIFACT_NOT_USED_ERROR_REASON] 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 and downloaded artifacts.

        List of required components can be configured via SDC_SERVICE_DISTRIBUTION_COMPONENTS
            setting value.

        Returns:
            bool: True if all required components were notified and artifacts were downloaded
                False otherwise

        """
        notified_components_set: Set[str] = {
            distribution.component_id for distribution in filter(
                lambda item: item.distributed, self.distribution_status_list)
        }
        return set(settings.SDC_SERVICE_DISTRIBUTION_COMPONENTS).issubset(notified_components_set)

    @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 sorted(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", []),
                                               key=lambda dist: dist["timestamp"],
                                               reverse=True):
            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