[entry_points]
xtesting.testcase =
basic_vm = onaptests.scenario.basic_vm:BasicVm
+ basic_vm_macro = onaptests.scenario.basic_vm_macro:BasicVmMacro
basic_network = onaptests.scenario.basic_network:BasicNetwork
basic_cnf = onaptests.scenario.basic_cnf:BasicCnf
basic_cds = onaptests.scenario.cds_blueprint_enrichment:CDSBlueprintEnrichment
--- /dev/null
+import os
+import openstack
+from pathlib import Path
+
+from yaml import load
+
+import onaptests.utils.exceptions as onap_test_exceptions
+from .settings import * # pylint: disable=W0614
+
+
+CLEANUP_FLAG = True
+
+CDS_DD_FILE = Path(Path(__file__).parent.parent, "templates/artifacts/dd.json")
+CDS_CBA_UNENRICHED = Path(Path(__file__).parent.parent, "templates/artifacts/basic_vm_cba.zip")
+CDS_CBA_ENRICHED = "/tmp/BASIC_VM_enriched.zip"
+
+ONLY_INSTANTIATE = False
+USE_MULTICLOUD = True
+
+VENDOR_NAME = "basicvm_macro_vendor"
+
+CLOUD_REGION_CLOUD_OWNER = "basicvm-macro-cloud-owner"
+CLOUD_REGION_TYPE = "openstack"
+CLOUD_REGION_VERSION = "pike"
+CLOUD_OWNER_DEFINED_TYPE = "N/A"
+
+AVAILABILITY_ZONE_NAME = "basicvm-macro-availability-zone"
+AVAILABILITY_ZONE_TYPE = "nova"
+COMPLEX_PHYSICAL_LOCATION_ID = "lannion"
+COMPLEX_DATA_CENTER_CODE = "1234-5"
+
+GLOBAL_CUSTOMER_ID = "basicvm-customer"
+
+TEST_CLOUD = os.getenv('OS_TEST_CLOUD')
+cloud = openstack.connect(cloud=TEST_CLOUD)
+VIM_USERNAME = cloud.config.auth.get('username','Fill me')
+VIM_PASSWORD = cloud.config.auth.get('password','Fill me')
+VIM_SERVICE_URL = cloud.config.auth.get('auth_url','Fill me')
+TENANT_ID = cloud.config.auth.get('project_id','Fill me')
+TENANT_NAME = cloud.config.auth.get('project_name','Fill me')
+CLOUD_REGION_ID = cloud.config.auth.get('region_name','RegionOne')
+CLOUD_DOMAIN = cloud.config.auth.get('project_domain_name','Default')
+
+OWNING_ENTITY = "basicvm-oe"
+PROJECT = "basicvm-project"
+LINE_OF_BUSINESS = "basicvm-lob"
+PLATFORM = "basicvm-platform"
+CLOUD_DOMAIN = "Default"
+SERVICE_YAML_TEMPLATE = Path(Path(__file__).parent.parent, "templates/vnf-services/" +
+ "basic_vm_macro-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 (FileNotFoundError, ValueError):
+ raise onap_test_exceptions.TestConfigurationException
+
+SERVICE_INSTANCE_NAME = "basic_vm_macro_service_instance"
"templates/artifacts/profile.tar.gz")
PNF_PROFILE_NAME = f"pnf-cnf-profile-{str(uuid4())}"
K8S_VERSION = "1.0"
-K8S_CONFIG = Path(Path(__file__).parent.parent, "templates/artifacts/config")
+K8S_CONFIG = str(Path(Path(__file__).parent.parent, "templates/artifacts/config"))
SERVICE_INSTANCE_NAME = "TestPNFMacroInstantiation"
--- /dev/null
+"""Instantiate basic vm using SO macro flow."""
+import logging
+import time
+from yaml import load
+
+from onapsdk.configuration import settings
+from onapsdk.exceptions import SDKException
+from xtesting.core import testcase
+
+from onaptests.steps.base import YamlTemplateBaseStep
+from onaptests.steps.onboard.cds import CbaPublishStep
+from onaptests.utils.exceptions import OnapTestException
+from onaptests.steps.instantiate.service_macro import YamlTemplateServiceMacroInstantiateStep
+
+
+class BasicVmMacroStep(YamlTemplateBaseStep):
+
+ def __init__(self, cleanup=False):
+ """Initialize step.
+
+ Substeps:
+ - CbaPublishStep
+ - YamlTemplateServiceAlaCarteInstantiateStep.
+ """
+ super().__init__(cleanup=cleanup)
+ self._yaml_template: dict = None
+ self.add_step(CbaPublishStep(
+ cleanup=settings.CLEANUP_FLAG
+ ))
+ self.add_step(YamlTemplateServiceMacroInstantiateStep(
+ cleanup=settings.CLEANUP_FLAG
+ ))
+
+ @property
+ def description(self) -> str:
+ """Step description.
+
+ Used for reports
+
+ Returns:
+ str: Step description
+
+ """
+ return "Basic VM macro scenario step"
+
+ @property
+ def component(self) -> str:
+ """Component name.
+
+ Name of component which step is related with.
+ Most is the name of ONAP component.
+
+ Returns:
+ str: Component name
+
+ """
+ return "PythonSDK-tests"
+
+ @property
+ def yaml_template(self) -> dict:
+ """YAML template abstract property.
+
+ Every YAML template step need to implement that property.
+
+ Returns:
+ dict: YAML template
+
+ """
+ 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
+
+ @property
+ def service_name(self) -> dict:
+ """Service name.
+
+ Get from YAML template.
+
+ Returns:
+ str: Service name
+
+ """
+ return next(iter(self.yaml_template.keys()))
+
+ @property
+ def service_instance_name(self) -> str:
+ """Service instance name.
+
+ Returns:
+ str: Service instance name
+
+ """
+ return settings.SERVICE_INSTANCE_NAME
+
+
+class BasicVmMacro(testcase.TestCase):
+ """Instantiate a basic vm macro."""
+
+ __logger = logging.getLogger(__name__)
+
+ def __init__(self, **kwargs):
+ """Init Basic Macro use case."""
+ if "case_name" not in kwargs:
+ kwargs["case_name"] = 'basic_vm_macro'
+ super().__init__(**kwargs)
+ self.__logger.debug("Basic VM macro init started")
+ self.test = BasicVmMacroStep(cleanup=settings.CLEANUP_FLAG)
+
+ def run(self):
+ """Run basic vm macro test."""
+ self.start_time = time.time()
+ try:
+ self.test.execute()
+ self.test.cleanup()
+ self.result = 100
+ except (OnapTestException, SDKException) as exc:
+ self.result = 0
+ self.__logger.error(exc.error_message)
+ finally:
+ self.stop_time = time.time()
+
+ def clean(self):
+ """Generate report."""
+ self.test.reports_collection.generate_report()
import time
from abc import ABC, abstractmethod
-from typing import Iterator, List
+from typing import Iterator, List, Optional
from onapsdk.aai.business import Customer
from onapsdk.configuration import settings
from onapsdk.exceptions import SDKException, SettingsError
try:
if cleanup:
self._start_cleanup_time = time.time()
+ execution_status: Optional[ReportStepStatus] = None
ret = fun(self, *args, **kwargs)
- execution_status: ReportStepStatus = ReportStepStatus.PASS
+ execution_status = ReportStepStatus.PASS
return ret
except SubstepExecutionException:
- execution_status: ReportStepStatus = ReportStepStatus.PASS if cleanup else ReportStepStatus.NOT_EXECUTED
+ execution_status = ReportStepStatus.PASS if cleanup else ReportStepStatus.NOT_EXECUTED
raise
except (OnapTestException, SDKException):
- execution_status: ReportStepStatus = ReportStepStatus.FAIL
+ execution_status = ReportStepStatus.FAIL
raise
finally:
if cleanup:
self._start_execution_time = time.time()
self._execution_report = Report(
step_description=f"[{self.component}] {self.name}: {self.description}",
- step_execution_status=execution_status,
+ step_execution_status=execution_status if execution_status else ReportStepStatus.FAIL,
step_execution_duration=time.time() - self._start_execution_time
)
return wrapper
- TENANT_NAME.
"""
super().execute()
- try:
- cloud_region: CloudRegion = CloudRegion.get_by_id(
- cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER,
- cloud_region_id=settings.CLOUD_REGION_ID)
- except ResourceNotFound:
- cloud_region.add_esr_system_info(
- esr_system_info_id=str(uuid4()),
- user_name=settings.VIM_USERNAME,
- password=settings.VIM_PASSWORD,
- system_type="VIM",
- service_url=settings.VIM_SERVICE_URL,
- ssl_insecure=False,
- system_status="active",
- cloud_domain=settings.CLOUD_DOMAIN,
- default_tenant=settings.TENANT_NAME
- )
- if settings.USE_MULTICLOUD:
- self._logger.info("*Multicloud registration *")
- cloud_region.register_to_multicloud()
- time.sleep(20)
- nb_try = 0
- nb_try_max = 3
- while nb_try < nb_try_max:
+ cloud_region: CloudRegion = CloudRegion.get_by_id(
+ cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER,
+ cloud_region_id=settings.CLOUD_REGION_ID)
+ cloud_region.add_esr_system_info(
+ esr_system_info_id=str(uuid4()),
+ user_name=settings.VIM_USERNAME,
+ password=settings.VIM_PASSWORD,
+ system_type="VIM",
+ service_url=settings.VIM_SERVICE_URL,
+ ssl_insecure=False,
+ system_status="active",
+ cloud_domain=settings.CLOUD_DOMAIN,
+ default_tenant=settings.TENANT_NAME
+ )
+ if settings.USE_MULTICLOUD:
+ self._logger.info("*Multicloud registration *")
+ cloud_region.register_to_multicloud()
+ time.sleep(20)
+ nb_try = 0
+ nb_try_max = 3
+ while nb_try < nb_try_max:
+ try:
if not cloud_region.tenants:
+ self._logger.debug("No tenats available, check one more time")
time.sleep(20)
else:
break
- nb_try += 1
+ except ResourceNotFound:
+ self._logger.debug("No tenats available, check one more time")
+ time.sleep(20)
+ nb_try += 1
# Retrieve the tenant, created by multicloud registration
# if it does not exist, create it
cloud_region.get_tenant(settings.TENANT_ID)
except ResourceNotFound:
self._logger.warning("Impossible to retrieve the Specificed Tenant")
- self._logger.debug("If no multicloud selected, add the tenant")
- cloud_region.add_tenant(
- tenant_id=settings.TENANT_ID,
- tenant_name=settings.TENANT_NAME)
+ self._logger.debug("If no multicloud selected, add the tenant, reraise otherwise")
+ if not settings.USE_MULTICLOUD:
+ cloud_region.add_tenant(
+ tenant_id=settings.TENANT_ID,
+ tenant_name=settings.TENANT_NAME)
+ else:
+ raise
# be sure that an availability zone has been created
# if not, create it
import time
+from typing import List
from uuid import uuid4
from onapsdk.aai.business.service import ServiceInstance
from yaml import load
from onapsdk.configuration import settings
from onapsdk.exceptions import ResourceNotFound
from onapsdk.sdc.service import Service
-from onapsdk.so.instantiation import ServiceInstantiation
+from onapsdk.so.instantiation import InstantiationParameter, ServiceInstantiation, VfmoduleParameters, VnfParameters
from onapsdk.vid import LineOfBusiness, Platform, Project
from onaptests.steps.cloud.customer_service_subscription_create import CustomerServiceSubscriptionCreateStep
if not settings.ONLY_INSTANTIATE:
self.add_step(YamlTemplateServiceOnboardStep(cleanup))
- are_pnfs: bool = "pnfs" in self.yaml_template[self.service_name]
if any(
filter(lambda x: x in self.yaml_template[self.service_name].keys(),
["vnfs", "networks"])):
# can additionally contain "pnfs", no difference
self.add_step(ConnectServiceSubToCloudRegionStep(cleanup))
- elif are_pnfs: # only pnfs
+ else: # only pnfs
self.add_step(CustomerServiceSubscriptionCreateStep(cleanup))
super().execute()
service = Service(self.service_name)
customer: Customer = Customer.get_by_global_customer_id(settings.GLOBAL_CUSTOMER_ID)
- if any(["networks", "vnfs"]) in self.yaml_template[self.service_name]:
+ if any(
+ filter(lambda x: x in self.yaml_template[self.service_name].keys(),
+ ["vnfs", "networks"])):
cloud_region: CloudRegion = CloudRegion.get_by_id(
cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER,
cloud_region_id=settings.CLOUD_REGION_ID,
"Service Distribution for %s failed !!",service.name)
raise onap_test_exceptions.ServiceDistributionException
+ vnf_params_list: List[VnfParameters] = []
+ for vnf_data in self.yaml_template[self.service_name].get("vnfs", []):
+ vnf_params_list.append(VnfParameters(
+ vnf_data["vnf_name"],
+ [InstantiationParameter(name=parameter["name"], value=parameter["value"]) for parameter in vnf_data.get("vnf_parameters", [])],
+ [VfmoduleParameters(vf_module_data["vf_module_name"],
+ [InstantiationParameter(name=parameter["name"], value=parameter["value"]) for parameter in vf_module_data.get("parameters", [])]) \
+ for vf_module_data in vnf_data.get("vf_module_parameters", [])]
+ ))
+
service_instantiation = ServiceInstantiation.instantiate_macro(
service,
customer=customer,
platform=platform,
cloud_region=cloud_region,
tenant=tenant,
- service_instance_name=self.service_instance_name
+ service_instance_name=self.service_instance_name,
+ vnf_parameters=vnf_params_list,
+ enable_multicloud=settings.USE_MULTICLOUD
)
try:
service_instantiation.wait_for_finish(timeout=settings.ORCHESTRATION_REQUEST_TIMEOUT)
Exception: Service cleaning failed
"""
- service_deletion = self._service_instance.delete()
- try:
- service_deletion.wait_for_finish(timeout=settings.ORCHESTRATION_REQUEST_TIMEOUT)
- except TimeoutError:
- self._logger.error("Service deletion %s timed out", self._service_instance_name)
- raise onap_test_exceptions.ServiceCleanupException
- if service_deletion.finished:
- self._logger.info("Service %s deleted", self._service_instance_name)
- else:
- self._logger.error("Service deletion %s failed", self._service_instance_name)
- raise onap_test_exceptions.ServiceCleanupException
+ if self._service_instance:
+ service_deletion = self._service_instance.delete()
+ try:
+ service_deletion.wait_for_finish(timeout=settings.ORCHESTRATION_REQUEST_TIMEOUT)
+ except TimeoutError:
+ self._logger.error("Service deletion %s timed out", self._service_instance_name)
+ raise onap_test_exceptions.ServiceCleanupException
+ if service_deletion.finished:
+ self._logger.info("Service %s deleted", self._service_instance_name)
+ else:
+ self._logger.error("Service deletion %s failed", self._service_instance_name)
+ raise onap_test_exceptions.ServiceCleanupException
super().cleanup()
"""
super().execute()
blueprint: Blueprint = Blueprint.load_from_file(settings.CDS_CBA_UNENRICHED)
- blueprint.enrich()
- blueprint.save(settings.CDS_CBA_ENRICHED)
+ enriched: Blueprint = blueprint.enrich()
+ enriched.save(settings.CDS_CBA_ENRICHED)
@BaseStep.store_state(cleanup=True)
def cleanup(self) -> None:
"""
super().cleanup()
Path(settings.CDS_CBA_ENRICHED).unlink()
+
+
+class CbaPublishStep(CDSBaseStep):
+ """Publish CBA file step."""
+
+ def __init__(self, cleanup=False) -> None:
+ """Initialize CBA publish step."""
+ super().__init__(cleanup=cleanup)
+ self.add_step(CbaEnrichStep(cleanup=cleanup))
+
+ @property
+ def description(self) -> str:
+ """Step description."""
+ return "Publish CBA file."
+
+ @BaseStep.store_state
+ def execute(self) -> None:
+ """Enrich CBA file.
+
+ Use settings values:
+ - CDS_DD_FILE.
+
+ """
+ super().execute()
+ blueprint: Blueprint = Blueprint.load_from_file(settings.CDS_CBA_ENRICHED)
+ blueprint.publish()
for vnf in self.yaml_template["vnfs"]:
vsp: Vsp = Vsp(name=f"{vnf['vnf_name']}_VSP")
vf: Vf = Vf(name=vnf['vnf_name'], vsp=vsp)
+ if all([x in vnf for x in ["vnf_artifact_type",
+ "vnf_artifact_name",
+ "vnf_artifact_label",
+ "vnf_artifact_file_path"]]):
+ vf.create()
+ vf.add_deployment_artifact(
+ artifact_type=vnf["vnf_artifact_type"],
+ artifact_name=vnf["vnf_artifact_name"],
+ artifact_label=vnf["vnf_artifact_label"],
+ artifact=vnf["vnf_artifact_file_path"]
+ )
vf.onboard()
--- /dev/null
+---
+basic_vm_macro:
+ instantiation_type: "Macro"
+ vnfs:
+ - vnf_name: basic_vm_macro
+ properties:
+ controller_actor: "CDS"
+ skip_post_instantiation_configuration: False
+ sdnc_artifact_name: "vnf"
+ sdnc_model_version: "1.0.0"
+ sdnc_model_name: "ubuntu20"
+ vnf_artifact_type: "CONTROLLER_BLUEPRINT_ARCHIVE"
+ vnf_artifact_name: "BASIC_VM_enriched.zip"
+ vnf_artifact_label: "vnfcds"
+ vnf_artifact_file_path: "/tmp/BASIC_VM_enriched.zip"
+ heat_files_to_upload: onaptests/templates/heat-files/ubuntu20/ubuntu20agent.zip
+ vf_module_parameters:
+ - vf_module_name: base_ubuntu20
+ parameters: [
+ {"name": "ubuntu20_image_name",
+ "value": "Ubuntu_2004"
+ },
+ {"name": "ubuntu20_key_name",
+ "value": "cleouverte"
+ },
+ {"name": "ubuntu20_pub_key",
+ "value": "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAA\
+ BAQDY15cdBmIs2XOpe4EiFCsaY6bmUmK/GysMoLl4UG51JCfJwvwoWCoA+6mDIbymZxhxq9IGx\
+ ilp/yTA6WQ9s/5pBag1cUMJmFuda9PjOkXl04jgqh5tR6I+GZ97AvCg93KAECis5ubSqw1xOCj4\
+ utfEUtPoF1OuzqM/lE5mY4N6VKXn+fT7pCD6cifBEs6JHhVNvs5OLLp/tO8Pa3kKYQOdyS0xc3r\
+ h+t2lrzvKUSWGZbX+dLiFiEpjsUL3tDqzkEMNUn4pdv69OJuzWHCxRWPfdrY9Wg0j3mJesP29EBh\
+ t+w+EC9/kBKq+1VKdmsXUXAcjEvjovVL8l1BrX3BY0R8D imported-openssh-key"
+ },
+ {"name": "ubuntu20_flavor_name",
+ "value": "m1.smaller"
+ },
+ {"name": "VM_name",
+ "value": "ubuntu20agent-VM-01"
+ },
+ {"name": "vnf_id",
+ "value": "ubuntu20agent-VNF-instance"
+ },
+ {"name": "vf_module_id",
+ "value": "ubuntu20agent-vfmodule-instance"
+ },
+ {"name": "vnf_name",
+ "value": "ubuntu20agent-VNF"
+ },
+ {"name": "admin_plane_net_name",
+ "value": "admin"
+ },
+ {"name": "ubuntu20_name_0",
+ "value": "ubuntu20agent-VNF"}
+ ]
Code Review / testsuite / pythonsdk-tests.git / commitdiff