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