path: root/src/onapsdk/sdc/__init__.py

"""SDC Element module."""
#   Copyright 2022 Orange, Deutsche Telekom AG
#
#   Licensed under the Apache License, Version 2.0 (the "License");
#   you may not use this file except in compliance with the License.
#   You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an "AS IS" BASIS,
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#   See the License for the specific language governing permissions and
#   limitations under the License.
from typing import Any, Dict, List, Optional, Union
from operator import attrgetter
from abc import ABC, abstractmethod

from requests import Response

from onapsdk.configuration import settings
from onapsdk.exceptions import APIError, RequestError
from onapsdk.onap_service import OnapService
import onapsdk.constants as const
from onapsdk.utils.jinja import jinja_env
from onapsdk.utils.gui import GuiItem, GuiList

class SDC(OnapService, ABC):
    """Mother Class of all SDC elements."""

    server: str = "SDC"
    base_front_url = settings.SDC_FE_URL
    base_back_url = settings.SDC_BE_URL

    def __init__(self, name: str = None) -> None:
        """Initialize SDC."""
        super().__init__()
        self.name: str = name

    def __eq__(self, other: Any) -> bool:
        """
        Check equality for SDC and children.

        Args:
            other: another object

        Returns:
            bool: True if same object, False if not

        """
        if isinstance(other, type(self)):
            return self.name == other.name
        return False

    @classmethod
    @abstractmethod
    def _get_all_url(cls) -> str:
        """
        Get URL for all elements in SDC.

        Raises:
            NotImplementedError: this is an abstract method.

        """

    @classmethod
    @abstractmethod
    def _get_objects_list(cls,
                          result: List[Dict[str, Any]]) -> List['SdcResource']:
        """
        Import objects created in SDC.

        Args:
            result (Dict[str, Any]): the result returned by SDC in a Dict

        Raises:
            NotImplementedError: this is an abstract method.

        """

    @classmethod
    @abstractmethod
    def _base_url(cls) -> str:
        """
        Give back the base url of Sdc.

        Raises:
            NotImplementedError: this is an abstract method.

        """

    @classmethod
    @abstractmethod
    def _base_create_url(cls) -> str:
        """
        Give back the base url of Sdc.

        Raises:
            NotImplementedError: this is an abstract method.

        """

    @abstractmethod
    def _copy_object(self, obj: 'SDC') -> None:
        """
        Copy relevant properties from object.

        Args:
            obj (Sdc): the object to "copy"

        Raises:
            NotImplementedError: this is an abstract method.

        """

    @classmethod
    @abstractmethod
    def import_from_sdc(cls, values: Dict[str, Any]) -> 'SDC':
        """
        Import Sdc object from SDC.

        Args:
            values (Dict[str, Any]): dict to parse returned from SDC.

        Raises:
            NotImplementedError: this is an abstract method.

        """

    @staticmethod
    def _get_mapped_version(item: "SDC") -> Optional[Union[float, str]]:
        """Map Sdc objects version to float.

        Mostly we need to get the newest version of the requested objects. To do
            so we use the version property of them. In most cases it's string
            formatted float value, but in some cases (like VSP objects) it isn't.
        That method checks if given object has "version" attribute and if it's not
            a None it tries to map it's value to float. If it's not possible it
            returns the alrady existing value.

        Args:
            item (SDC): SDC item to map version to float

        Returns:
            Optional[Union[float, str]]: Float format version if possible,
                string otherwise. If object doesn't have "version"
                attribut returns None.

        """
        if hasattr(item, "version") and item.version is not None:
            try:
                return float(item.version)
            except ValueError:
                return item.version
        else:
            return None

    @classmethod
    def get_all(cls, **kwargs) -> List['SDC']:
        """
        Get the objects list created in SDC.

        Returns:
            the list of the objects

        """
        cls._logger.info("retrieving all objects of type %s from SDC",
                         cls.__name__)
        url = cls._get_all_url()
        objects = []

        try:
            result = \
                cls.send_message_json('GET', "get {}s".format(cls.__name__),
                                      url, **kwargs)

            for obj_info in cls._get_objects_list(result):
                objects.append(cls.import_from_sdc(obj_info))

        except APIError as exc:
            cls._logger.debug("Couldn't get %s: %s", cls.__name__, exc)
        except KeyError as exc:
            cls._logger.debug("Invalid result dictionary: %s", exc)

        cls._logger.debug("number of %s returned: %s", cls.__name__,
                          len(objects))
        return objects

    def exists(self) -> bool:
        """
        Check if object already exists in SDC and update infos.

        Returns:
            True if exists, False either

        """
        self._logger.debug("check if %s %s exists in SDC",
                           type(self).__name__, self.name)
        objects = self.get_all()

        self._logger.debug("filtering objects of all versions to be %s",
                           self.name)
        relevant_objects = list(filter(lambda obj: obj == self, objects))

        if not relevant_objects:

            self._logger.info("%s %s doesn't exist in SDC",
                              type(self).__name__, self.name)
            return False

        if hasattr(self, 'version_filter') and self.version_filter is not None: # pylint: disable=no-member

            self._logger.debug("filtering %s objects by version %s",
                               self.name, self.version_filter) # pylint: disable=no-member

            all_versioned = filter(
                lambda obj: obj.version == self.version_filter, relevant_objects) # pylint: disable=no-member

            try:
                versioned_object = next(all_versioned)
            except StopIteration:
                self._logger.info("Version %s of %s %s, doesn't exist in SDC",
                                  self.version_filter, type(self).__name__,  # pylint: disable=no-member
                                  self.name)
                return False

        else:
            versioned_object = max(relevant_objects, key=self._get_mapped_version)

        self._logger.info("%s found, updating information", type(self).__name__)
        self._copy_object(versioned_object)
        return True

    @classmethod
    def get_guis(cls) -> GuiItem:
        """Retrieve the status of the SDC GUIs.

        Only one GUI is referenced for SDC
        the SDC Front End

        Return the list of GUIs
        """
        gui_url = settings.SDC_GUI_SERVICE
        sdc_gui_response = cls.send_message(
            "GET", "Get SDC GUI Status", gui_url)
        guilist = GuiList([])
        guilist.add(GuiItem(
            gui_url,
            sdc_gui_response.status_code))
        return guilist

