src/onaptests/steps/instantiate/vf_module_ala_carte.py

   1 import time
   2 from typing import Iterable
   3 from uuid import uuid4
   4 from yaml import load
   5
   6 from onapsdk.aai.cloud_infrastructure import CloudRegion, Tenant
   7 from onapsdk.aai.business import Customer, ServiceInstance, ServiceSubscription
   8 from onapsdk.configuration import settings
   9 from onapsdk.so.instantiation import VnfParameter
  10
  11 import onaptests.utils.exceptions as onap_test_exceptions
  12 from ..base import YamlTemplateBaseStep
  13 from .vnf_ala_carte import YamlTemplateVnfAlaCarteInstantiateStep
  14 from .k8s_profile_create import K8SProfileStep
  15
  16 class YamlTemplateVfModuleAlaCarteInstantiateStep(YamlTemplateBaseStep):
  17     """Instantiate vf module a'la carte using YAML template."""
  18
  19     def __init__(self, cleanup=False):
  20         """Initialize step.
  21
  22         Substeps:
  23             - YamlTemplateVnfAlaCarteInstantiateStep.
  24         """
  25         super().__init__(cleanup=cleanup)
  26
  27         self._yaml_template: dict = None
  28         self._service_instance_name: str = None
  29         self._service_instance: ServiceInstance = None
  30         if settings.CLOUD_REGION_TYPE == settings.K8S_REGION_TYPE:
  31             # K8SProfileStep creates the requested profile and then calls
  32             # YamlTemplateVnfAlaCarteInstantiateStep step
  33             self.add_step(K8SProfileStep(cleanup))
  34         else:
  35             self.add_step(YamlTemplateVnfAlaCarteInstantiateStep(cleanup))
  36
  37
  38
  39     @property
  40     def yaml_template(self) -> dict:
  41         """Step YAML template.
  42
  43         Load from file if it's a root step, get from parent otherwise.
  44
  45         Returns:
  46             dict: Step YAML template
  47
  48         """
  49         if self.is_root:
  50             if not self._yaml_template:
  51                 with open(settings.SERVICE_YAML_TEMPLATE, "r") as yaml_template:
  52                     self._yaml_template: dict = load(yaml_template)
  53             return self._yaml_template
  54         return self.parent.yaml_template
  55
  56     @property
  57     def service_name(self) -> str:
  58         """Service name.
  59
  60         Get from YAML template if it's a root step, get from parent otherwise.
  61
  62         Returns:
  63             str: Service name
  64
  65         """
  66         if self.is_root:
  67             return next(iter(self.yaml_template.keys()))
  68         return self.parent.service_name
  69
  70     @property
  71     def service_instance_name(self) -> str:
  72         """Service instance name.
  73
  74         Generate using `service_name` and `uuid4()` function if it's a root step,
  75             get from parent otherwise.
  76
  77         Returns:
  78             str: Service instance name
  79
  80         """
  81         if self.is_root:
  82             if not self._service_instance_name:
  83                 self._service_instance_name: str = f"{self.service_name}-{str(uuid4())}"
  84             return self._service_instance_name
  85         return self.parent.service_instance_name
  86
  87     def get_vnf_parameters(self, vnf_name: str) -> Iterable[VnfParameter]:
  88         """Get VNF parameters from YAML template.
  89
  90         Args:
  91             vnf_name (str): VNF name to get parameters for.
  92
  93         Yields:
  94             Iterator[Iterable[VnfParameter]]: VNF parameter
  95
  96         """
  97
  98         # workaround, as VNF name differs from model name (added " 0")
  99         vnf_name = vnf_name.split()[0]
 100         for vnf in self.yaml_template[self.service_name]["vnfs"]:
 101             if vnf["vnf_name"] == vnf_name:
 102                 for vnf_parameter in vnf["vnf_parameters"]:
 103                     yield VnfParameter(
 104                         name=vnf_parameter["name"],
 105                         value=vnf_parameter["value"]
 106                     )
 107
 108     @YamlTemplateBaseStep.store_state
 109     def execute(self) -> None:
 110         """Instantiate Vf module.
 111
 112         Use settings values:
 113          - GLOBAL_CUSTOMER_ID.
 114
 115         Raises:
 116             Exception: Vf module instantiation failed
 117
 118         """
 119         super().execute()
 120         customer: Customer = Customer.get_by_global_customer_id(settings.GLOBAL_CUSTOMER_ID)
 121         service_subscription: ServiceSubscription = customer.get_service_subscription_by_service_type(self.service_name)
 122         self._service_instance: ServiceInstance = service_subscription.get_service_instance_by_name(self.service_instance_name)
 123         cloud_region: CloudRegion = CloudRegion.get_by_id(
 124             cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER,
 125             cloud_region_id=settings.CLOUD_REGION_ID,
 126         )
 127         tenant: Tenant = cloud_region.get_tenant(settings.TENANT_ID)
 128
 129         for vnf_instance in self._service_instance.vnf_instances:
 130             # possible to have several moduels for 1 VNF
 131             for vf_module in vnf_instance.vnf.vf_modules:
 132                 vf_module_instantiation = vnf_instance.add_vf_module(
 133                     vf_module,
 134                     cloud_region,
 135                     tenant,
 136                     self._service_instance_name,
 137                     vnf_parameters=self.get_vnf_parameters(vnf_instance.vnf.name))
 138             while not vf_module_instantiation.finished:
 139                 time.sleep(10)
 140             if vf_module_instantiation.failed:
 141                 raise onap_test_exceptions.VfModuleInstantiateException
 142
 143
 144     def cleanup(self) -> None:
 145         """Cleanup Vf module.
 146
 147         Raises:
 148             Exception: Vf module cleaning failed
 149
 150         """
 151         for vnf_instance in self._service_instance.vnf_instances:
 152             self._logger.debug("VNF instance %s found in Service Instance ",
 153                                vnf_instance.name)
 154             self._logger.info("Get VF Modules")
 155             for vf_module in vnf_instance.vf_modules:
 156                 self._logger.info("Delete VF Module %s",
 157                                   vf_module.name)
 158                 vf_module_deletion = vf_module.delete()
 159                 nb_try = 0
 160                 nb_try_max = 30
 161
 162                 while not vf_module_deletion.finished and nb_try < nb_try_max:
 163                     self._logger.info("Wait for vf module deletion")
 164                     nb_try += 1
 165                     time.sleep(20)
 166                 if vf_module_deletion.finished:
 167                     self._logger.info("VfModule %s deleted", vf_module.name)
 168                 else:
 169                     self._logger.error("VfModule deletion %s failed", vf_module.name)
 170                     raise onap_test_exceptions.VfModuleCleanupException
 171         super.cleanup()