###
"""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__)
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()
### 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()
Code Review / it / dep.git / blobdiff