src/onaptests/steps/onboard/service.py

   1 import time
   2 from typing import Any, Dict
   3 from yaml import load, SafeLoader
   4
   5 import onapsdk.constants as onapsdk_const
   6 from onapsdk.configuration import settings
   7 from onapsdk.exceptions import APIError, ResourceNotFound
   8 from onapsdk.sdc.component import Component
   9 from onapsdk.sdc.pnf import Pnf
  10 from onapsdk.sdc.properties import ComponentProperty
  11 from onapsdk.sdc.service import Service, ServiceInstantiationType
  12 from onapsdk.sdc.vf import Vf
  13 from onapsdk.sdc.vl import Vl
  14
  15 from ..base import BaseStep, YamlTemplateBaseStep
  16 from .pnf import PnfOnboardStep, YamlTemplatePnfOnboardStep
  17 from .vf import VfOnboardStep, YamlTemplateVfOnboardStep
  18
  19
  20 class ServiceOnboardStep(BaseStep):
  21     """Service onboard step."""
  22
  23     def __init__(self, cleanup=False):
  24         """Initialize step.
  25
  26         Substeps:
  27             - VfOnboardStep.
  28         """
  29         super().__init__(cleanup=cleanup)
  30         if settings.VF_NAME != "":
  31             self.add_step(VfOnboardStep(cleanup=cleanup))
  32         if settings.PNF_NAME != "":
  33             self.add_step(PnfOnboardStep(cleanup=cleanup))
  34
  35     @property
  36     def description(self) -> str:
  37         """Step description."""
  38         return "Onboard service in SDC."
  39
  40     @property
  41     def component(self) -> str:
  42         """Component name."""
  43         return "SDC"
  44
  45     @BaseStep.store_state
  46     def execute(self):
  47         """Onboard service.
  48
  49         Use settings values:
  50          - VL_NAME,
  51          - VF_NAME,
  52          - PNF_NAME,
  53          - SERVICE_NAME,
  54          - SERVICE_INSTANTIATION_TYPE.
  55
  56         """
  57         super().execute()
  58         service: Service = Service(name=settings.SERVICE_NAME, instantiation_type=settings.SERVICE_INSTANTIATION_TYPE)
  59         if not service.created():
  60             service.create()
  61             if settings.VL_NAME != "":
  62                 vl: Vl = Vl(name=settings.VL_NAME)
  63                 service.add_resource(vl)
  64             if settings.VF_NAME != "":
  65                 vf: Vf = Vf(name=settings.VF_NAME)
  66                 service.add_resource(vf)
  67             if settings.PNF_NAME != "":
  68                 pnf: Pnf = Pnf(name=settings.PNF_NAME)
  69                 service.add_resource(pnf)
  70         # If the service is already distributed, do not try to checkin/onboard (replay of tests)
  71         # checkin is done if needed
  72         # If service is replayed, no need to try to re-onboard the model
  73         # Double check because of: https://gitlab.com/Orange-OpenSource/lfn/onap/python-onapsdk/-/issues/176
  74         if not service.distributed and service.status != onapsdk_const.DISTRIBUTED:
  75             time.sleep(10)
  76             service.checkin()
  77             service.onboard()
  78
  79
  80 class YamlTemplateServiceOnboardStep(YamlTemplateBaseStep):
  81     """Service onboard using YAML template step."""
  82
  83     def __init__(self, cleanup=False):
  84         """Initialize step.
  85
  86         Substeps:
  87             - YamlTemplateVfOnboardStep.
  88         """
  89         super().__init__(cleanup=cleanup)
  90         self._yaml_template: dict = None
  91         self._model_yaml_template: dict = None
  92         if "vnfs" in self.yaml_template[self.service_name]:
  93             self.add_step(YamlTemplateVfOnboardStep(cleanup=cleanup))
  94         if "pnfs" in self.yaml_template[self.service_name]:
  95             self.add_step(YamlTemplatePnfOnboardStep(cleanup=cleanup))
  96
  97     @property
  98     def description(self) -> str:
  99         """Step description."""
 100         return "Onboard service described in YAML file in SDC."
 101
 102     @property
 103     def component(self) -> str:
 104         """Component name."""
 105         return "SDC"
 106
 107     @property
 108     def yaml_template(self) -> dict:
 109         """Step YAML template.
 110
 111         Load from file if it's a root step, get from parent otherwise.
 112
 113         Returns:
 114             dict: Step YAML template
 115
 116         """
 117         if settings.MODEL_YAML_TEMPLATE:
 118             return self.model_yaml_template
 119         elif self.is_root:
 120             if not self._yaml_template:
 121                 with open(settings.SERVICE_YAML_TEMPLATE, "r") as yaml_template:
 122                     self._yaml_template: dict = load(yaml_template, SafeLoader)
 123             return self._yaml_template
 124         return self.parent.yaml_template
 125
 126     @property
 127     def model_yaml_template(self) -> dict:
 128         """Step Model YAML template.
 129
 130         Load from file if it's a root step, get from parent otherwise.
 131
 132         Returns:
 133             dict: Step YAML template
 134
 135         """
 136         if self.is_root:
 137             if not self._model_yaml_template:
 138                 with open(settings.MODEL_YAML_TEMPLATE, "r") as model_yaml_template:
 139                     self._model_yaml_template: dict = load(model_yaml_template)
 140             return self._model_yaml_template
 141         return self.parent.model_yaml_template
 142
 143     @property
 144     def service_name(self) -> str:
 145         """Service name.
 146
 147         Get from YAML template if it's a root step, get from parent otherwise.
 148
 149         Returns:
 150             str: Service name
 151
 152         """
 153         if self.is_root:
 154             return next(iter(self.yaml_template.keys()))
 155         return self.parent.service_name
 156
 157     @YamlTemplateBaseStep.store_state
 158     def execute(self):
 159         """Onboard service."""
 160         super().execute()
 161         if "instantiation_type" in self.yaml_template[self.service_name]:
 162             instantiation_type: ServiceInstantiationType = ServiceInstantiationType(
 163                 self.yaml_template[self.service_name]["instantiation_type"])
 164         else:
 165             instantiation_type: ServiceInstantiationType = ServiceInstantiationType.A_LA_CARTE
 166         service: Service = Service(name=self.service_name, instantiation_type=instantiation_type)
 167         if not service.created():
 168             service.create()
 169             self.declare_resources(service)
 170             self.assign_properties(service)
 171             # If the service is already distributed, do not try to checkin/onboard (replay of tests)
 172             # checkin is done if needed
 173             # If service is replayed, no need to try to re-onboard the model
 174         # Double check because of: https://gitlab.com/Orange-OpenSource/lfn/onap/python-onapsdk/-/issues/176
 175         if not service.distributed and service.status != onapsdk_const.DISTRIBUTED:
 176             try:
 177                 service.checkin()
 178             except (APIError, ResourceNotFound):
 179                 # Retry as checkin may be a bit long
 180                 # Temp workaround to avoid internal race in SDC
 181                 time.sleep(10)
 182                 service.checkin()
 183             service.onboard()
 184
 185     def declare_resources(self, service: Service) -> None:
 186         """Declare resources.
 187
 188         Resources defined in YAML template are declared.
 189
 190         Args:
 191             service (Service): Service object
 192
 193         """
 194         if "networks" in self.yaml_template[self.service_name]:
 195             for net in self.yaml_template[self.service_name]["networks"]:
 196                 vl: Vl = Vl(name=net['vl_name'])
 197                 service.add_resource(vl)
 198         if "vnfs" in self.yaml_template[self.service_name]:
 199             for vnf in self.yaml_template[self.service_name]["vnfs"]:
 200                 vf: Vf = Vf(name=vnf["vnf_name"])
 201                 service.add_resource(vf)
 202         if "pnfs" in self.yaml_template[self.service_name]:
 203             for pnf in self.yaml_template[self.service_name]["pnfs"]:
 204                 pnf_obj: Pnf = Pnf(name=pnf["pnf_name"])
 205                 service.add_resource(pnf_obj)
 206
 207     def assign_properties(self, service: Service) -> None:
 208         """Assign components properties.
 209
 210         For each component set properties and it's value if are declared
 211             in YAML template.
 212
 213         Args:
 214             service (Service): Service object
 215
 216         """
 217         if "networks" in self.yaml_template[self.service_name]:
 218             for net in self.yaml_template[self.service_name]["networks"]:
 219                 if "properties" in net:
 220                     vl: Vl = Vl(name=net['vl_name'])
 221                     vl_component: Component = service.get_component(vl)
 222                     self.assign_properties_to_component(vl_component, net["properties"])
 223         if "vnfs" in self.yaml_template[self.service_name]:
 224             for vnf in self.yaml_template[self.service_name]["vnfs"]:
 225                 if "properties" in vnf:
 226                     vf: Vf = Vf(name=vnf["vnf_name"])
 227                     vf_component: Component = service.get_component(vf)
 228                     self.assign_properties_to_component(vf_component, vnf["properties"])
 229         if "pnfs" in self.yaml_template[self.service_name]:
 230             for pnf in self.yaml_template[self.service_name]["pnfs"]:
 231                 if "properties" in pnf:
 232                     pnf_obj: Pnf = Pnf(name=pnf["pnf_name"])
 233                     pnf_component: Component = service.get_component(pnf_obj)
 234                     self.assign_properties_to_component(pnf_component, pnf["properties"])
 235
 236     def assign_properties_to_component(self,
 237                                        component: Component,
 238                                        component_properties: Dict[str, Any]) -> None:
 239         """Assign properties to component.
 240
 241         Args:
 242             component (Component): Component to which properites are going to be assigned
 243             component_properties (Dict[str, Any]): Properties dictionary
 244
 245         """
 246         for property_name, property_value in component_properties.items():
 247             prop: ComponentProperty = component.get_property(property_name)
 248             prop.value = property_value