+++ /dev/null
-import os
-import openstack
-import sys
-from yaml import load
-
-from .settings import * # pylint: disable=W0614
-
-""" Specific basic_cnf with multicloud-k8s and yaml config scenario."""
-
-# This scenario uses multicloud-k8s and not multicloud
-# (no registration requested)
-USE_MULTICLOUD = False
-# Set ONLY_INSTANTIATE to true to run an instantiation without repeating
-# onboarding and related AAI configuration (Cloud config)
-ONLY_INSTANTIATE= False
-
-# if a yaml file is define, retrieve info from this yaml files
-# if not declare the parameters in the settings
-SERVICE_YAML_TEMPLATE = (sys.path[-1] + "/onaptests/templates/vnf-services/" +
- "basic_cnf-service.yaml")
-
-try:
- # Try to retrieve the SERVICE NAME from the yaml file
- with open(SERVICE_YAML_TEMPLATE, "r") as yaml_template:
- yaml_config_file = load(yaml_template)
- SERVICE_NAME = next(iter(yaml_config_file.keys()))
-except ValueError:
- SERVICE_NAME = "" # Fill me
-
-CLEANUP_FLAG = True
-# nb of seconds before cleanup in case cleanup option is set
-CLEANUP_ACTIVITY_TIMER = 10
-
-# Definition of k8s profile version
-K8S_PROFILE_K8S_VERSION = "1.0"
-# Relative path to k8s profile artifact in the python package (so under /src)
-K8S_PROFILE_ARTIFACT_PATH = (sys.path[-1] +
- "/onaptests/templates/artifacts/k8sprof.tar.gz")
-# Relative path to config file to set k8s connectivity information
-K8S_KUBECONFIG_FILE = (sys.path[-1] +
- "/onaptests/templates/artifacts/config")
-
-VENDOR_NAME = "basicnf_vendor"
-
-CLOUD_REGION_CLOUD_OWNER = "basicnf-owner" # must not contain _
-CLOUD_REGION_ID = "k8sregion"
-CLOUD_REGION_TYPE = "k8s"
-CLOUD_REGION_VERSION = "1.0"
-CLOUD_DOMAIN = "Default"
-CLOUD_OWNER_DEFINED_TYPE = "t1"
-
-COMPLEX_PHYSICAL_LOCATION_ID = "lannion"
-COMPLEX_DATA_CENTER_CODE = "1234-5"
-AVAILABILITY_ZONE_NAME = "basicnf-availability-zone"
-AVAILABILITY_ZONE_TYPE = "nova"
-
-GLOBAL_CUSTOMER_ID = "basicnf-customer"
-
-OWNING_ENTITY = "basicnf_owning_entity"
-PROJECT = "basicnf_project"
-LINE_OF_BUSINESS = "basicnf_lob"
-PLATFORM = "basicnf_platform"
-
-SERVICE_INSTANCE_NAME = "basic_cnf_service_instance"
-
-# The cloud Part
-# Assuming a cloud.yaml is available, use the openstack client
-# to retrieve cloud info and avoid data duplication
-TEST_CLOUD = os.getenv('OS_TEST_CLOUD')
-try:
- if TEST_CLOUD is not None:
- cloud = openstack.connect(cloud=TEST_CLOUD)
- VIM_USERNAME = cloud.config.auth['username']
- VIM_PASSWORD = cloud.config.auth['password']
- VIM_SERVICE_URL = cloud.config.auth['auth_url']
- TENANT_ID = cloud.config.auth['project_id']
- TENANT_NAME = cloud.config.auth['project_name']
- else:
- raise KeyError
-except KeyError:
- # If you do not use the cloud.yaml as imput for your openstack
- # put the input data here
- # Note if 1 parameter is missing in the clouds.yaml, we fallback here
- "Dummy definition - not used"
- TENANT_ID = "123456"
- TENANT_NAME = "dummy_test"
- VIM_USERNAME = "dummy"
- VIM_PASSWORD = "dummy123"
- VIM_SERVICE_URL = "http://10.12.25.2:5000/v3" # Fill me
CLOUD_REGION_CLOUD_OWNER = "sdktestsOwner" # must not contain _
CLOUD_REGION_ID = "RegionOne" # should be valid, as otherwise MultiCloud fails
CLOUD_REGION_TYPE = "openstack"
-CLOUD_OWNER_DEFINED_TYPE = "N/A"
CLOUD_REGION_VERSION = "titanium_cloud"
CLOUD_DOMAIN = "Default"
CLOUD_REGION_CLOUD_OWNER = "basicvm-cloud-owner"
CLOUD_REGION_TYPE = "openstack"
CLOUD_REGION_VERSION = "openstack"
-CLOUD_OWNER_DEFINED_TYPE = "N/A"
AVAILABILITY_ZONE_NAME = "basicvm-availability-zone"
AVAILABILITY_ZONE_TYPE = "nova"
+++ /dev/null
-#!/usr/bin/env python
-"""Basic CNF test case."""
-import logging
-import time
-
-from xtesting.core import testcase
-from onapsdk.configuration import settings
-from onaptests.steps.instantiate.vf_module_ala_carte import YamlTemplateVfModuleAlaCarteInstantiateStep
-
-class BasicCNF(testcase.TestCase):
- """Onboard then instantiate a simple CNF with ONAP."""
-
- __logger = logging.getLogger(__name__)
-
- def __init__(self, **kwargs):
- """Init BasicCNF."""
- if "case_name" not in kwargs:
- kwargs["case_name"] = 'basic_cnf'
- super(BasicCNF, self).__init__(**kwargs)
- self.__logger.debug("BasicCNF init started")
- self.test = YamlTemplateVfModuleAlaCarteInstantiateStep(
- cleanup=settings.CLEANUP_FLAG)
- self.start_time = None
- self.stop_time = None
- self.result = 0
-
- def run(self):
- """Run onap_tests with basic_cnf VM."""
- self.start_time = time.time()
- self.__logger.debug("start time")
- self.test.execute()
- self.__logger.info("basic_cnf successfully created")
- self.stop_time = time.time()
- # The cleanup is part of the test, not only a teardown action
- if settings.CLEANUP_FLAG:
- self.__logger.info("basic_cnf cleanup called")
- time.sleep(settings.CLEANUP_ACTIVITY_TIMER)
- self.test.cleanup()
- self.result = 100
- else:
- self.__logger.info("No cleanup requested. Test completed.")
- self.result = 100
-
-
- def clean(self):
- """Clean Additional resources if needed."""
- pass
from .customer_service_subscription_create import CustomerServiceSubscriptionCreateStep
from .link_cloud_to_complex import LinkCloudRegionToComplexStep
from .register_cloud import RegisterCloudRegionStep
-from .k8s_connectivity_info_create import K8SConnectivityInfoStep
class ConnectServiceSubToCloudRegionStep(BaseStep):
"""
super().__init__(cleanup=cleanup)
- if settings.CLOUD_REGION_TYPE == "k8s":
- self.add_step(K8SConnectivityInfoStep(cleanup=cleanup))
self.add_step(RegisterCloudRegionStep(cleanup=cleanup))
self.add_step(LinkCloudRegionToComplexStep(cleanup=cleanup))
self.add_step(CustomerServiceSubscriptionCreateStep(cleanup=cleanup))
+++ /dev/null
-from onapsdk.configuration import settings
-from onapsdk.msb.k8s import ConnectivityInfo
-
-from ..base import BaseStep
-
-class K8SConnectivityInfoStep(BaseStep):
- """CreateConnnectivityInfoStep."""
-
- @BaseStep.store_state
- def execute(self):
- """Creation k8s connectivity information
-
- Use settings values:
- - CLOUD_REGION_ID,
- - CLOUD_REGION_CLOUD_OWNER,
- - K8S_KUBECONFIG_FILE.
- """
- super().execute()
- ######## Create Connectivity Info #########################################
- try:
- ConnectivityInfo.get_connectivity_info_by_region_id(settings.CLOUD_REGION_ID)
- except ValueError:
- ConnectivityInfo.create(settings.CLOUD_REGION_ID,
- settings.CLOUD_REGION_CLOUD_OWNER,
- open(settings.K8S_KUBECONFIG_FILE, 'rb').read())
-
- def cleanup(self) -> None:
- """Cleanup K8S Connectivity information.
- """
- self._logger.info("*Clean the k8s connectivity information *")
- super().cleanup()
- connectinfo = ConnectivityInfo.get_connectivity_info_by_region_id(settings.CLOUD_REGION_ID)
- connectinfo.delete()
- CLOUD_REGION_CLOUD_OWNER,
- CLOUD_REGION_ID,
- CLOUD_DOMAIN,
- - CLOUD_REGION_VERSION,
- - CLOUD_OWNER_DEFINED_TYPE,
- - COMPLEX_PHYSICAL_LOCATION_ID,
- VIM_USERNAME,
- VIM_PASSWORD,
- VIM_SERVICE_URL,
orchestration_disabled=False,
in_maint=False,
cloud_type=settings.CLOUD_REGION_TYPE,
- cloud_region_version=settings.CLOUD_REGION_VERSION,
- owner_defined_type=settings.CLOUD_OWNER_DEFINED_TYPE,
- complex_name=settings.COMPLEX_PHYSICAL_LOCATION_ID
- # cloud_extra_info=settings.CLOUD_EXTRA_INFO
+ cloud_region_version=settings.CLOUD_REGION_VERSION
)
cloud_region.add_esr_system_info(
esr_system_info_id=str(uuid4()),
+++ /dev/null
-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()
Exception: Service cleaning failed
"""
+ super().cleanup()
service_deletion = self._service_instance.delete()
nb_try = 0
nb_try_max = 30
else:
self._logger.error("Service deletion %s failed", self._service_instance_name)
raise Exception("Service cleanup failed")
- super().cleanup()
from ..base import YamlTemplateBaseStep
from .vnf_ala_carte import YamlTemplateVnfAlaCarteInstantiateStep
-from .k8s_profile_create import K8SProfileStep
class YamlTemplateVfModuleAlaCarteInstantiateStep(YamlTemplateBaseStep):
"""Instantiate vf module a'la carte using YAML template."""
self._yaml_template: dict = None
self._service_instance_name: str = None
self._service_instance: ServiceInstance = None
- if settings.CLOUD_REGION_TYPE == "k8s":
- self.add_step(K8SProfileStep(cleanup))
- else:
- self.add_step(YamlTemplateVnfAlaCarteInstantiateStep(cleanup))
-
-
+ self.add_step(YamlTemplateVnfAlaCarteInstantiateStep(cleanup))
@property
def yaml_template(self) -> dict:
"""
# workaround, as VNF name differs from model name (added " 0")
- vnf_name = vnf_name.split()[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"]:
cloud_region,
tenant,
self._service_instance_name,
- vnf_parameters=self.get_vnf_parameters(vnf_instance.vnf.name))
+ vnf_parameters= self.get_vnf_parameters(vnf_instance.vnf.name))
while not vf_module_instantiation.finished:
time.sleep(10)
if vf_module_instantiation.failed:
Exception: Vf module cleaning failed
"""
+ super().cleanup()
for vnf_instance in self._service_instance.vnf_instances:
self._logger.debug("VNF instance %s found in Service Instance ",
vnf_instance.name)
else:
self._logger.error("VfModule deletion %s failed", vf_module.name)
raise Exception("Vf module cleanup failed")
- super().cleanup()
Exception: VNF cleaning failed
"""
+ super().cleanup()
+
for vnf_instance in self._service_instance.vnf_instances:
vnf_deletion = vnf_instance.delete()
nb_try = 0
else:
self._logger.error("VNF deletion %s failed", vnf_instance.name)
raise Exception("VNF Cleanup failed")
- super().cleanup()
+++ /dev/null
----
-basic_cnf:
- tosca_file_from_SDC: service-basic_cnf-template
- version: "1.0"
- subscription_type: "basic_cnf"
- vnfs:
- - vnf_name: basic_cnf
- heat_files_to_upload: onaptests/templates/heat-files/basic_cnf/basic_cnf.zip
- vnf_parameters: [
- {"name": "dummy_name_0",
- "value": "dummy_name"
- },
- {"name": "k8s-rb-profile-name",
- "value": "cnftest"
- },
- {"name": "k8s-rb-profile-namespace",
- "value": "k8s"
- }
- ]
Code Review / testsuite / pythonsdk-tests.git / commitdiff