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
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
14 # http://www.apache.org/licenses/LICENSE-2.0
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.
22 # SPDX-License-Identifier: Apache-2.0
23 # ============LICENSE_END=====================================================
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
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
41 class SdcTemplate(Onap):
42 """Onap Sdc Template class."""
44 def healthcheck(self) -> dict:
45 """Healchcheck SDC components.
48 status of SDC components
50 status = self.send_message_json("GET",
52 f"{settings.SDC_FE_URL}/sdc1/feProxy/rest/healthCheck")
56 def create_service_category(self, category_names) -> None:
57 """Create service category by names.
60 category_names : The list of category names
62 for cn in category_names:
63 self._logger.info('creating service category [%s]', cn)
64 OranServiceCategory.create(name=cn)
67 def create_vendor(self, vendor_name) -> dict:
68 """Create Vendor by names.
71 vendor_name : The vendor names
75 vendor = Vendor(vendor_name)
80 self._logger.error("Exception during vendor onboarding: %s", e)
84 def create_vsp(self, name, vendor, onboard=False) -> dict:
89 vendor : The vendor name
90 onboard : The onboard flag
94 self._logger.info("creating vsp: [%s:%s]", name, vendor)
98 vsp = Vsp(name=name, vendor=vendor)
104 except ResourceNotFound as e:
105 self._logger.error("Failed to onboard %s : %s", name, e)
109 except APIError as e:
110 self._logger.error("Exception during vsp onboarding: %s", e)
117 def create_vf(self, name, category, subcategory, vendor, onboard=False) -> dict:
122 category : The category name
123 subcategory : The subcategory name
124 vendor : The vendor name
125 onboard : The onboard flag
129 self._logger.error("create vf: [%s:%s]", name, category)
131 vfc = Vfc('AllottedResource') # seemd incorrect
132 vf = Vf(name=name, category=category, subcategory=subcategory, vendor=vendor)
133 self._logger.error("create vf 2: ")
135 if vf.status == const.DRAFT:
137 self._logger.error("create vf 3:")
143 def onboard_vf(self, vf) -> None:
147 vf : The vf to onboard
156 except ResourceNotFound as e:
164 self._logger.info("onboarded vf: [%s]", vf.name)
167 def create_service(self, name, category, vnfs=None, properties=None, inputs=None, role=None, service_type=None) -> dict:
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
180 self._logger.info("create service: [%s:%s]", name, category)
186 if properties is None:
189 srvc = OranService(name=name, category=category, properties=properties, inputs=inputs, role=role, service_type=service_type)
194 if srvc.status == const.DRAFT:
196 srvc.add_resource(vnf)
198 if srvc.status != const.DISTRIBUTED:
200 except ResourceNotFound as e:
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.
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
224 self._logger.info("create service: [%s:%s]", name, category)
230 if properties is None:
233 srvc = OranService(name=name, category=category, inputs=inputs, complex_input=complex_input, properties=properties, role=role, service_type=service_type)
238 if srvc.status == const.DRAFT:
240 srvc.add_resource(vnf)
241 for c in srvc.components:
242 self.set_property_input_slice_ar(vnf, srvc, c)
244 if srvc.status != const.DISTRIBUTED:
246 except ResourceNotFound as e:
255 def set_property_input_slice_ar(self, vnf, service, component) -> None:
256 """Get component property.
259 vnf: The vnf of the input
260 service : The service
261 component : The component
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')
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,
274 raise ParameterError('no property providing_service_invariant_uuid found')
276 cp = self.get_component_property(component, 'allottedresource0_providing_service_uuid')
278 service.declare_input(NestedInput(sdc_resource=vnf, input_obj=Input(unique_id="123",
279 input_type=cp.property_type,
283 raise ParameterError('no property providing_service_uuid found')
285 def get_component_property(self, component, name) -> dict:
286 """Get component property.
289 component : The component
290 name : The property name
296 prop = list(filter(lambda x: x.name == name, component.properties))
300 raise ParameterError('no property found')
301 except ParameterError as e:
302 self._logger.error("component [%s] has no property [%s]", component.name, name)
305 self._logger.error("retrived property [%s] for component [%s]", prop.name if prop else 'null', component.name)