aboutsummaryrefslogtreecommitdiffstats
path: root/src/onaptests/steps/instantiate/k8s_profile_create.py
diff options
context:
space:
mode:
authorThierry Hardy <thierry.hardy@orange.com>2020-10-02 16:18:17 +0200
committerThierry Hardy <thierry.hardy@orange.com>2020-10-08 15:47:00 +0200
commit0123c525b0182a5d6f9cd1bee3830eb1956239c7 (patch)
tree2ba5a5b7ae88e53aa634cabd113338b42d49c8d0 /src/onaptests/steps/instantiate/k8s_profile_create.py
parentd1d44781c7d03c8794dd626184f77d9836264212 (diff)
Create basic_cnf test leveraging onapsdk
Add the scenario basic_cnf that uses multicloud-k8s plugin to instantiate cnf on k8s Add the removal of profile in cleanup part Add store_state Issue-ID: TEST-243 Signed-off-by: Thierry Hardy <thierry.hardy@orange.com> Change-Id: Ib743c259decf95cdc69975e1ef7d4ba7aadccfae
Diffstat (limited to 'src/onaptests/steps/instantiate/k8s_profile_create.py')
-rw-r--r--src/onaptests/steps/instantiate/k8s_profile_create.py165
1 files changed, 165 insertions, 0 deletions
diff --git a/src/onaptests/steps/instantiate/k8s_profile_create.py b/src/onaptests/steps/instantiate/k8s_profile_create.py
new file mode 100644
index 0000000..375aa2c
--- /dev/null
+++ b/src/onaptests/steps/instantiate/k8s_profile_create.py
@@ -0,0 +1,165 @@
+from typing import Iterable
+from uuid import uuid4
+from yaml import load
+
+from onapsdk.aai.business import Customer, ServiceInstance, ServiceSubscription
+from onapsdk.configuration import settings
+from onapsdk.msb.k8s import Definition
+from onapsdk.so.instantiation import VnfParameter
+
+from ..base import BaseStep
+from .vnf_ala_carte import YamlTemplateVnfAlaCarteInstantiateStep
+
+class K8SProfileStep(BaseStep):
+ """CreateK8sProfileStep."""
+
+ def __init__(self, cleanup=False):
+ """Initialize step.
+ """
+ super().__init__(cleanup=cleanup)
+
+ self._yaml_template: dict = None
+ self._service_instance_name: str = None
+ self._service_instance: ServiceInstance = None
+ self.add_step(YamlTemplateVnfAlaCarteInstantiateStep(cleanup))
+
+ @property
+ def yaml_template(self) -> dict:
+ """Step YAML template.
+
+ Load from file if it's a root step, get from parent otherwise.
+
+ Returns:
+ dict: Step YAML template
+
+ """
+ if self.is_root:
+ if not self._yaml_template:
+ with open(settings.SERVICE_YAML_TEMPLATE, "r") as yaml_template:
+ self._yaml_template: dict = load(yaml_template)
+ return self._yaml_template
+ return self.parent.yaml_template
+
+ @property
+ def service_name(self) -> str:
+ """Service name.
+
+ Get from YAML template if it's a root step, get from parent otherwise.
+
+ Returns:
+ str: Service name
+
+ """
+ if self.is_root:
+ return next(iter(self.yaml_template.keys()))
+ return self.parent.service_name
+
+ @property
+ def service_instance_name(self) -> str:
+ """Service instance name.
+
+ Generate using `service_name` and `uuid4()` function if it's a root step,
+ get from parent otherwise.
+
+ Returns:
+ str: Service instance name
+
+ """
+ if self.is_root:
+ if not self._service_instance_name:
+ self._service_instance_name: str = f"{self.service_name}-{str(uuid4())}"
+ return self._service_instance_name
+ return self.parent.service_instance_name
+
+ def get_vnf_parameters(self, vnf_name: str) -> Iterable[VnfParameter]:
+ """Get VNF parameters from YAML template.
+
+ Args:
+ vnf_name (str): VNF name to get parameters for.
+
+ Yields:
+ Iterator[Iterable[VnfParameter]]: VNF parameter
+
+ """
+
+ # workaround, as VNF name differs from model name (added " 0")
+ vnf_name = vnf_name.split()[0]
+ for vnf in self.yaml_template[self.service_name]["vnfs"]:
+ if vnf["vnf_name"] == vnf_name:
+ for vnf_parameter in vnf["vnf_parameters"]:
+ yield VnfParameter(
+ name=vnf_parameter["name"],
+ value=vnf_parameter["value"]
+ )
+
+ @BaseStep.store_state
+ def execute(self):
+ """Creation of k8s profile for resource bundle definition
+
+ Use settings values:
+ - GLOBAL_CUSTOMER_ID
+ - K8S_PROFILE_K8S_VERSION
+ - K8S_PROFILE_ARTIFACT_PATH.
+ """
+ super().execute()
+ customer: Customer = Customer.get_by_global_customer_id(settings.GLOBAL_CUSTOMER_ID)
+ service_subscription: ServiceSubscription = customer.get_service_subscription_by_service_type(self.service_name)
+ self._service_instance: ServiceInstance = service_subscription.get_service_instance_by_name(self.service_instance_name)
+
+ for vnf_instance in self._service_instance.vnf_instances:
+ # possible to have several modules for 1 VNF
+ for vf_module in vnf_instance.vnf.vf_modules:
+ # Define profile (rb_profile) for resource bundle definition
+ # Retrieve resource bundle definition (rbdef) corresponding to vf module
+ rbdef_name = vf_module.metadata["vfModuleModelInvariantUUID"]
+ rbdef_version = vf_module.metadata["vfModuleModelUUID"]
+ rbdef = Definition.get_definition_by_name_version(rbdef_name, rbdef_version)
+ # Get k8s profile name from yaml service template
+ vnf_parameters = self.get_vnf_parameters(vnf_instance.vnf.name)
+ k8s_profile_name = ""
+ k8s_profile_namespace = ""
+ for param in vnf_parameters:
+ if param.name == "k8s-rb-profile-name":
+ k8s_profile_name = param.value
+ if param.name == "k8s-rb-profile-namespace":
+ k8s_profile_namespace = param.value
+ if k8s_profile_name == "" or k8s_profile_namespace == "":
+ raise Exception("Vf module instantiation failed, missing rb profile information")
+ ######## Check profile for Definition ###################################
+ try:
+ rbdef.get_profile_by_name(k8s_profile_name)
+ except ValueError:
+ ######## Create profile for Definition ###################################
+ profile = rbdef.create_profile(k8s_profile_name,
+ k8s_profile_namespace,
+ settings.K8S_PROFILE_K8S_VERSION)
+ ####### Upload artifact for created profile ##############################
+ profile.upload_artifact(open(settings.K8S_PROFILE_ARTIFACT_PATH, 'rb').read())
+
+ def cleanup(self) -> None:
+ """Cleanup K8S profiles.
+ """
+ self._logger.info("*Clean the k8s profile *")
+ for vnf_instance in self._service_instance.vnf_instances:
+ # possible to have several modules for 1 VNF
+ for vf_module in vnf_instance.vnf.vf_modules:
+ # Retrieve resource bundle definition (rbdef) corresponding to vf module
+ rbdef_name = vf_module.metadata["vfModuleModelInvariantUUID"]
+ rbdef_version = vf_module.metadata["vfModuleModelUUID"]
+ rbdef = Definition.get_definition_by_name_version(rbdef_name, rbdef_version)
+ # Get k8s profile name from yaml service template
+ vnf_parameters = self.get_vnf_parameters(vnf_instance.vnf.name)
+ k8s_profile_name = ""
+ for param in vnf_parameters:
+ if param.name == "k8s-rb-profile-name":
+ k8s_profile_name = param.value
+ if k8s_profile_name == "":
+ raise Exception("K8s profile deletion failed, missing rb profile name")
+ ######## Delete profile for Definition ###################################
+ try:
+ profile = rbdef.get_profile_by_name(k8s_profile_name)
+ profile.delete()
+ except ValueError:
+ self._logger.error("K8s profile deletion %s failed", k8s_profile_name)
+ raise Exception("K8s profile deletion failed")
+ super().cleanup()