Merge "Fix/add use cases under SMO package"
[it/dep.git] / 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