}
}
}
+CDS_NODE_PORT = 30449
# Disable YAML SDC model definition which means all SDC config reside in SERVICE_YAML_TEMPLATE
MODEL_YAML_TEMPLATE = None
+USE_SIMULATOR = True
+VES_NODE_PORT = 30417
\ No newline at end of file
SERVICE_DISTRIBUTION_SLEEP_TIME = 60
EXPOSE_SERVICES_NODE_PORTS = True
IN_CLUSTER = False
+VES_BASIC_AUTH = {'username': 'sample1', 'password': 'sample1'}
\ No newline at end of file
"""Instantiate service with PNF using SO macro flow."""
from onapsdk.configuration import settings
+from yaml import SafeLoader, load
+
from onaptests.scenario.scenario_base import BaseStep, ScenarioBase, YamlTemplateBaseScenarioStep
+from onaptests.steps.instantiate.pnf_register_ves import \
+ SendPnfRegisterVesEvent
from onaptests.steps.instantiate.service_macro import \
YamlTemplateServiceMacroInstantiateStep
from onaptests.steps.onboard.cds import CbaEnrichStep
from onaptests.steps.simulator.pnf_simulator_cnf.pnf_register import \
PnfSimulatorCnfRegisterStep
-from yaml import SafeLoader, load
class PnfMacroScenarioStep(YamlTemplateBaseScenarioStep):
"""
super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
self._yaml_template: dict = None
- self.add_step(PnfSimulatorCnfRegisterStep())
+ if settings.USE_SIMULATOR:
+ self.add_step(PnfSimulatorCnfRegisterStep())
+ else:
+ self.add_step(SendPnfRegisterVesEvent())
self.add_step(CbaEnrichStep())
self.add_step(YamlTemplateServiceMacroInstantiateStep())
--- /dev/null
+# http://www.apache.org/licenses/LICENSE-2.0
+"""Expose service NodePort module."""
+
+from typing import Any, Dict
+
+import urllib3
+from kubernetes import client, config
+from kubernetes.client.exceptions import ApiException
+from onapsdk.configuration import settings
+
+from onaptests.steps.base import BaseStep
+from onaptests.utils.exceptions import OnapTestException
+
+
+class ExposeServiceNodePortStep(BaseStep):
+ """Expose Service NodePort."""
+
+ def __init__(self, component: str, service_name: str, port: int, node_port: int) -> None:
+ """Initialize step."""
+ super().__init__(cleanup=settings.CLEANUP_FLAG)
+ self.component_value = component
+ self.service_name = service_name
+ self.port = port
+ self.node_port = node_port
+ if settings.IN_CLUSTER:
+ config.load_incluster_config()
+ else:
+ config.load_kube_config(config_file=settings.K8S_CONFIG)
+ self.k8s_client: client.CoreV1Api = client.CoreV1Api()
+
+ @property
+ def component(self) -> str:
+ return self.component_value
+
+ @property
+ def description(self) -> str:
+ """Step description."""
+ return "Expose service NodePort."
+
+ def is_service_node_port_type(self) -> bool:
+ """Check if service type is 'NodePort'
+
+ Raises:
+ OnapTestException: Kubernetes API error
+
+ Returns:
+ bool: True if service type is 'NodePort', False otherwise
+
+ """
+ try:
+ service_data: Dict[str, Any] = self.k8s_client.read_namespaced_service(
+ self.service_name,
+ settings.K8S_ONAP_NAMESPACE
+ )
+ return service_data.spec.type == "NodePort"
+ except ApiException:
+ self._logger.exception("Kubernetes API exception")
+ raise OnapTestException
+
+ @BaseStep.store_state
+ def execute(self) -> None:
+ """Expose services ports using kubernetes client.
+
+ Use settings values:
+ - K8S_CONFIG,
+ - K8S_ONAP_NAMESPACE.
+ - EXPOSE_SERVICES_NODE_PORTS
+
+ """
+ super().execute()
+ if not self.is_service_node_port_type():
+ try:
+ self.k8s_client.patch_namespaced_service(
+ self.service_name,
+ settings.K8S_ONAP_NAMESPACE,
+ {"spec": {"ports": [{"port": self.port, "nodePort": self.node_port}], "type": "NodePort"}}
+ )
+ except ApiException:
+ self._logger.exception("Kubernetes API exception")
+ raise OnapTestException
+ except urllib3.exceptions.HTTPError:
+ self._logger.exception("Can't connect with k8s")
+ raise OnapTestException
+ else:
+ self._logger.debug("Service already patched, skip")
+
+ def cleanup(self) -> None:
+ """Step cleanup.
+
+ Restore service.
+
+ """
+ if self.is_service_node_port_type():
+ try:
+ self.k8s_client.patch_namespaced_service(
+ self.service_name,
+ settings.K8S_ONAP_NAMESPACE,
+ [
+ {
+ "op": "remove",
+ "path": "/spec/ports/0/nodePort"
+ },
+ {
+ "op": "replace",
+ "path": "/spec/type",
+ "value": "ClusterIP"
+ }
+ ]
+ )
+ except ApiException:
+ self._logger.exception("Kubernetes API exception")
+ raise OnapTestException
+ except urllib3.exceptions.HTTPError:
+ self._logger.exception("Can't connect with k8s")
+ raise OnapTestException
+ else:
+ self._logger.debug("Service is not 'NodePort' type, skip")
+ return super().cleanup()
--- /dev/null
+# http://www.apache.org/licenses/LICENSE-2.0
+"""PNF simulator registration module."""
+
+import time
+
+import requests
+from jinja2 import Environment, PackageLoader, select_autoescape
+from onapsdk.configuration import settings
+from onapsdk.ves.ves import Ves
+
+from onaptests.steps.base import BaseStep
+from onaptests.steps.cloud.expose_service_node_port import \
+ ExposeServiceNodePortStep
+from onaptests.utils.exceptions import OnapTestException
+
+
+class SendPnfRegisterVesEvent(BaseStep):
+ """PNF VES registration step."""
+
+ def __init__(self) -> None:
+ """Initialize step."""
+ super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
+ if settings.EXPOSE_SERVICES_NODE_PORTS:
+ self.add_step(ExposeServiceNodePortStep(component="VES-Collector",
+ service_name="dcae-ves-collector", port=8080, node_port=settings.VES_NODE_PORT))
+
+ @property
+ def description(self) -> str:
+ """Step description."""
+ return "Register PNF with VES."
+
+ @property
+ def component(self) -> str:
+ """Component name."""
+ return "Environment"
+
+ @BaseStep.store_state
+ def execute(self) -> None:
+ """Send PNF registration event."""
+ super().execute()
+ registration_number: int = 0
+
+ source_name = settings.SERVICE_INSTANCE_NAME
+ jinja_env = Environment(autoescape=select_autoescape(['jinja']),
+ loader=PackageLoader('onaptests.templates',
+ 'artifacts'))
+ template = jinja_env.get_template("pnf_register_ves_message.jinja")
+ event_data = template.render(
+ source_name=source_name)
+
+ registered_successfully: bool = False
+ while (registration_number < settings.PNF_REGISTRATION_NUMBER_OF_TRIES and
+ not registered_successfully):
+ try:
+ response = Ves.send_event(version="v7", json_event=event_data,
+ basic_auth=settings.VES_BASIC_AUTH)
+ if response is None:
+ raise OnapTestException("Failed to send event to VES SERVER")
+ response.raise_for_status()
+ registered_successfully = True
+ self._logger.info(f"PNF registered with {settings.SERVICE_INSTANCE_NAME} "
+ "source name")
+ except (requests.ConnectionError, requests.HTTPError) as http_error:
+ self._logger.debug(f"Can't send to ves: {str(http_error)}")
+ registration_number += 1
+ time.sleep(settings.PNF_WAIT_TIME)
+ if not registered_successfully:
+ raise OnapTestException("PNF not registered successfully")
from pathlib import Path
from typing import Any, Dict
-from kubernetes import client, config
-from kubernetes.client.exceptions import ApiException
from onapsdk.cds import Blueprint, DataDictionarySet
from onapsdk.cds.blueprint import Workflow
from onapsdk.cds.blueprint_processor import Blueprintprocessor
from onapsdk.configuration import settings
-import urllib3
from onaptests.steps.base import BaseStep
+from onaptests.steps.cloud.expose_service_node_port import \
+ ExposeServiceNodePortStep
from onaptests.utils.exceptions import OnapTestException
return "CDS"
-class ExposeCDSBlueprintprocessorNodePortStep(CDSBaseStep):
+class ExposeCDSBlueprintprocessorNodePortStep(CDSBaseStep, ExposeServiceNodePortStep):
"""Expose CDS blueprintsprocessor port."""
-
def __init__(self) -> None:
"""Initialize step."""
- super().__init__(cleanup=settings.CLEANUP_FLAG)
- self.service_name: str = "cds-blueprints-processor-http"
- if settings.IN_CLUSTER:
- config.load_incluster_config()
- else:
- config.load_kube_config(config_file=settings.K8S_CONFIG)
- self.k8s_client: client.CoreV1Api = client.CoreV1Api()
-
- @property
- def description(self) -> str:
- """Step description."""
- return "Expose CDS blueprintsprocessor NodePort."
-
- def is_service_node_port_type(self) -> bool:
- """Check if CDS blueprints processor service type is 'NodePort'
-
- Raises:
- OnapTestException: Kubernetes API error
-
- Returns:
- bool: True if service type is 'NodePort', False otherwise
-
- """
- try:
- service_data: Dict[str, Any] = self.k8s_client.read_namespaced_service(
- self.service_name,
- settings.K8S_ONAP_NAMESPACE
- )
- return service_data.spec.type == "NodePort"
- except ApiException:
- self._logger.exception("Kubernetes API exception")
- raise OnapTestException
-
- @BaseStep.store_state
- def execute(self) -> None:
- """Expose CDS blueprintprocessor port using kubernetes client.
-
- Use settings values:
- - K8S_CONFIG,
- - K8S_ONAP_NAMESPACE.
- - EXPOSE_SERVICES_NODE_PORTS
-
- """
- super().execute()
- if not self.is_service_node_port_type():
- try:
- self.k8s_client.patch_namespaced_service(
- self.service_name,
- settings.K8S_ONAP_NAMESPACE,
- {"spec": {"ports": [{"port": 8080, "nodePort": 30449}], "type": "NodePort"}}
- )
- except ApiException:
- self._logger.exception("Kubernetes API exception")
- raise OnapTestException
- except urllib3.exceptions.HTTPError:
- self._logger.exception("Can't connect with k8s")
- raise OnapTestException
- else:
- self._logger.debug("Service already patched, skip")
-
- @BaseStep.store_state(cleanup=True)
- def cleanup(self) -> None:
- """Step cleanup.
-
- Restore CDS blueprintprocessor service.
-
- """
- if self.is_service_node_port_type():
- try:
- self.k8s_client.patch_namespaced_service(
- self.service_name,
- settings.K8S_ONAP_NAMESPACE,
- [
- {
- "op": "remove",
- "path": "/spec/ports/0/nodePort"
- },
- {
- "op": "replace",
- "path": "/spec/type",
- "value": "ClusterIP"
- }
- ]
- )
- except ApiException:
- self._logger.exception("Kubernetes API exception")
- raise OnapTestException
- except urllib3.exceptions.HTTPError:
- self._logger.exception("Can't connect with k8s")
- raise OnapTestException
- else:
- self._logger.debug("Service is not 'NodePort' type, skip")
- return super().cleanup()
+ super().__init__(component = "CDS", service_name="cds-blueprints-processor-http", port=8080, node_port=settings.CDS_NODE_PORT)
class BootstrapBlueprintprocessor(CDSBaseStep):
--- /dev/null
+{
+ "event": {
+ "commonEventHeader": {
+ "eventId": "registration_39239592",
+ "eventType": "pnfRegistration",
+ "reportingEntityName": "NOK6061ZW3",
+ "domain": "pnfRegistration",
+ "nfcNamingCode": "oam",
+ "sequence": 0,
+ "sourceId": "val13",
+ "internalHeaderFields": {},
+ "priority": "Normal",
+ "sourceName": "{{ source_name }}",
+ "eventName": "pnfRegistration_Nokia_5gDu",
+ "version": "4.0.1",
+ "nfNamingCode": "gNB",
+ "startEpochMicrosec": 1539239592379,
+ "vesEventListenerVersion": "7.0.1",
+ "lastEpochMicrosec": 1539239592379
+ },
+ "pnfRegistrationFields": {
+ "pnfRegistrationFieldsVersion": "2.0",
+ "serialNumber": "6061ZW3",
+ "vendorName": "Nokia",
+ "oamV4IpAddress": "192.168.0.1",
+ "oamV6IpAddress": "2001:db8::1428:57ab",
+ "unitFamily": "BBU",
+ "modelNumber": "val6",
+ "softwareVersion": "val7",
+ "unitType": "val8"
+ }
+ }
+ }
\ No newline at end of file
Code Review / testsuite / pythonsdk-tests.git / commitdiff