Add Onap Jakarta support

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

diff --git a/smo-install/test/pythonsdk/src/orantests/test_apex_policy.py b/smo-install/test/pythonsdk/src/orantests/test_apex_policy.py
index 7210dfd..010ae80 100644 (file)
--- a/smo-install/test/pythonsdk/src/orantests/test_apex_policy.py
+++ b/smo-install/test/pythonsdk/src/orantests/test_apex_policy.py
@@ -33,6 +33,7 @@ from oransdk.dmaap.dmaap import OranDmaap
 from oransdk.policy.policy import OranPolicy, PolicyType
 from oransdk.sdnc.sdnc import OranSdnc
 from oransdk.utils.jinja import jinja_env
+from smo.dmaap import DmaapUtils
 from smo.network_simulators import NetworkSimulators
 
 # Set working dir as python script location
@@ -44,6 +45,7 @@ os.chdir(dname)
 logging.config.dictConfig(settings.LOG_CONFIG)
 logger = logging.getLogger("test APEX policy")
 dmaap = OranDmaap()
+dmaap_utils = DmaapUtils()
 policy = OranPolicy()
 network_simulators = NetworkSimulators("./resources")
 
@@ -61,17 +63,11 @@ deployment_port = "12345"
 def setup_simulators():
     """Setup the simulators before the executing the tests."""
     logger.info("Test class setup for Apex tests")
-
-    dmaap.create_topic(settings.DMAAP_TOPIC_FAULT_JSON)
-    dmaap.create_topic(settings.DMAAP_TOPIC_PNFREG_JSON)
-    # Purge the FAULT TOPIC
-    wait(lambda: (dmaap.get_message_from_topic(settings.DMAAP_TOPIC_FAULT, 5000, settings.DMAAP_GROUP, settings.DMAAP_USER).json() == []), sleep_seconds=10, timeout_seconds=60, waiting_for="DMaap topic SEC_FAULT_OUTPUT to be empty")
-    wait(lambda: (dmaap.get_message_from_topic(settings.DMAAP_TOPIC_PNFREG, 5000, settings.DMAAP_GROUP, settings.DMAAP_USER).json() == []), sleep_seconds=10, timeout_seconds=60, waiting_for="DMaap topic unauthenticated.VES_PNFREG_OUTPUT to be empty")
-
-    network_simulators.start_network_simulators()
-    network_simulators.wait_for_network_simulators_to_be_running()
+    dmaap_utils.clean_dmaap(settings.DMAAP_GROUP, settings.DMAAP_USER)
+    network_simulators.start_and_wait_network_simulators()
 
     # Wait enough time to have at least the SDNR notifications sent
+
     logger.info("Waiting 10s that SDNR sends all registration events to VES...")
     time.sleep(10)
     logger.info("Test Session setup completed successfully")
@@ -81,6 +77,7 @@ def setup_simulators():
     network_simulators.stop_network_simulators()
     policy.undeploy_policy(policy_id, policy_version, settings.POLICY_BASICAUTH)
     policy.delete_policy(policy_type, policy_id, policy_version, settings.POLICY_BASICAUTH)
+    time.sleep(10)
     logger.info("Test Session cleanup done")
 
 def create_policy():