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