class SdcOnboardable(SDC, ABC):
    """Base class for onboardable SDC resources (Vendors, Services, ...)."""

    ACTION_TEMPLATE: str
    ACTION_METHOD: str

    def __init__(self, name: str = None) -> None:
        """Initialize the object."""
        super().__init__(name)
        self._identifier: str = None
        self._status: str = None
        self._version: str = None

    @property
    def identifier(self) -> str:
        """Return and lazy load the identifier."""
        if not self._identifier:
            self.load()
        return self._identifier

    @property
    def status(self) -> str:
        """Return and lazy load the status."""
        if self.created() and not self._status:
            self.load()
        return self._status

    @property
    def version(self) -> str:
        """Return and lazy load the version."""
        if self.created() and not self._version:
            self.load()
        return self._version

    @identifier.setter
    def identifier(self, value: str) -> None:
        """Set value for identifier."""
        self._identifier = value

    @status.setter
    def status(self, status: str) -> None:
        """Return and lazy load the status."""
        self._status = status

    @version.setter
    def version(self, version: str) -> None:
        """Return and lazy load the status."""
        self._version = version

    def created(self) -> bool:
        """Determine if SDC is created."""
        if self.name and not self._identifier:
            return self.exists()
        return bool(self._identifier)

    def submit(self) -> None:
        """Submit the SDC object in order to enable it."""
        self._logger.info("attempting to certify/sumbit %s %s in SDC",
                          type(self).__name__, self.name)
        if self.status != const.CERTIFIED and self.created():
            self._really_submit()
        elif self.status == const.CERTIFIED:
            self._logger.warning("%s %s in SDC is already submitted/certified",
                                 type(self).__name__, self.name)
        elif not self.created():
            self._logger.warning("%s %s in SDC is not created",
                                 type(self).__name__, self.name)

    def _create(self, template_name: str, **kwargs) -> None:
        """Create the object in SDC if not already existing."""
        self._logger.info("attempting to create %s %s in SDC",
                          type(self).__name__, self.name)
        if not self.exists():
            url = "{}/{}".format(self._base_create_url(), self._sdc_path())
            template = jinja_env().get_template(template_name)
            data = template.render(**kwargs)
            try:
                create_result = self.send_message_json('POST',
                                                       "create {}".format(
                                                           type(self).__name__),
                                                       url,
                                                       data=data)
            except RequestError as exc:
                self._logger.error(
                    "an error occured during creation of %s %s in SDC",
                    type(self).__name__, self.name)
                raise exc
            else:
                self._logger.info("%s %s is created in SDC",
                                  type(self).__name__, self.name)
                self._status = const.DRAFT
                self.identifier = self._get_identifier_from_sdc(create_result)
                self._version = self._get_version_from_sdc(create_result)
                self.update_informations_from_sdc_creation(create_result)

        else:
            self._logger.warning("%s %s is already created in SDC",
                                 type(self).__name__, self.name)

    def _action_to_sdc(self, action: str, action_type: str = None,
                       **kwargs) -> Response:
        """
        Really do an action in the SDC.

        Args:
            action (str): the action to perform
            action_type (str, optional): the type of action
            headers (Dict[str, str], optional): headers to use if any

        Returns:
            Response: the response

        """
        subpath = self._generate_action_subpath(action)
        url = self._action_url(self._base_create_url(),
                               subpath,
                               self._version_path(),
                               action_type=action_type)
        template = jinja_env().get_template(self.ACTION_TEMPLATE)
        data = template.render(action=action, const=const)

        return self.send_message(self.ACTION_METHOD,
                                 "{} {}".format(action,
                                                type(self).__name__),
                                 url,
                                 data=data,
                                 **kwargs)

    @abstractmethod
    def update_informations_from_sdc(self, details: Dict[str, Any]) -> None:
        """

        Update instance with details from SDC.

        Args:
            details ([type]): [description]

        """
    @abstractmethod
    def update_informations_from_sdc_creation(self,
                                              details: Dict[str, Any]) -> None:
        """

        Update instance with details from SDC after creation.

        Args:
            details ([type]): the details from SDC

        """

    @abstractmethod
    def load(self) -> None:
        """
        Load Object information from SDC.

        Raises:
            NotImplementedError: this is an abstract method.

        """

    @abstractmethod
    def _get_version_from_sdc(self, sdc_infos: Dict[str, Any]) -> str:
        """
        Get version from SDC results.

        Args:
            sdc_infos (Dict[str, Any]): the result dict from SDC

        Raises:
            NotImplementedError: this is an abstract method.

        """
    @abstractmethod
    def _get_identifier_from_sdc(self, sdc_infos: Dict[str, Any]) -> str:
        """
        Get identifier from SDC results.

        Args:
            sdc_infos (Dict[str, Any]): the result dict from SDC

        Raises:
            NotImplementedError: this is an abstract method.

        """
    @abstractmethod
    def _generate_action_subpath(self, action: str) -> str:
        """

        Generate subpath part of SDC action url.

        Args:
            action (str): the action that will be done

        Raises:
            NotImplementedError: this is an abstract method.

        """
    @abstractmethod
    def _version_path(self) -> str:
        """
        Give the end of the path for a version.

        Raises:
            NotImplementedError: this is an abstract method.

        """
    @abstractmethod
    def _really_submit(self) -> None:
        """Really submit the SDC Vf in order to enable it."""
    @staticmethod
    @abstractmethod
    def _action_url(base: str,
                    subpath: str,
                    version_path: str,
                    action_type: str = None) -> str:
        """
        Generate action URL for SDC.

        Raises:
            NotImplementedError: this is an abstract method.

        """
    @classmethod
    @abstractmethod
    def _sdc_path(cls) -> None:
        """Give back the end of SDC path."""

    @abstractmethod
    def onboard(self) -> None:
        """Onboard resource.

        Onboarding is a full stack of actions which needs to be done to
            make SDC resource ready to use. It depends on the type of object
            but most of them needs to be created and submitted.
        """