smo-install/test/pythonsdk/src/oransdk/sdc/service.py

   1 #!/usr/bin/env python3
   2 ###
   3 # ============LICENSE_START=======================================================
   4 # ORAN SMO PACKAGE - PYTHONSDK TESTS
   5 # ================================================================================
   6 # Copyright (C) 2022 AT&T Intellectual Property. All rights
   7 #                             reserved.
   8 # ================================================================================
   9 # Licensed under the Apache License, Version 2.0 (the "License");
  10 # you may not use this file except in compliance with the License.
  11 # You may obtain a copy of the License at
  12 #
  13 # http://www.apache.org/licenses/LICENSE-2.0
  14 #
  15 # Unless required by applicable law or agreed to in writing, software
  16 # distributed under the License is distributed on an "AS IS" BASIS,
  17 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  18 # See the License for the specific language governing permissions and
  19 # limitations under the License.
  20 # ============LICENSE_END============================================
  21 # ===================================================================
  22 #
  23 ###
  24 """SDC Service module."""
  25
  26 from typing import Dict, Iterator, List, Optional, Union
  27 from onapsdk.exceptions import ResourceNotFound
  28 from onapsdk.sdc.component import Component
  29 from onapsdk.sdc.properties import Property, NestedInput
  30 from onapsdk.sdc.service import Service, ServiceInstantiationType
  31 from onapsdk.sdc.sdc_resource import SdcResource
  32 from onapsdk.utils.headers_creator import headers_sdc_creator
  33 from oransdk.utils.jinja import jinja_env
  34
  35
  36 class OranService(Service):  # pylint: disable=too-many-instance-attributes, too-many-public-methods
  37     """ONAP Service Object used for SDC operations."""
  38
  39     def __init__(self, name: str = None, version: str = None, sdc_values: Dict[str, str] = None,  # pylint: disable=too-many-arguments
  40                  resources: List[SdcResource] = None, properties: List[Property] = None, complex_input: Property = None,
  41                  inputs: List[Union[Property, NestedInput]] = None,
  42                  instantiation_type: Optional[ServiceInstantiationType] = \
  43                      None,
  44                  category: str = None, role: str = "", function: str = "", service_type: str = ""):
  45         """
  46         Initialize service object.
  47
  48         Args:
  49             name (str, optional): the name of the service
  50             version (str, optional): the version of the service
  51             sdc_values (Dict[str, str], optional): dictionary of values
  52                 returned by SDC
  53             resources (List[SdcResource], optional): list of SDC resources
  54             properties (List[Property], optional): list of properties to add to service.
  55                 None by default.
  56             inputs (List[Union[Property, NestedInput]], optional): list of inputs
  57                 to declare for service. It can be both Property or NestedInput object.
  58                 None by default.
  59             instantiation_type (ServiceInstantiationType, optional): service instantiation
  60                 type. ServiceInstantiationType.A_LA_CARTE by default
  61             category (str, optional): service category name
  62             role (str, optional): service role
  63             function (str, optional): service function. Empty by default
  64             service_type (str, optional): service type. Empty by default
  65             complex_input (List[Property], optional): internal defined property type, that needs to be declared as input.
  66                 None by default.
  67
  68         """
  69         super().__init__(name=name, sdc_values=sdc_values, version=version, properties=properties,
  70                          inputs=inputs, category=category, resources=resources,
  71                          instantiation_type=instantiation_type, role=role, function=function, service_type=service_type)
  72         self.complex_input = complex_input
  73
  74     @property
  75     def components(self) -> Iterator[Component]:
  76         """Resource components.
  77
  78         Iterate resource components.
  79
  80         Yields:
  81             Component: Resource component object
  82
  83         """
  84         for component_instance in self.send_message_json(\
  85                 "GET",
  86                 f"Get {self.name} resource inputs",
  87                 f"{self.resource_inputs_url}/filteredDataByParams?include=componentInstances"
  88                 ).get("componentInstances", []):
  89             sdc_resource: "SdcResource" = None
  90
  91             if component_instance['originType'] == "ServiceProxy":
  92                 sdc_resource = SdcResource.import_from_sdc(self.send_message_json(\
  93                             "GET",
  94                             f"Get {self.name} component's SDC resource metadata",
  95                             (f"{self.base_front_url}/sdc1/feProxy/rest/v1/catalog/services/"
  96                              f"{component_instance['actualComponentUid']}/"
  97                              "filteredDataByParams?include=metadata"))["metadata"])
  98             else:
  99                 sdc_resource = SdcResource.import_from_sdc(self.send_message_json(\
 100                     "GET",
 101                     f"Get {self.name} component's SDC resource metadata",
 102                     (f"{self.base_front_url}/sdc1/feProxy/rest/v1/catalog/resources/"
 103                      f"{component_instance['actualComponentUid']}/"
 104                      "filteredDataByParams?include=metadata"))["metadata"])
 105
 106             yield Component.create_from_api_response(api_response=component_instance,
 107                                                      sdc_resource=sdc_resource,
 108                                                      parent_sdc_resource=self)
 109
 110     def declare_complex_input(self, property_obj: Property) -> None:
 111         """Declare complex input for resource's property.
 112
 113         For each property input can be declared.
 114
 115         Args:
 116             property_obj (Property): Property to declare input
 117
 118         """
 119         self._logger.debug("Declare input for Complex property")
 120         if property_obj.property_type == "org.openecomp.datatypes.SliceProfile":
 121             self._logger.debug("Declare input for SliceProfile")
 122             self.send_message_json("POST",
 123                                    f"Declare new input for {property_obj.name} property",
 124                                    f"{self.resource_inputs_url}/create/inputs",
 125                                    data=jinja_env().get_template(\
 126                                    "sdc_add_slice_profile_input.json.j2").\
 127                                        render(\
 128                                            sdc_resource=self,
 129                                            property=property_obj))
 130         elif property_obj.property_type == "org.openecomp.datatypes.ServiceProfile":
 131             self._logger.debug("Declare input for ServiceProfile")
 132             self.send_message_json("POST",
 133                                    f"Declare new input for {property_obj.name} property",
 134                                    f"{self.resource_inputs_url}/create/inputs",
 135                                    data=jinja_env().get_template(\
 136                                        "sdc_add_service_profile_input.json.j2").\
 137                                            render(\
 138                                                sdc_resource=self,
 139                                                property=property_obj))
 140         elif property_obj.property_type == "org.openecomp.datatypes.CSProperties":
 141             self._logger.debug("Declare input for CSProperties")
 142             self.send_message_json("POST",
 143                                    f"Declare new input for {property_obj.name} property",
 144                                    f"{self.resource_inputs_url}/create/inputs",
 145                                    data=jinja_env().get_template(\
 146                                        "sdc_add_cs_properties_input.json.j2").\
 147                                             render(\
 148                                                 sdc_resource=self,
 149                                                 property=property_obj))
 150         else:
 151             self._logger.error("Data type %s not supported", property_obj.property_type)
 152
 153     def declare_resource_input(self,
 154                                input_to_declare: Property) -> None:
 155         """Declare input for given property, nested input or component property object.
 156
 157         Call SDC FE API to declare input for given property.
 158
 159         Args:
 160             input_declaration (Union[Property, NestedInput]): Property or ComponentProperty
 161                 to declare input or NestedInput object
 162
 163         Raises:
 164             ParameterError: if the given property is not SDC resource property
 165
 166         """
 167         self.send_message("POST",
 168                           f"Declare new input for {input_to_declare.name} property",
 169                           f"{self.resource_inputs_url}/create/inputs",
 170                           data=jinja_env().get_template(\
 171                               "component_declare_input.json.j2").\
 172                                   render(\
 173                                       component=input_to_declare.component,
 174                                       property=input_to_declare))
 175
 176     def declare_resources_and_properties(self) -> None:
 177         """Delcare resources and properties.
 178
 179         It declares also inputs.
 180
 181         """
 182         for resource in self.resources:
 183             self.add_resource(resource)
 184         for property_to_add in self._properties_to_add:
 185             self.add_property(property_to_add)
 186         for input_to_add in self._inputs_to_add:
 187             self.declare_input(input_to_add)
 188         if self.complex_input is not None:
 189             self.declare_complex_input(self.complex_input)
 190
 191
 192     def get_distribution_status(self) -> dict:
 193         """Get service distribution status."""
 194         url = "{}/services/distribution/{}".format(self._base_create_url(),
 195                                                    self.distribution_id)
 196         headers = headers_sdc_creator(SdcResource.headers)
 197         try:
 198             result = self.send_message_json("GET",
 199                                             "Check distribution for {}".format(
 200                                                 self.name),
 201                                             url,
 202                                             headers=headers)
 203         except ResourceNotFound:
 204             msg = f"No distributions found for {self.name} of {self.__class__.__name__}."
 205             self._logger.debug(msg)
 206         else:
 207             return result