3 # ============LICENSE_START=======================================================
4 # ORAN SMO PACKAGE - PYTHONSDK TESTS
5 # ================================================================================
6 # Copyright (C) 2022 AT&T Intellectual Property. All rights
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
13 # http://www.apache.org/licenses/LICENSE-2.0
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 # ===================================================================
24 """SDC Service module."""
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
36 class OranService(Service): # pylint: disable=too-many-instance-attributes, too-many-public-methods
37 """ONAP Service Object used for SDC operations."""
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] = \
44 category: str = None, role: str = "", function: str = "", service_type: str = ""):
46 Initialize service object.
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
53 resources (List[SdcResource], optional): list of SDC resources
54 properties (List[Property], optional): list of properties to add to service.
56 inputs (List[Union[Property, NestedInput]], optional): list of inputs
57 to declare for service. It can be both Property or NestedInput object.
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.
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
75 def components(self) -> Iterator[Component]:
76 """Resource components.
78 Iterate resource components.
81 Component: Resource component object
84 for component_instance in self.send_message_json(\
86 f"Get {self.name} resource inputs",
87 f"{self.resource_inputs_url}/filteredDataByParams?include=componentInstances"
88 ).get("componentInstances", []):
89 sdc_resource: "SdcResource" = None
91 if component_instance['originType'] == "ServiceProxy":
92 sdc_resource = SdcResource.import_from_sdc(self.send_message_json(\
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"])
99 sdc_resource = SdcResource.import_from_sdc(self.send_message_json(\
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"])
106 yield Component.create_from_api_response(api_response=component_instance,
107 sdc_resource=sdc_resource,
108 parent_sdc_resource=self)
110 def declare_complex_input(self, property_obj: Property) -> None:
111 """Declare complex input for resource's property.
113 For each property input can be declared.
116 property_obj (Property): Property to declare input
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").\
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").\
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").\
149 property=property_obj))
151 self._logger.error("Data type %s not supported", property_obj.property_type)
153 def declare_resource_input(self,
154 input_to_declare: Property) -> None:
155 """Declare input for given property, nested input or component property object.
157 Call SDC FE API to declare input for given property.
160 input_declaration (Union[Property, NestedInput]): Property or ComponentProperty
161 to declare input or NestedInput object
164 ParameterError: if the given property is not SDC resource property
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").\
173 component=input_to_declare.component,
174 property=input_to_declare))
176 def declare_resources_and_properties(self) -> None:
177 """Delcare resources and properties.
179 It declares also inputs.
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)
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)
198 result = self.send_message_json("GET",
199 "Check distribution for {}".format(
203 except ResourceNotFound:
204 msg = f"No distributions found for {self.name} of {self.__class__.__name__}."
205 self._logger.debug(msg)