--- /dev/null
+from pathlib import Path
+from uuid import uuid4
+
+from .settings import * # pylint: disable=W0614
+
+ONLY_INSTANTIATE = False
+CLEANUP_FLAG = True
+USE_MULTICLOUD = False
+
+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"
+
+CLOUD_REGION_CLOUD_OWNER = "basicnf-owner" # must not contain _
+CLOUD_REGION_ID = "k8sregion"
+CLOUD_REGION_TYPE = "k8s"
+CLOUD_REGION_VERSION = "1.0"
+CLOUD_OWNER_DEFINED_TYPE = "N/A"
+COMPLEX_PHYSICAL_LOCATION_ID = "sdktests"
+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_DEFINITION_ATRIFACT_FILE_PATH = Path(Path(__file__).parent.parent,
+ "templates/artifacts/pnf-simulator.tar.gz")
+PNF_RB_NAME = f"pnf-cnf-rb-{str(uuid4())}"
+PNF_RB_VERSION = "v1"
+PNF_PROFILE_ARTIFACT_FILE_PATH = Path(Path(__file__).parent.parent,
+ "templates/artifacts/profile.tar.gz")
+PNF_PROFILE_NAME = f"pnf-cnf-profile-{str(uuid4())}"
+K8S_VERSION = "1.0"
+K8S_CONFIG = str(Path(Path(__file__).parent.parent, "templates/artifacts/config"))
+
+SERVICE_INSTANCE_NAME = "TestPNFMacroInstantiation"
+
+DCAE_VES_COLLECTOR_POD_NAME = "xdcae-ves-collector"
+PNF_WAIT_TIME = 60.0
+PNF_REGISTRATION_NUMBER_OF_TRIES = 5
from onapsdk.configuration import settings
from onapsdk.sdc.pnf import Pnf
from onapsdk.sdc.vendor import Vendor
+from onapsdk.sdc.vsp import Vsp
from ..base import BaseStep, YamlTemplateBaseStep
-from .vendor import VendorOnboardStep
+from .vsp import VspOnboardStep, YamlTemplateVspOnboardStep
class PnfOnboardStep(BaseStep):
"""
super().__init__(cleanup=cleanup)
- self.add_step(VendorOnboardStep(cleanup=cleanup))
+ self.add_step(VspOnboardStep(cleanup=cleanup))
@property
def description(self) -> str:
"""
super().__init__(cleanup=cleanup)
- self.add_step(VendorOnboardStep(cleanup=cleanup))
+ self.add_step(YamlTemplateVspOnboardStep(cleanup=cleanup))
@property
def description(self) -> str:
if "pnfs" in self.yaml_template:
vendor: Vendor = Vendor(name=settings.VENDOR_NAME)
for pnf in self.yaml_template["pnfs"]:
- pnf_obj: Pnf = Pnf(name=pnf["pnf_name"], vendor=vendor)
+ if "heat_files_to_upload" in pnf:
+ vsp: Vsp = Vsp(name=f"{pnf['pnf_name']}_VSP")
+ else:
+ vsp = None
+ pnf_obj: Pnf = Pnf(name=pnf["pnf_name"], vendor=vendor, vsp=vsp)
pnf_obj.create()
pnf_obj.add_deployment_artifact(
artifact_type=pnf["pnf_artifact_type"],
vendor=vendor,
package=package)
vsp.onboard()
+ elif "pnfs" in self.yaml_template:
+ for pnf in self.yaml_template["pnfs"]:
+ if "heat_files_to_upload" in pnf:
+ with open(
+ sys.path[-1] + "/" + pnf["heat_files_to_upload"], "rb") as package:
+ vsp: Vsp = Vsp(name=f"{pnf['pnf_name']}_VSP",
+ vendor=vendor,
+ package=package)
+ vsp.onboard()
--- /dev/null
+# http://www.apache.org/licenses/LICENSE-2.0
+"""PNF simulator registration module."""
+
+import time
+from typing import Tuple
+
+import requests
+from kubernetes import client, config, watch
+from onapsdk.configuration import settings
+
+from onaptests.steps.base import BaseStep
+from onaptests.steps.instantiate.msb_k8s import CreateInstanceStep
+from onaptests.utils.exceptions import EnvironmentPreparationException, OnapTestException
+
+
+class PnfSimulatorCnfRegisterStep(BaseStep):
+ """PNF simulator registration step."""
+
+ def __init__(self, cleanup: bool = False) -> None:
+ """Initialize step.
+
+ Substeps:
+ - CreateInstanceStep.
+ """
+ super().__init__(cleanup=cleanup)
+ self.add_step(CreateInstanceStep(cleanup=cleanup))
+
+ @property
+ def description(self) -> str:
+ """Step description."""
+ return "Register PNF simulator with VES."
+
+ @property
+ def component(self) -> str:
+ """Component name."""
+ return "Environment"
+
+ @staticmethod
+ def is_pnf_pod_running(timeout_seconds=120) -> bool:
+ """Check if PNF simulator pod is running.
+
+ Args:
+ timeout_seconds (int, optional): Timeout. Defaults to 120.
+
+ Returns:
+ bool: True if PNF simulator pod is running, False otherwise
+
+ """
+ config.load_kube_config(settings.K8S_CONFIG)
+ k8s_client: "CoreV1API" = client.CoreV1Api()
+ k8s_watch: "Watch" = watch.Watch()
+ for event in k8s_watch.stream(k8s_client.list_namespaced_pod,
+ namespace=settings.K8S_NAMESPACE,
+ timeout_seconds=timeout_seconds):
+ if event["object"].metadata.name == "pnf-simulator":
+ if not event["object"].status.phase in ["Pending", "Running"]:
+ # Invalid pod state
+ return False
+ return event["object"].status.phase == "Running"
+ return False
+
+ @staticmethod
+ def get_ves_ip_and_port() -> Tuple[str, str]:
+ """Static method to get VES ip address and port.
+
+ Raises:
+ EnvironmentPreparationException: VES pod is not running
+
+ Returns:
+ Tuple[str, str]: VES IP and port
+
+ """
+ config.load_kube_config(settings.K8S_CONFIG)
+ k8s_client: "CoreV1API" = client.CoreV1Api()
+ for service in k8s_client.list_namespaced_service(namespace=settings.K8S_NAMESPACE).items:
+ if service.metadata.name == settings.DCAE_VES_COLLECTOR_POD_NAME:
+ return service.spec.cluster_ip, service.spec.ports[0].port
+ raise EnvironmentPreparationException("Couldn't get VES ip and port")
+
+ @BaseStep.store_state
+ def execute(self) -> None:
+ """Send PNF registration event."""
+ super().execute()
+ if not self.is_pnf_pod_running():
+ EnvironmentPreparationException("PNF simulator is not running")
+ time.sleep(settings.PNF_WAIT_TIME) # Let's still wait for PNF simulator to make sure it's initialized
+ ves_ip, ves_port = self.get_ves_ip_and_port()
+ registration_number: int = 0
+ registered_successfully: bool = False
+ while registration_number < settings.PNF_REGISTRATION_NUMBER_OF_TRIES and not registered_successfully:
+ try:
+ response = requests.post(
+ "http://portal.api.simpledemo.onap.org:30999/simulator/start",
+ json={
+ "simulatorParams": {
+ "repeatCount": 9999,
+ "repeatInterval": 30,
+ "vesServerUrl": f"https://{ves_ip}:{ves_port}/eventListener/v7"
+ },
+ "templateName": "registration.json",
+ "patch": {
+ "event": {
+ "commonEventHeader": {
+ "sourceName": settings.SERVICE_INSTANCE_NAME
+ }
+ }
+ }
+ }
+ )
+ response.raise_for_status()
+ registered_successfully = True
+ self._logger.info(f"PNF registered with {settings.SERVICE_INSTANCE_NAME} source name")
+ except requests.HTTPError as http_error:
+ self._logger.debug(f"Can't connect with PNF simulator: {str(http_error)}")
+ registration_number = registration_number + 1
+ time.sleep(settings.PNF_WAIT_TIME)
+ if not registered_successfully:
+ raise OnapTestException("PNF not registered successfully")
--- /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"
+ heat_files_to_upload: onaptests/templates/artifacts/pNF.csar
+ 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"
Code Review / testsuite / pythonsdk-tests.git / commitdiff