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 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
 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()
 logging.config.dictConfig(settings.LOG_CONFIG)
 logger = logging.getLogger("test APEX policy")
 dmaap = OranDmaap()

+dmaap_utils = DmaapUtils()

 policy = OranPolicy()
 network_simulators = NetworkSimulators("./resources")
 
 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")
 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
 
     # 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")
     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)
     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():
     logger.info("Test Session cleanup done")
 
 def create_policy():