Fix SDNC health check

- the latest SDNC version is not returning a response body
  anymore for the health check endpoint

Issue-ID: INT-2348
Change-Id: Ia11c82315f238221f892125ee03620f832df0f81
Signed-off-by: Fiete Ostkamp <fiete.ostkamp@telekom.de>

diff --git a/src/onaptests/steps/instantiate/sdnc_service.py b/src/onaptests/steps/instantiate/sdnc_service.py
index 747251f..58d20ea 100644 (file)
--- a/src/onaptests/steps/instantiate/sdnc_service.py
+++ b/src/onaptests/steps/instantiate/sdnc_service.py
@@ -1,3 +1,5 @@
+"""SDNC service instantiation and health check steps."""
+
 import base64
 import logging
 from typing import Dict
@@ -14,8 +16,10 @@ from onapsdk.utils.headers_creator import headers_sdnc_creator
 
 from onaptests.scenario.scenario_base import BaseScenarioStep
 from onaptests.steps.base import BaseStep
-from onaptests.utils.exceptions import (EnvironmentPreparationException,
-                                        OnapTestException)
+from onaptests.utils.exceptions import (
+    EnvironmentPreparationException,
+    OnapTestException,
+)
 
 
 class BaseSdncStep(BaseStep):
@@ -61,7 +65,6 @@ class CheckSdncDbStep(BaseSdncStep):
 
     def get_database_credentials(self):
         """Resolve SDNC datbase credentials from k8s secret."""
-
         if settings.IN_CLUSTER:
             config.load_incluster_config()
         else:
@@ -69,16 +72,21 @@ class CheckSdncDbStep(BaseSdncStep):
         api_instance = client.CoreV1Api()
         try:
             secret = api_instance.read_namespaced_secret(
-                settings.SDNC_SECRET_NAME, settings.K8S_TESTS_NAMESPACE)
+                settings.SDNC_SECRET_NAME, settings.K8S_TESTS_NAMESPACE
+            )
             if secret.data:
-                if (self.SDNC_DB_LOGIN in secret.data and self.SDNC_DB_PASSWORD in secret.data):
+                if (
+                    self.SDNC_DB_LOGIN in secret.data
+                    and self.SDNC_DB_PASSWORD in secret.data
+                ):
                     login_base64 = secret.data[self.SDNC_DB_LOGIN]
                     self.login = base64.b64decode(login_base64).decode("utf-8")
                     password_base64 = secret.data[self.SDNC_DB_PASSWORD]
                     self.password = base64.b64decode(password_base64).decode("utf-8")
                 else:
                     raise EnvironmentPreparationException(
-                        "Login key or password key not found in secret")
+                        "Login key or password key not found in secret"
+                    )
             else:
                 raise EnvironmentPreparationException("Secret data not found in secret")
         except client.rest.ApiException as e:
@@ -105,11 +113,11 @@ class CheckSdncDbStep(BaseSdncStep):
                 host=settings.SDNC_DB_PRIMARY_HOST,
                 port=settings.SDNC_DB_PORT,
                 user=self.login,
-                password=self.password)
+                password=self.password,
+            )
             self._check_query(conn, self.SDNC_QUERY_LOGIC)
             self._check_query(conn, self.SDNC_QUERY_MODEL)
-        except (mysql.errors.ProgrammingError,
-                mysql.errors.DatabaseError) as e:
+        except (mysql.errors.ProgrammingError, mysql.errors.DatabaseError) as e:
             raise OnapTestException(e) from e
         except Exception as e:
             raise OnapTestException("Cannot connect to SDNC Database") from e
@@ -143,7 +151,7 @@ class ServiceCreateStep(BaseSdncStep):
             self.service = Service(
                 service_instance_id=settings.SERVICE_ID,
                 service_status=settings.SERVICE_STATUS,
-                service_data=settings.SERVICE_DATA
+                service_data=settings.SERVICE_DATA,
             )
             self.service.create()
             self._logger.info("SDNC service is created.")
@@ -191,8 +199,11 @@ class UpdateSdncService(BaseSdncStep):
 
     @BaseSdncStep.store_state
     def execute(self) -> None:
+        """Execute SDNC service update step."""
         super().execute()
-        self._logger.info("Get existing SDNC service instance and update it over GR-API")
+        self._logger.info(
+            "Get existing SDNC service instance and update it over GR-API"
+        )
         try:
             service = Service.get(settings.SERVICE_ID)
             service.service_status = settings.SERVICE_CHANGED_STATUS
@@ -227,15 +238,13 @@ class UploadVfModulePreloadStep(BaseSdncStep):
 
     @BaseSdncStep.store_state
     def execute(self) -> None:
+        """Execute VF module preload upload step."""
         super().execute()
         self._logger.info("Upload VfModule preload information over GR-API")
         VfModulePreload.upload_vf_module_preload(
-            {
-                "vnf_name": settings.VNF_NAME,
-                "vnf_type": settings.VNF_TYPE
-            },
+            {"vnf_name": settings.VNF_NAME, "vnf_type": settings.VNF_TYPE},
             settings.VF_MODULE_NAME,
-            None
+            None,
         )
 
 
@@ -262,17 +271,19 @@ class CheckSdncHealthStep(BaseSdncStep, SdncElement):
 
     @BaseSdncStep.store_state
     def execute(self) -> None:
+        """Execute SDNC health check step."""
         super().execute()
-        result = self.send_message_json(
+
+        response = self.send_message(
             "POST",
             "SDNC SLI API Healthcheck",
-            f"{self.base_url}/rests/operations/SLI-API:healthcheck")
-        message = ""
-        if result and result["SLI-API:output"]:
-            if result["SLI-API:output"]["response-code"] == "200":
-                return
-            message = result["SLI-API:output"]["response-message"]
-        raise OnapTestException("SDNC is not healthy. %s" % message)
+            f"{self.base_url}/rests/operations/SLI-API:healthcheck",
+        )
+        if response and response.status_code == 204:
+            return
+        raise OnapTestException(
+            "SDNC is not healthy. Status code %s" % response.status_code
+        )
 
 
 class GetSdncPreloadStep(BaseSdncStep):
@@ -306,8 +317,11 @@ class GetSdncPreloadStep(BaseSdncStep):
 
     @BaseSdncStep.store_state
     def execute(self) -> None:
+        """Execute SDNC preload information retrieval step."""
         super().execute()
-        self._logger.info("Get existing SDNC service instance and update it over GR-API")
+        self._logger.info(
+            "Get existing SDNC service instance and update it over GR-API"
+        )
         preloads = PreloadInformation.get_all()
         for preload_information in preloads:
             self.__logger.debug(preload_information)