Add Onap Jakarta support

[it/dep.git] / smo-install / test / pythonsdk / src / orantests / conftest.py

diff --git a/smo-install/test/pythonsdk/src/orantests/conftest.py b/smo-install/test/pythonsdk/src/orantests/conftest.py
index 2e78b5b..dd4e469 100644 (file)
--- a/smo-install/test/pythonsdk/src/orantests/conftest.py
+++ b/smo-install/test/pythonsdk/src/orantests/conftest.py
@@ -25,10 +25,15 @@
 import logging
 import logging.config
 import os
-from oransdk.dmaap.dmaap import OranDmaap
-from oransdk.sdnc.sdnc import OranSdnc
+from requests import RequestException
 from onapsdk.configuration import settings
+from onapsdk.exceptions import ConnectionFailed, APIError
 from waiting import wait
+from urllib3.exceptions import NewConnectionError
+from oransdk.dmaap.dmaap import OranDmaap
+from oransdk.policy.clamp import ClampToscaTemplate
+from oransdk.policy.policy import OranPolicy
+from oransdk.sdnc.sdnc import OranSdnc
 from smo.smo import Smo
 from smo.network_simulators import NetworkSimulators
 
@@ -42,40 +47,51 @@ logger = logging.getLogger("Test Session setup")
 
 network_sims = NetworkSimulators("./resources")
 smo = Smo()
+clamp = ClampToscaTemplate(settings.CLAMP_BASICAUTH)
 dmaap = OranDmaap()
 sdnc = OranSdnc()
+policy = OranPolicy()
 
-TOPIC_PNFREG = '{"topicName": "unauthenticated.VES_PNFREG_OUTPUT"}'
+def policy_component_ready():
+    """Check if components are ready."""
+    logger.info("Verify policy components are ready")
+    try:
+        policy_ready = {"api_ready": False, "pap_ready": False, "apex_ready": False}
+    except (RequestException, NewConnectionError, ConnectionFailed, APIError) as e:
+        logger.error(e)
+        return False
+    policy_status = policy.get_components_status(settings.POLICY_BASICAUTH)
+    if (policy_status["api"]["healthy"] and not policy_ready["api_ready"]):
+        logger.info("Policy Api is ready")
+        policy_ready["api_ready"] = True
+    if (policy_status["pap"]["healthy"] and not policy_ready["pap_ready"]):
+        logger.info("Policy Pap is ready")
+        policy_ready["pap_ready"] = True
+    if (len(policy_status["pdps"]["apex"]) > 0 and policy_status["pdps"]["apex"][0]["healthy"] == "HEALTHY" and not policy_ready["apex_ready"]):
+        logger.info("Policy Apex is ready")
+        policy_ready["apex_ready"] = True
+    return policy_ready["api_ready"] and policy_ready["pap_ready"] and policy_ready["apex_ready"]
 
-TOPIC_FAULT = '{"topicName": "unauthenticated.SEC_FAULT_OUTPUT"}'
+def sdnc_component_ready():
+    """Check if SDNC component is ready."""
+    logger.info("Verify sdnc component is ready")
 
-#TOPIC_PNFREG = {
-#    "owner": "",
-#    "readerAcl": {
-#        "enabled": "true",
-#        "users": []
-#    },
-#    "name": "unauthenticated.VES_PNFREG_OUTPUT",
-#    "description": "",
-#    "writerAcl": {
-#        "enabled": "true",
-#        "users": []
-#    }
-#}
+    try:
+        response = OranSdnc.get_events(settings.SDNC_BASICAUTH, "test")
+    except (RequestException, NewConnectionError, ConnectionFailed, APIError) as e:
+        logger.error(e)
+        return False
+    return response.status_code == 200
 
-#TOPIC_FAULT = {
-#    "owner": "",
-#    "readerAcl": {
-#        "enabled": "true",
-#        "users": []
-#    },
-#    "name": "unauthenticated.SEC_FAULT_OUTPUT",
-#    "description": "",
-#    "writerAcl": {
-#        "enabled": "true",
-#        "users": []
-#    }
-#}
+def clamp_component_ready():
+    """Check if Clamp component is ready."""
+    logger.info("Verify Clamp component is ready")
+    try:
+        response = clamp.get_template_instance()
+    except (RequestException, NewConnectionError, ConnectionFailed, APIError) as e:
+        logger.error(e)
+        return False
+    return response["automationCompositionList"] is not None
 
 ###### Entry points of PYTEST Session
 def pytest_sessionstart():
@@ -83,15 +99,12 @@ def pytest_sessionstart():
     logger.info("Check and wait for SMO to be running")
     smo.wait_for_smo_to_be_running()
     logger.info("Check and for for SDNC to be running")
-    wait(lambda: OranSdnc.get_events(settings.SDNC_BASICAUTH, "test").status_code == 200, sleep_seconds=10, timeout_seconds=300, waiting_for="SDNC to be ready")
+    wait(lambda: policy_component_ready(), sleep_seconds=settings.POLICY_CHECK_RETRY, timeout_seconds=settings.POLICY_CHECK_TIMEOUT, waiting_for="Policy to be ready")
+    wait(lambda: sdnc_component_ready(), sleep_seconds=settings.SDNC_CHECK_RETRY, timeout_seconds=settings.SDNC_CHECK_TIMEOUT, waiting_for="SDNC to be ready")
+    # disable for now, until policy/clamp issue has been fixed
+    ##wait(lambda: clamp_component_ready(), sleep_seconds=settings.CLAMP_CHECK_RETRY, timeout_seconds=settings.CLAMP_CHECK_TIMEOUT, waiting_for="Clamp to be ready")
 
-    dmaap.create_topic(TOPIC_PNFREG)
-    dmaap.create_topic(TOPIC_FAULT)
-    ### Due to an Onap Ves/dmaap behavior !!! DU sims must send messages
-    ### twice so we need to create/delete the sims
-#    network_sims.start_network_simulators()
-#    network_sims.wait_for_network_simulators_to_be_running()
-#    time.sleep(20)
-    ## Now kill the simulators and restart them for the test session
+    ## Just kill any simulators that could already be runnin
     network_sims.stop_network_simulators()
     ###### END of FIRST start, now we can start the sims for the real tests.
+    logger.info("Tests session setup is ready")