--- /dev/null
+from pathlib import Path
+
+from .settings import * # pylint: disable=W0614
+
+ONLY_INSTANTIATE = False
+CLEANUP_FLAG = True
+
+VENDOR_NAME = "pnf_macro_vendor"
+SERVICE_NAME = "test_pnf_macro"
+SERVICE_INSTANCE_NAME = "TestPNFMacroInstantiation"
+SERVICE_YAML_TEMPLATE = Path(Path(__file__).parent.parent, "templates/vnf-services/pnf-service.yaml")
+
+CDS_DD_FILE = Path(Path(__file__).parent.parent, "templates/artifacts/dd.json")
+CDS_CBA_UNENRICHED = Path(Path(__file__).parent.parent, "templates/artifacts/PNF_DEMO.zip")
+CDS_CBA_ENRICHED = "/tmp/PNF_DEMO_enriched.zip"
+
+GLOBAL_CUSTOMER_ID = "pnf_macrocustomer"
+OWNING_ENTITY = "pnf_macro_owning_entity"
+PROJECT = "pnf_macro_project"
+LINE_OF_BUSINESS = "pnf_macro_line_of_business"
+PLATFORM = "pnf_macro_platform"
+
+INSTANTIATION_TIMEOUT = 600
+
+PNF_VES_CONFIG = dict(
+ count=1,
+ vesprotocol="https",
+ vesip="", # Due to it's not possible to get these value from SDK settings now it's going to be updated later
+ vesport="", # Due to it's not possible to get these value from SDK settings now it's going to be updated later
+ vesresource="eventListener",
+ vesversion="v7",
+ ipstart="10.11.0.16",
+ user="sample1",
+ password="sample1",
+ ipfileserver="127.0.0.1",
+ typefileserver="sftp",
+)
+PNF_CUSTOM_DATA = dict(
+ commonEventHeaderParams=dict(
+ sourceName=SERVICE_INSTANCE_NAME,
+ reportingEntityName=SERVICE_INSTANCE_NAME
+ )
+)
--- /dev/null
+"""Instantiate service with PNF using SO macro flow."""
+from yaml import load
+
+from xtesting.core import testcase
+from onapsdk.configuration import settings
+
+from onaptests.steps.base import YamlTemplateBaseStep
+from onaptests.steps.onboard.cds import CbaEnrichStep
+from onaptests.steps.simulator.pnf.pnf_register import PNFRegisterStep
+from onaptests.steps.instantiate.service_macro import YamlTemplateServiceMacroInstantiateStep
+
+
+class PnfMacroScenarioStep(YamlTemplateBaseStep):
+ """Step created to run scenarion and generate report."""
+
+ def __init__(self, cleanup=False):
+ """Initialize step.
+
+ Substeps:
+ - YamlTemplateServiceAlaCarteInstantiateStep.
+ """
+ super().__init__(cleanup=cleanup)
+ self._yaml_template: dict = None
+ self.add_step(PNFRegisterStep(
+ cleanup=settings.CLEANUP_FLAG
+ ))
+ self.add_step(CbaEnrichStep(
+ 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 "PNF 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 PnfMacro(testcase.TestCase):
+ """Run PNF simulator and onboard then instantiate a service with PNF."""
+
+ def __init__(self, **kwargs):
+ """Init Basic Network use case."""
+ if "case_name" not in kwargs:
+ kwargs["case_name"] = 'pnf_macro'
+ super().__init__(**kwargs)
+ self.__logger.debug("PnfMacro init started")
+ self.test = PnfMacroScenarioStep(cleanup=settings.CLEANUP_FLAG)
+
+ def run(self):
+ """Run PNF macro test."""
+ self.test.execute()
+ self.test.cleanup()
+
+ def clean(self):
+ """Generate report."""
+ self.test.reports_collection.generate_report()
--- /dev/null
+
+import time
+from uuid import uuid4
+from onapsdk.aai.business.service import ServiceInstance
+from yaml import load
+
+from onapsdk.aai.business.customer import Customer, ServiceSubscription
+from onapsdk.aai.business.owning_entity import OwningEntity
+from onapsdk.aai.cloud_infrastructure.cloud_region import CloudRegion
+from onapsdk.aai.cloud_infrastructure.tenant import Tenant
+from onapsdk.configuration import settings
+from onapsdk.sdc.service import Service
+from onapsdk.so.instantiation import ServiceInstantiation
+from onapsdk.vid import LineOfBusiness, Platform, Project
+from onaptests.steps.cloud.customer_service_subscription_create import CustomerServiceSubscriptionCreateStep
+
+import onaptests.utils.exceptions as onap_test_exceptions
+from onaptests.steps.base import YamlTemplateBaseStep
+from onaptests.steps.onboard.service import YamlTemplateServiceOnboardStep
+from onaptests.steps.cloud.connect_service_subscription_to_cloud_region import ConnectServiceSubToCloudRegionStep
+
+
+class YamlTemplateServiceMacroInstantiateStep(YamlTemplateBaseStep):
+ """Instantiate service a'la carte using YAML template."""
+
+ def __init__(self, cleanup=False):
+ """Initialize step.
+
+ Substeps:
+ - YamlTemplateServiceOnboardStep,
+ - ConnectServiceSubToCloudRegionStep,
+ - CustomerServiceSubscriptionCreateStep.
+ """
+ super().__init__(cleanup=cleanup)
+ self._yaml_template: dict = None
+ self._service_instance_name: str = None
+ self._service_instance: str = None
+ 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
+ self.add_step(CustomerServiceSubscriptionCreateStep(cleanup))
+
+
+ @property
+ def description(self) -> str:
+ """Step description."""
+ return "Instantiate service described in YAML using SO macro method."
+
+ @property
+ def component(self) -> str:
+ """Component name."""
+ return "SO"
+
+ @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
+
+ @YamlTemplateBaseStep.store_state
+ def execute(self):
+ """Instantiate service.
+
+ Use settings values:
+ - GLOBAL_CUSTOMER_ID,
+ - CLOUD_REGION_CLOUD_OWNER,
+ - CLOUD_REGION_ID,
+ - TENANT_ID,
+ - OWNING_ENTITY,
+ - PROJECT.
+
+ Raises:
+ Exception: Service instantiation failed
+
+ """
+ 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]:
+ cloud_region: CloudRegion = CloudRegion.get_by_id(
+ cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER,
+ cloud_region_id=settings.CLOUD_REGION_ID,
+ )
+ tenant: Tenant = cloud_region.get_tenant(settings.TENANT_ID)
+ else:
+ cloud_region, tenant = None, None # Only PNF is going to be instantiated so
+ # neither cloud_region nor tenant are needed
+ try:
+ owning_entity = OwningEntity.get_by_owning_entity_name(
+ settings.OWNING_ENTITY)
+ except ValueError:
+ self._logger.info("Owning entity not found, create it")
+ owning_entity = OwningEntity.create(settings.OWNING_ENTITY)
+ vid_project: Project = Project(settings.PROJECT)
+ line_of_business: LineOfBusiness = LineOfBusiness(settings.LINE_OF_BUSINESS)
+ platform: Platform = Platform(settings.PLATFORM)
+
+ # Before instantiating, be sure that the service has been distributed
+ self._logger.info("******** Check Service Distribution *******")
+ distribution_completed = False
+ nb_try = 0
+ nb_try_max = 10
+ while distribution_completed is False and nb_try < nb_try_max:
+ distribution_completed = service.distributed
+ if distribution_completed is True:
+ self._logger.info(
+ "Service Distribution for %s is sucessfully finished",
+ service.name)
+ break
+ self._logger.info(
+ "Service Distribution for %s ongoing, Wait for 60 s",
+ service.name)
+ time.sleep(60)
+ nb_try += 1
+
+ if distribution_completed is False:
+ self._logger.error(
+ "Service Distribution for %s failed !!",service.name)
+ raise onap_test_exceptions.ServiceDistributionException
+
+ service_instantiation = ServiceInstantiation.instantiate_macro(
+ service,
+ customer=customer,
+ owning_entity=owning_entity,
+ project=vid_project,
+ line_of_business=line_of_business,
+ platform=platform,
+ cloud_region=cloud_region,
+ tenant=tenant,
+ service_instance_name=self.service_instance_name
+ )
+ service_instantiation.wait_for_finish(timeout=settings.INSTANTIATION_TIMEOUT)
+ if service_instantiation.failed:
+ raise onap_test_exceptions.ServiceInstantiateException
+ else:
+ 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)
+
+
+ def cleanup(self) -> None:
+ """Cleanup Service.
+
+ Raises:
+ Exception: Service cleaning failed
+
+ """
+ service_deletion = self._service_instance.delete()
+ service_deletion.wait_for_finish(timeout=600)
+ 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()
self.yaml_template[self.service_name]["instantiation_type"])
else:
instantiation_type: ServiceInstantiationType = ServiceInstantiationType.A_LA_CARTE
- service: Service = Service(name=settings.SERVICE_NAME, instantiation_type=instantiation_type)
+ service: Service = Service(name=self.service_name, instantiation_type=instantiation_type)
service.create()
self.declare_resources(service)
self.assign_properties(service)
utils.bootstrap_simulator()
utils.run_container()
- @BaseStep.store_state
def cleanup(self) -> None:
"""Remove containers and images."""
utils.stop_container()
import os
import sys
import time
+import urllib.parse
import yaml
from ipaddress import ip_address
from typing import Dict, Optional
from decorator import decorator
import docker
+
+from onapsdk.configuration import settings
from onaptests.masspnfsimulator.MassPnfSim import (
MassPnfSim, get_parser
)
# collect settings that will be placed in the simulator directory
vesprotocol = config["setup"].get('vesprotocol', "http")
- vesip = config["setup"].get('vesip', "")
- vesport = config["setup"].get('vesport', "")
+ ves_url = urllib.parse.urlparse(settings.VES_URL)
+ vesip = ves_url.hostname
+ vesport = ves_url.port
vesresource = config["setup"].get('vesresource', "")
vesversion = config["setup"].get('vesversion', "")
--- /dev/null
+---
+test_pnf_macro:
+ tosca_file_from_SDC: service-basic_network-template
+ version: "1.0"
+ subscription_type: "net"
+ instantiation_type: "Macro"
+ pnfs:
+ - pnf_name: "test-pnf"
+ pnf_artifact_type: "CONTROLLER_BLUEPRINT_ARCHIVE"
+ pnf_artifact_name: "CBA_enriched.zip"
+ pnf_artifact_label: "cbapnf"
+ pnf_artifact_file_path: "/tmp/PNF_DEMO_enriched.zip"
+ properties:
+ controller_actor: "CDS"
+ skip_post_instantiation_configuration: False
+ sdnc_artifact_name: "test"
+ sdnc_model_version: "1.0.0"
+ sdnc_model_name: "CBA_PNF_SIM"
--- /dev/null
+---
+test_pnfs:
+ pnfs:
+ - pnf_name: pnf
--- /dev/null
+---
+test_vnfs:
+ vnfs:
+ - vnf_name: vnf
+ pnfs:
+ - pnf_name: pnf
--- /dev/null
+from unittest import mock
+
+from onaptests.steps.instantiate.service_macro import (
+ YamlTemplateServiceMacroInstantiateStep
+)
+
+VNFS_PNFS_YAML = './tests/data/service_macro_template_vnfs.yaml'
+PNFS_YAML = './tests/data/service_macro_template_pnfs.yaml'
+
+
+@mock.patch("onaptests.steps.base.BaseStep.add_step")
+@mock.patch("onaptests.steps.instantiate.service_macro.settings")
+@mock.patch("onaptests.steps.instantiate.service_macro.YamlTemplateServiceOnboardStep")
+@mock.patch("onaptests.steps.instantiate.service_macro.ConnectServiceSubToCloudRegionStep")
+@mock.patch("onaptests.steps.instantiate.service_macro.CustomerServiceSubscriptionCreateStep")
+def test_are_vnfs(CustomerStep, CloudStep, OnboardStep, settings, add_step):
+
+ settings.SERVICE_YAML_TEMPLATE = VNFS_PNFS_YAML
+ settings.ONLY_INSTANTIATE = False
+
+ YamlTemplateServiceMacroInstantiateStep()
+
+ CustomerStep.assert_not_called()
+ assert add_step.mock_calls == [
+ mock.call(OnboardStep()), mock.call(CloudStep())]
+
+
+@mock.patch("onaptests.steps.base.BaseStep.add_step")
+@mock.patch("onaptests.steps.instantiate.service_macro.settings")
+@mock.patch("onaptests.steps.instantiate.service_macro.YamlTemplateServiceOnboardStep")
+@mock.patch("onaptests.steps.instantiate.service_macro.ConnectServiceSubToCloudRegionStep")
+@mock.patch("onaptests.steps.instantiate.service_macro.CustomerServiceSubscriptionCreateStep")
+def test_are_pnfs(CustomerStep, CloudStep, OnboardStep, settings, add_step):
+
+ settings.SERVICE_YAML_TEMPLATE = PNFS_YAML
+ settings.ONLY_INSTANTIATE = False
+
+ YamlTemplateServiceMacroInstantiateStep()
+
+ CloudStep.assert_not_called()
+ assert add_step.mock_calls == [
+ mock.call(OnboardStep()), mock.call(CustomerStep())]
Code Review / testsuite / pythonsdk-tests.git / commitdiff