X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=smo-install%2Ftest%2Fpythonsdk%2Fsrc%2Forantests%2Fdisable_apex_policy.py;h=fa42df2b8486adbe4c7ac1370c6f0f078e55fd45;hb=a8414e7e44dea72da6f4f235e14f7fef59b4e266;hp=d5d261ade77e666ac4960f72198c8e289f80bf10;hpb=6c7e3c3c5e51182a891a2a21676f1cc2646ba484;p=it%2Fdep.git diff --git a/smo-install/test/pythonsdk/src/orantests/disable_apex_policy.py b/smo-install/test/pythonsdk/src/orantests/disable_apex_policy.py index d5d261ad..fa42df2b 100644 --- a/smo-install/test/pythonsdk/src/orantests/disable_apex_policy.py +++ b/smo-install/test/pythonsdk/src/orantests/disable_apex_policy.py @@ -26,60 +26,51 @@ import time import subprocess import logging import logging.config +import os +import pytest +from waiting import wait +from smo.network_simulators import NetworkSimulators from onapsdk.configuration import settings 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 +# Set working dir as python script location +abspath = os.path.abspath(__file__) +dname = os.path.dirname(abspath) +os.chdir(dname) + + logging.config.dictConfig(settings.LOG_CONFIG) logger = logging.getLogger("test APEX policy") dmaap = OranDmaap() policy = OranPolicy() +network_simulators = NetworkSimulators("./resources") -def create_new_topic(): - """Create new topic.""" - logger.info("Create new topic") - topic = '{ "topicName": "unauthenticated.SEC_FAULT_OUTPUT", "topicDescription": "test topic", "partitionCount": 1,"replicationCnCount": 1, "transactionEnabled": "false"}' - response = dmaap.create_topic(topic) - logger.info("response is: %s", response) - - logger.info("Verify topic created") - topiclist = dmaap.get_all_topics({}) - topic_created = False - for topic in topiclist: - if topic["topicName"] == "unauthenticated.SEC_FAULT_OUTPUT": - topic_created = True - break +TOPIC_FAULT = '{"topicName": "unauthenticated.SEC_FAULT_OUTPUT"}' - if topic_created: - logger.info("Topic created successfully") - else: - logger.info("Topic creation failed") - -def policy_component_ready(): - """Check if components are ready.""" - logger.info("Verify policy components are ready") - policy_ready = {"api_ready": False, "pap_ready": False, "apex_ready": False} - for x in range(60): - logger.info("Iteration %s", x) - 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 (policy_status["pdps"]["apex"][0]["healthy"] == "HEALTHY" and not policy_ready["apex_ready"]): - logger.info("Policy Apex is ready") - policy_ready["apex_ready"] = True - if (policy_ready["api_ready"] and policy_ready["pap_ready"] and policy_ready["apex_ready"]): - logger.info("Policy status all ready") - break +@pytest.fixture(scope="module", autouse=True) +def setup_simulators(): + """Setup the simulators before the executing the tests.""" + logger.info("Test class setup for Apex tests") + + dmaap.create_topic(TOPIC_FAULT) + # Purge the FAULT TOPIC + wait(lambda: (dmaap.get_message_from_topic("unauthenticated.SEC_FAULT_OUTPUT", 5000, settings.DMAAP_GROUP, settings.DMAAP_USER).json() == []), sleep_seconds=10, timeout_seconds=60, waiting_for="DMaap topic unauthenticated.SEC_FAULT_OUTPUT to be empty") + + network_simulators.start_network_simulators() + network_simulators.wait_for_network_simulators_to_be_running() + # 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") - if (not policy_ready["api_ready"] or not policy_ready["pap_ready"] or not policy_ready["apex_ready"]): - logger.info("Policy components are not ready. Exit the test.") + ### Cleanup code + yield + network_simulators.stop_network_simulators() + logger.info("Test Session cleanup done") def create_policy(): """Create the policy.""" @@ -88,11 +79,7 @@ def create_policy(): policy.create_policy(PolicyType(type="onap.policies.native.Apex", version="1.0.0"), policy_data, settings.POLICY_BASICAUTH) logger.info("Verify whether policy created successfully") - policy_response = policy.get_policy(PolicyType(type="onap.policies.native.Apex", version="1.0.0"), "onap.policies.native.apex.LinkMonitor", "1.0.0", settings.POLICY_BASICAUTH) - if policy_response: - logger.info("Policy created successfully") - else: - logger.info("Policy creation failed") + assert policy.get_policy(PolicyType(type="onap.policies.native.Apex", version="1.0.0"), "onap.policies.native.apex.LinkMonitor", "1.0.0", settings.POLICY_BASICAUTH) def deploy_policy(): """Deploy the policy.""" @@ -109,39 +96,32 @@ def deploy_policy(): policy_deployed = True break - if policy_deployed: - logger.info("Policy deployed successfully") - else: - logger.info("Failed to deploy policy") + assert policy_deployed + +def policy_log_detected(): + logger.info("Wait for a while for Apex engine to be ready before sending Dmaap event") + event = jinja_env().get_template("LinkFailureEvent.json.j2").render() + dmaap.send_link_failure_event(event) + if int(subprocess.getoutput('kubectl logs onap-policy-apex-pdp-0 -n onap | grep "Task Selection Execution: \'LinkMonitorPolicy:0.0.1:NULL:LinkFailureOrClearedState\'" | wc -l')) > 0: + logger.info("Apex engine is ready. LinkFailureEvent sent to Dmaap") + return True + return False def check_sdnc(): """Verify that apex has changed the sdnc status.""" logger.info("Check O-du/O-ru status") - SDNC_BASICAUTH = {'username': 'admin', 'password': 'Kp8bJ4SXszM0WXlhak3eHlcse2gAw84vaoGGmJvUy2U'} sdnc = OranSdnc() - status = sdnc.get_odu_oru_status("o-du-1122", "o-ru-11221", SDNC_BASICAUTH) - if status["o-ran-sc-du-hello-world:du-to-ru-connection"][0]["administrative-state"] == "LOCKED": - logger.info("The initial state of o-du o-ru connection is LOCKED") + status = sdnc.get_odu_oru_status("o-du-1122", "rrm-pol-2", settings.SDNC_BASICAUTH) + assert status["o-ran-sc-du-hello-world:radio-resource-management-policy-ratio"][0]["administrative-state"] == "locked" - logger.info("Wait for a while for Apex engine to be ready before sending Dmaap event") - event = jinja_env().get_template("LinkFailureEvent.json.j2").render() - for _ in range(60): - dmaap.send_link_failure_event(event) - if int(subprocess.getoutput('kubectl logs onap-policy-apex-pdp-0 -n onap | grep "Task Selection Execution: \'LinkMonitorPolicy:0.0.1:NULL:LinkFailureOrClearedState\'" | wc -l')) > 0: - logger.info("Apex engine is ready. LinkFailureEvent sent to Dmaap") - break - logger.info("Apex engine not ready yet, wait for a while and try again") - time.sleep(2) + wait(lambda: policy_log_detected(), sleep_seconds=10, timeout_seconds=60, waiting_for="Policy apex log") logger.info("Check O-du/O-ru status again") - status = sdnc.get_odu_oru_status("o-du-1122", "o-ru-11221", SDNC_BASICAUTH) - if status["o-ran-sc-du-hello-world:du-to-ru-connection"][0]["administrative-state"] == "UNLOCKED": - logger.info("The updated state of o-du o-ru connection is UNLOCKED") + status = sdnc.get_odu_oru_status("o-du-1122", "rrm-pol-2", settings.SDNC_BASICAUTH) + assert status["o-ran-sc-du-hello-world:radio-resource-management-policy-ratio"][0]["administrative-state"] == "unlocked" def test_apex_policy(): """Test the apex policy.""" - create_new_topic() - policy_component_ready() create_policy() deploy_policy() check_sdnc()