a2ae1f5dfaf5cfdb50245c64b6116966aa90996e
[it/dep.git] / smo-install / test / pythonsdk / src / oransdk / sdc / sdc.py
1 #!/usr/bin/env python3
2 ###
3 # ============LICENSE_START===================================================
4 # ORAN SMO PACKAGE - PYTHONSDK TESTS
5 # ================================================================================
6 #  Copyright (C) 2021 Samsung Electronics
7 #  Copyright (C) 2022 AT&T Intellectual Property. All rights
8 #                             reserved.
9 # ============================================================================
10 # Licensed under the Apache License, Version 2.0 (the "License");
11 # you may not use this file except in compliance with the License.
12 # You may obtain a copy of the License at
13 #
14 #      http://www.apache.org/licenses/LICENSE-2.0
15 #
16 # Unless required by applicable law or agreed to in writing, software
17 # distributed under the License is distributed on an "AS IS" BASIS,
18 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
19 # See the License for the specific language governing permissions and
20 # limitations under the License.
21 #
22 # SPDX-License-Identifier: Apache-2.0
23 # ============LICENSE_END=====================================================
24 #
25 ###
26 """Onap Sdc module."""
27 from time import sleep
28 import onapsdk.constants as const
29 from onapsdk.configuration import settings
30 from onapsdk.exceptions import APIError, ResourceNotFound
31 from onapsdk.onap_service import OnapService as Onap
32 from onapsdk.sdc.properties import Input, NestedInput, ParameterError
33
34 from onapsdk.sdc.vf import Vf
35 from onapsdk.sdc.vsp import Vsp
36 from onapsdk.sdc.vfc import Vfc
37 from onapsdk.sdc.vendor import Vendor
38 from oransdk.sdc.category_management import OranServiceCategory
39 from oransdk.sdc.service import OranService
40
41 class SdcTemplate(Onap):
42     """Onap Sdc Template class."""
43
44     def healthcheck(self) -> dict:
45         """Healchcheck SDC components.
46
47         Returns:
48             status of SDC components
49         """
50         status = self.send_message_json("GET",
51                                         "SDC Healchcheck",
52                                         f"{settings.SDC_FE_URL}/sdc1/feProxy/rest/healthCheck")
53
54         return status
55
56     def create_service_category(self, category_names) -> None:
57         """Create service category by names.
58
59         Args:
60             category_names : The list of category names
61         """
62         for cn in category_names:
63             self._logger.info('creating service category [%s]', cn)
64             OranServiceCategory.create(name=cn)
65
66
67     def create_vendor(self, vendor_name) -> dict:
68         """Create Vendor by names.
69
70         Args:
71             vendor_name : The vendor names
72         Returns:
73             the vendor
74         """
75         vendor = Vendor(vendor_name)
76         vendor.create()
77         try:
78             vendor.onboard()
79         except APIError as e:
80             self._logger.error("Exception during vendor onboarding: %s", e)
81             raise e
82         return vendor
83
84     def create_vsp(self, name, vendor, onboard=False) -> dict:
85         """Create vsp.
86
87         Args:
88             name : The vsp name
89             vendor : The vendor name
90             onboard : The onboard flag
91         Returns:
92             the vsp
93         """
94         self._logger.info("creating vsp: [%s:%s]", name, vendor)
95         retry = 0
96         done = False
97
98         vsp = Vsp(name=name, vendor=vendor)
99         if onboard:
100             while not done:
101                 try:
102                     vsp.create()
103                     vsp.onboard()
104                 except ResourceNotFound as e:
105                     self._logger.error("Failed to onboard %s : %s", name, e)
106                     retry = retry + 1
107                     if retry >= 5:
108                         raise e
109                 except APIError as e:
110                     self._logger.error("Exception during vsp onboarding: %s", e)
111                     raise e
112                 else:
113                     done = True
114         return vsp
115
116
117     def create_vf(self, name, category, subcategory, vendor, onboard=False) -> dict:
118         """Create vf.
119
120         Args:
121             name : The vf name
122             category :  The category name
123             subcategory : The subcategory name
124             vendor : The vendor name
125             onboard : The onboard flag
126         Returns:
127             the vf
128         """
129         self._logger.error("create vf: [%s:%s]", name, category)
130
131         vfc = Vfc('AllottedResource')  # seemd incorrect
132         vf = Vf(name=name, category=category, subcategory=subcategory, vendor=vendor)
133         self._logger.error("create vf 2: ")
134         vf.create()
135         if vf.status == const.DRAFT:
136             vf.add_resource(vfc)
137             self._logger.error("create vf 3:")
138             if onboard:
139                 self.onboard_vf(vf)
140         return vf
141
142
143     def onboard_vf(self, vf) -> None:
144         """Onboard the vf.
145
146         Args:
147             vf : The vf to onboard
148         """
149         retry = 0
150         done = False
151         to = 2
152
153         while not done:
154             try:
155                 vf.onboard()
156             except ResourceNotFound as e:
157                 retry += 1
158                 if retry > 5:
159                     raise e
160                 sleep(to)
161                 to = 2 * to + 1
162             else:
163                 done = True
164         self._logger.info("onboarded vf: [%s]", vf.name)
165
166
167     def create_service(self, name, category, vnfs=None, properties=None, inputs=None, role=None, service_type=None) -> dict:
168         """Create service.
169
170         Args:
171              name : The service name
172              category :  The category name
173              vnfs : The list of vnfs
174              properties : the list of properties
175              role : the role value
176              service_type : the service type
177         Returns:
178              the created service
179         """
180         self._logger.info("create service: [%s:%s]", name, category)
181         retry = 0
182         done = False
183
184         if vnfs is None:
185             vnfs = []
186         if properties is None:
187             properties = []
188
189         srvc = OranService(name=name, category=category, properties=properties, inputs=inputs, role=role, service_type=service_type)
190         srvc.create()
191
192         while not done:
193             try:
194                 if srvc.status == const.DRAFT:
195                     for vnf in vnfs:
196                         srvc.add_resource(vnf)
197
198                 if srvc.status != const.DISTRIBUTED:
199                     srvc.onboard()
200             except ResourceNotFound as e:
201                 retry += 1
202                 if retry > 5:
203                     raise e
204             else:
205                 done = True
206
207         return srvc
208
209     def create_service_1(self, name, category, vnfs=None, properties=None, inputs=None, complex_input=None, role=None, service_type=None) -> dict:
210         """Create slicing profile service.
211
212         Args:
213              name : The service name
214              category :  The category name
215              vnfs : The list of vnfs
216              properties : the list of properties
217              inputs : the list of inputs
218              complex_input : the predefined property type, that should be declared as input
219              role : the role value
220              service_type : the service type
221         Returns:
222              the created service
223         """
224         self._logger.info("create service: [%s:%s]", name, category)
225         retry = 0
226         done = False
227
228         if vnfs is None:
229             vnfs = []
230         if properties is None:
231             properties = []
232
233         srvc = OranService(name=name, category=category, inputs=inputs, complex_input=complex_input, properties=properties, role=role, service_type=service_type)
234         srvc.create()
235
236         while not done:
237             try:
238                 if srvc.status == const.DRAFT:
239                     for vnf in vnfs:
240                         srvc.add_resource(vnf)
241                         for c in srvc.components:
242                             self.set_property_input_slice_ar(vnf, srvc, c)
243
244                 if srvc.status != const.DISTRIBUTED:
245                     srvc.onboard()
246             except ResourceNotFound as e:
247                 retry += 1
248                 if retry > 5:
249                     raise e
250             else:
251                 done = True
252
253         return srvc
254
255     def set_property_input_slice_ar(self, vnf, service, component) -> None:
256         """Get component property.
257
258         Args:
259             vnf: The vnf of the input
260             service : The service
261             component :  The component
262         """
263         self._logger.info("set property input slice ar: %s", component.name)
264         if component.name.startswith("Slice_AR"):
265             self._logger.info("get component Slice_AR 0")
266             cp = self.get_component_property(component, 'allottedresource0_providing_service_invariant_uuid')
267             if cp:
268                 self._logger.info('setting value on property [%s]', cp)
269                 service.declare_input(NestedInput(sdc_resource=vnf, input_obj=Input(unique_id="123",
270                                                                                     input_type=cp.property_type,
271                                                                                     name=cp.name,
272                                                                                     sdc_resource=vnf)))
273             else:
274                 raise ParameterError('no property providing_service_invariant_uuid found')
275
276             cp = self.get_component_property(component, 'allottedresource0_providing_service_uuid')
277             if cp:
278                 service.declare_input(NestedInput(sdc_resource=vnf, input_obj=Input(unique_id="123",
279                                                                                     input_type=cp.property_type,
280                                                                                     name=cp.name,
281                                                                                     sdc_resource=vnf)))
282             else:
283                 raise ParameterError('no property providing_service_uuid found')
284
285     def get_component_property(self, component, name) -> dict:
286         """Get component property.
287
288         Args:
289              component : The component
290              name :  The property name
291         Returns:
292              the property
293         """
294         prop = None
295         try:
296             prop = list(filter(lambda x: x.name == name, component.properties))
297             if prop:
298                 prop = prop[0]
299             else:
300                 raise ParameterError('no property found')
301         except ParameterError as e:
302             self._logger.error("component [%s] has no property [%s]", component.name, name)
303             raise e
304
305         self._logger.error("retrived property [%s] for component [%s]", prop.name if prop else 'null', component.name)
306         return prop