X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=smo-install%2Ftest%2Fpythonsdk%2Fsrc%2Forantests%2Ftest_apex_policy.py;fp=smo-install%2Ftest%2Fpythonsdk%2Fsrc%2Forantests%2Fdisable_apex_policy.py;h=7210dfdb6f2914551b2da9bb340cc46b46443464;hb=7b4b4007db003d28ee347e3d373e8ac605581cb5;hp=fa42df2b8486adbe4c7ac1370c6f0f078e55fd45;hpb=a8414e7e44dea72da6f4f235e14f7fef59b4e266;p=it%2Fdep.git diff --git a/smo-install/test/pythonsdk/src/orantests/disable_apex_policy.py b/smo-install/test/pythonsdk/src/orantests/test_apex_policy.py similarity index 62% rename from smo-install/test/pythonsdk/src/orantests/disable_apex_policy.py rename to smo-install/test/pythonsdk/src/orantests/test_apex_policy.py index fa42df2b..7210dfdb 100644 --- a/smo-install/test/pythonsdk/src/orantests/disable_apex_policy.py +++ b/smo-install/test/pythonsdk/src/orantests/test_apex_policy.py @@ -23,18 +23,17 @@ ### """Apex policy tests module.""" 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 +from smo.network_simulators import NetworkSimulators # Set working dir as python script location abspath = os.path.abspath(__file__) @@ -48,16 +47,26 @@ dmaap = OranDmaap() policy = OranPolicy() network_simulators = NetworkSimulators("./resources") -TOPIC_FAULT = '{"topicName": "unauthenticated.SEC_FAULT_OUTPUT"}' +policy_id = "onap.policies.native.apex.LinkMonitor" +policy_version = "1.0.0" +policy_type_id = "onap.policies.native.Apex" +policy_type_version = "1.0.0" +policy_type = PolicyType(type=policy_type_id, version=policy_type_version) +engine_name = "LinkMonitorApexEngine" +engine_version = "0.0.1" +engine_id = "101" +deployment_port = "12345" @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) + 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("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") + 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() @@ -70,58 +79,56 @@ def setup_simulators(): ### Cleanup code yield 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) logger.info("Test Session cleanup done") def create_policy(): """Create the policy.""" logger.info("Create policy") - policy_data = jinja_env().get_template("ToscaPolicy.json.j2").render() - policy.create_policy(PolicyType(type="onap.policies.native.Apex", version="1.0.0"), policy_data, settings.POLICY_BASICAUTH) + policy_data = jinja_env().get_template("ToscaPolicy.json.j2").render(policyId=policy_id, policyVersion=policy_version, policyTypeId=policy_type_id, policyTypeVersion=policy_type_version, engineName=engine_name, engineVersion=engine_version, engineId=engine_id, deploymentPort=deployment_port, dmaapGroup=settings.DMAAP_GROUP, dmaapUser=settings.DMAAP_USER) + policy.create_policy(policy_type, policy_data, settings.POLICY_BASICAUTH) logger.info("Verify whether policy created successfully") - 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) + assert policy.get_policy(policy_type, policy_id, policy_version, settings.POLICY_BASICAUTH).status_code == 200 + def deploy_policy(): """Deploy the policy.""" logger.info("Deploy policy") - policy_to_deploy = jinja_env().get_template("DeployPolicyPAP.json.j2").render() + policy_to_deploy = jinja_env().get_template("DeployPolicyPAP.json.j2").render(policyId=policy_id, policyVersion=policy_version) policy.deploy_policy(policy_to_deploy, settings.POLICY_BASICAUTH) + wait(lambda: check_policy_deployment(), sleep_seconds=10, timeout_seconds=60, waiting_for="Policy Deployment to be OK") - logger.info("Verify the policy is deployed") +def check_policy_deployment(): + """Verify the policy deployment.""" + logger.info("Verify if the policy is deployed") policy_status_list = policy.get_policy_status(settings.POLICY_BASICAUTH) - policy_deployed = False + for status in policy_status_list: logger.info("the status %s,%s,%s:", status["policy"]["name"], status["policy"]["version"], status["deploy"]) - if (status["policy"]["name"] == "onap.policies.native.apex.LinkMonitor" and status["policy"]["version"] == "1.0.0" and status["deploy"]): - policy_deployed = True - break - - assert policy_deployed + if (status["policy"]["name"] == policy_id and status["policy"]["version"] == policy_version and status["deploy"] and status["state"] == "SUCCESS"): + logger.info("Policy deployement OK") + return True + logger.info("Policy deployement not yet OK") + return False -def policy_log_detected(): - logger.info("Wait for a while for Apex engine to be ready before sending Dmaap event") +def send_dmaap_event(): + """Send a event to Dmaap that should trigger the apex policy.""" 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.""" +def test_apex_policy(): + """Test the apex policy.""" logger.info("Check O-du/O-ru status") sdnc = OranSdnc() 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" - - wait(lambda: policy_log_detected(), sleep_seconds=10, timeout_seconds=60, waiting_for="Policy apex log") + send_dmaap_event() + create_policy() + deploy_policy() + time.sleep(10) logger.info("Check O-du/O-ru status again") 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_policy() - deploy_policy() - check_sdnc()