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")
+K8S_CONFIG = (sys.path[-1] +
+ "/onaptests/templates/artifacts/config")
VENDOR_NAME = "basicnf_vendor"
from pathlib import Path
+from uuid import uuid4
from .settings import * # pylint: disable=W0614
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")
+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_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"
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
- )
-)
+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 = Path(Path(__file__).parent.parent, "templates/artifacts/config")
+
+SERVICE_INSTANCE_NAME = "TestPNFMacroInstantiation"
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.simulator.pnf_simulator_cnf.pnf_register import PnfSimulatorCnfRegisterStep
from onaptests.steps.instantiate.service_macro import YamlTemplateServiceMacroInstantiateStep
"""
super().__init__(cleanup=cleanup)
self._yaml_template: dict = None
- self.add_step(PNFRegisterStep(
+ self.add_step(PnfSimulatorCnfRegisterStep(
cleanup=settings.CLEANUP_FLAG
))
self.add_step(CbaEnrichStep(
--- /dev/null
+"""A&AI cloud region creation module."""
+from onapsdk.aai.cloud_infrastructure import CloudRegion
+from onapsdk.configuration import settings
+
+from ..base import BaseStep
+
+
+class CloudRegionCreateStep(BaseStep):
+ """Cloud region creation step."""
+
+ @property
+ def description(self) -> str:
+ """Step description."""
+ return "Create cloud region."
+
+ @property
+ def component(self) -> str:
+ """Component name."""
+ return "AAI"
+
+ @BaseStep.store_state
+ def execute(self):
+ """Create cloud region.
+
+ Use settings values:
+ - CLOUD_REGION_CLOUD_OWNER,
+ - CLOUD_REGION_ID,
+ - CLOUD_REGION_TYPE,
+ - CLOUD_REGION_VERSION,
+ - CLOUD_OWNER_DEFINED_TYPE,
+ - COMPLEX_PHYSICAL_LOCATION_ID.
+
+ """
+ super().execute()
+ self._logger.info("*Check if cloud region exists *")
+ try:
+ CloudRegion.get_by_id(
+ cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER,
+ cloud_region_id=settings.CLOUD_REGION_ID,
+ )
+ except ValueError:
+ CloudRegion.create(
+ cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER,
+ cloud_region_id=settings.CLOUD_REGION_ID,
+ 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
+ )
+"""Connectivity info creation module."""
from onapsdk.configuration import settings
from onapsdk.msb.k8s import ConnectivityInfo
@BaseStep.store_state
def execute(self):
- """Creation k8s connectivity information
+ """Creation k8s connectivity information.
Use settings values:
- CLOUD_REGION_ID,
- CLOUD_REGION_CLOUD_OWNER,
- - K8S_KUBECONFIG_FILE.
+ - K8S_CONFIG.
"""
super().execute()
######## Create Connectivity Info #########################################
self._logger.info("Create the k8s connectivity information")
ConnectivityInfo.create(settings.CLOUD_REGION_ID,
settings.CLOUD_REGION_CLOUD_OWNER,
- open(settings.K8S_KUBECONFIG_FILE, 'rb').read())
+ open(settings.K8S_CONFIG, 'rb').read())
def cleanup(self) -> None:
- """Cleanup K8S Connectivity information.
- """
+ """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)
+"""A&AI cloud region registstation module."""
import time
from uuid import uuid4
from onapsdk.configuration import settings
from ..base import BaseStep
+from onaptests.steps.cloud.cloud_region_create import CloudRegionCreateStep
class RegisterCloudRegionStep(BaseStep):
"""Cloud region registration step."""
+ def __init__(self, cleanup: bool) -> None:
+ """Initialize step.
+
+ Substeps:
+ - CloudRegionCreateStep.
+ """
+ super().__init__(cleanup=cleanup)
+ self.add_step(CloudRegionCreateStep(cleanup=cleanup))
+
@property
def description(self) -> str:
"""Step description."""
@BaseStep.store_state
def execute(self):
- """Register cloud region
+ """Register cloud region.
Use settings values:
- 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,
- TENANT_NAME.
"""
super().execute()
- self._logger.info("*Check if cloud region exists *")
- try:
- cloud_region: CloudRegion = CloudRegion.get_by_id(
- cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER,
- cloud_region_id=settings.CLOUD_REGION_ID,
- )
- except ValueError:
- self._logger.info("*Create the cloud region *")
- cloud_region: CloudRegion = CloudRegion.create(
- cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER,
- cloud_region_id=settings.CLOUD_REGION_ID,
- 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_region: CloudRegion = CloudRegion.get_by_id(
+ cloud_owner=settings.CLOUD_REGION_CLOUD_OWNER,
+ cloud_region_id=settings.CLOUD_REGION_ID,
+ )
+ if not list(cloud_region.esr_system_infos):
cloud_region.add_esr_system_info(
esr_system_info_id=str(uuid4()),
user_name=settings.VIM_USERNAME,
break
nb_try += 1
- # Retrieve the tenant, created by multicloud registration
- # if it does not exist, create it
- try:
- cloud_region.get_tenant(settings.TENANT_ID)
- except ValueError:
- 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)
+ # Retrieve the tenant, created by multicloud registration
+ # if it does not exist, create it
+ try:
+ cloud_region.get_tenant(settings.TENANT_ID)
+ except ValueError:
+ 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)
- # be sure that an availability zone has been created
- # if not, create it
- try:
- cloud_region.get_availability_zone_by_name(
- settings.AVAILABILITY_ZONE_NAME)
- except ValueError:
- cloud_region.add_availability_zone(
- settings.AVAILABILITY_ZONE_NAME,
- settings.AVAILABILITY_ZONE_TYPE)
+ # be sure that an availability zone has been created
+ # if not, create it
+ try:
+ cloud_region.get_availability_zone_by_name(
+ settings.AVAILABILITY_ZONE_NAME)
+ except ValueError:
+ cloud_region.add_availability_zone(
+ settings.AVAILABILITY_ZONE_NAME,
+ settings.AVAILABILITY_ZONE_TYPE)
--- /dev/null
+"""MSB k8s instantiation module."""
+from onapsdk.configuration import settings
+from onapsdk.msb.k8s import Instance
+
+from onaptests.steps.base import BaseStep
+from onaptests.steps.onboard.msb_k8s import CreateProfileStep
+
+
+class CreateInstanceStep(BaseStep):
+ """Create MSB k8s instance step."""
+
+ def __init__(self, cleanup: bool = False) -> None:
+ """Initialize step.
+
+ Substeps:
+ - CreateProfileStep.
+ """
+ super().__init__(cleanup=cleanup)
+ self.add_step(CreateProfileStep(cleanup=cleanup))
+ self.instance: Instance = None
+
+ @property
+ def description(self) -> str:
+ """Step description."""
+ return "Create K8S instance."
+
+ @property
+ def component(self) -> str:
+ """Component name."""
+ return "K8S plugin"
+
+ @BaseStep.store_state
+ def execute(self) -> None:
+ """Create instance using MSB K8S plugin."""
+ super().execute()
+ self.instance = Instance.create(cloud_region_id=settings.CLOUD_REGION_ID,
+ profile_name=settings.PNF_PROFILE_NAME,
+ rb_name=settings.PNF_RB_NAME,
+ rb_version=settings.PNF_RB_VERSION)
+
+ def cleanup(self) -> None:
+ """Delete instance."""
+ if self.instance:
+ self.instance.delete()
+ return super().cleanup()
"""Bootstrap blueprintsprocessor."""
def __init__(self, cleanup: bool = False) -> None:
+ """Initialize step.
+
+ Substeps:
+ - ExposeCDSBlueprintprocessorNodePortStep.
+ """
super().__init__(cleanup=cleanup)
self.add_step(ExposeCDSBlueprintprocessorNodePortStep(cleanup=cleanup))
@BaseStep.store_state
def execute(self) -> None:
- """Bootsrap CDS blueprintprocessor"""
+ """Bootsrap CDS blueprintprocessor."""
super().execute()
Blueprintprocessor.bootstrap()
"""
super().cleanup()
- Path(settings.CDA_CBA_ENRICHED).unlink()
+ Path(settings.CDS_CBA_ENRICHED).unlink()
--- /dev/null
+"""MSB k8s plugin module."""
+from onapsdk.configuration import settings
+from onapsdk.msb.k8s import Definition, Profile
+
+from onaptests.steps.base import BaseStep
+from onaptests.steps.cloud.cloud_region_create import CloudRegionCreateStep
+from onaptests.steps.cloud.k8s_connectivity_info_create import K8SConnectivityInfoStep
+
+
+class CreateDefinitionStep(BaseStep):
+ """Create definition step initialization."""
+
+ def __init__(self, cleanup: bool = False) -> None:
+ """Initialize step.
+
+ Substeps:
+ - CloudRegionCreateStep,
+ - K8SConnectivityInfoStep.
+ """
+ super().__init__(cleanup=cleanup)
+ self.add_step(CloudRegionCreateStep(cleanup=cleanup))
+ self.add_step(K8SConnectivityInfoStep(cleanup=cleanup))
+ self.definition: Definition = None
+
+ @property
+ def description(self) -> str:
+ """Step description."""
+ return "Create K8S definition."
+
+ @property
+ def component(self) -> str:
+ """Component name."""
+ return "K8S plugin"
+
+ @BaseStep.store_state
+ def execute(self) -> None:
+ """Create definition."""
+ super().execute()
+ with open(settings.PNF_DEFINITION_ATRIFACT_FILE_PATH, "rb") as definition_file:
+ self.definition = Definition.create(rb_name=settings.PNF_RB_NAME,
+ rb_version=settings.PNF_RB_VERSION)
+ self.definition.upload_artifact(definition_file.read())
+
+
+class CreateProfileStep(BaseStep):
+ """Create profile step."""
+
+ def __init__(self, cleanup: bool = False) -> None:
+ """Initialize step.
+
+ Substeps:
+ - CreateDefinitionStep.
+ """
+ super().__init__(cleanup=cleanup)
+ self.add_step(CreateDefinitionStep(cleanup=cleanup))
+ self.profile: Profile = None
+
+ @property
+ def description(self) -> str:
+ """Step description."""
+ return "Create K8S profile."
+
+ @property
+ def component(self) -> str:
+ """Component name."""
+ return "K8S plugin"
+
+ @BaseStep.store_state
+ def execute(self) -> None:
+ """Create profile."""
+ super().execute()
+ definition: Definition = Definition.get_definition_by_name_version(\
+ rb_name=settings.PNF_RB_NAME,
+ rb_version=settings.PNF_RB_VERSION)
+ with open(settings.PNF_PROFILE_ARTIFACT_FILE_PATH, "rb") as profile_file:
+ self.profile = definition.create_profile(profile_name=settings.PNF_PROFILE_NAME,
+ namespace=settings.K8S_NAMESPACE,
+ kubernetes_version=settings.K8S_VERSION)
+ self.profile.upload_artifact(profile_file.read())
--- /dev/null
+"""PNF simulator CNF package."""
--- /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
+
+
+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 == "xdcae-ves-collector":
+ 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(5.0) # Let's still wait for PNF simulator to make sure it's initialized
+ ves_ip, ves_port = self.get_ves_ip_and_port()
+ response = requests.post(
+ "http://portal.api.simpledemo.onap.org:30999/simulator/event",
+ json={
+ "vesServerUrl": f"https://{ves_ip}:{ves_port}/eventListener/v7",
+ "event": {
+ "event": {
+ "commonEventHeader": {
+ "domain": "pnfRegistration",
+ "eventId": "ORAN_SIM_400600927_2020-04-02T17:20:22.2Z",
+ "eventName": "pnfRegistration",
+ "eventType": "EventType5G",
+ "sequence": 0,
+ "priority": "Low",
+ "reportingEntityId": "",
+ "reportingEntityName": "ORAN_SIM_400600927",
+ "sourceId": "",
+ "sourceName": settings.SERVICE_INSTANCE_NAME,
+ "startEpochMicrosec": 94262132085746,
+ "lastEpochMicrosec": 94262132085746,
+ "nfNamingCode": "sdn controller",
+ "nfVendorName": "sdn",
+ "timeZoneOffset": "+00:00",
+ "version": "4.0.1",
+ "vesEventListenerVersion": "7.0.1"
+ },
+ "pnfRegistrationFields": {
+ "pnfRegistrationFieldsVersion": "2.0",
+ "lastServiceDate": "2019-08-16",
+ "macAddress": "D7:64:C8:CC:E9:32",
+ "manufactureDate": "2019-08-16",
+ "modelNumber": "Simulated Device",
+ "oamV4IpAddress": "172.30.1.6",
+ "oamV6IpAddress": "0:0:0:0:0:ffff:a0a:011",
+ "serialNumber": "Simulated Device",
+ "softwareVersion": "2.3.5",
+ "unitFamily": "Simulated Device",
+ "unitType": "ntsim_oran",
+ "vendorName": "Melacon",
+ "additionalFields": {
+ "oamPort": "830",
+ "protocol": "SSH",
+ "username": "netconf",
+ "password": "netconf",
+ "reconnectOnChangedSchema": "false",
+ "sleep-factor": "1.5",
+ "tcpOnly": "false",
+ "connectionTimeout": "20000",
+ "maxConnectionAttempts": "100",
+ "betweenAttemptsTimeout": "2000",
+ "keepaliveDelay": "120"
+ }
+ }
+ }
+ }
+ }
+ )
+ response.raise_for_status()
class ProfileCleanupException(OnapTestException):
"""K8s profile cannot be cleaned."""
error_message="Profile can't be cleaned"
+
+class EnvironmentPreparationException(OnapTestException):
+ """Test environment preparation exception."""
+ error_message="Test can't be run properly due to preparation error"
Code Review / testsuite / pythonsdk-tests.git / commitdiff