Wait for pnf simulator

Add VSP for PNF

Issue-ID: TEST-280
Signed-off-by: Michal Jagiello <michal.jagiello@t-mobile.pl>
Change-Id: If2476f5f9f9fc8d4c14c5dd398a8a659e7fe3e9a

1 files changed, 118 insertions, 0 deletions

diff --git a/src/onaptests/steps/simulator/pnf_simulator_cnf/pnf_register.py b/src/onaptests/steps/simulator/pnf_simulator_cnf/pnf_register.py
new file mode 100644
index 0000000..c1d76b0
--- /dev/null
+++ b/src/onaptests/steps/simulator/pnf_simulator_cnf/pnf_register.py
@@ -0,0 +1,118 @@
+# 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")