2 # SPDX-License-Identifier: Apache-2.0
8 from onapsdk.configuration import settings
9 from oransdk.dmaap.dmaap import OranDmaap
10 from oransdk.policy.policy import OranPolicy, PolicyType
11 from oransdk.sdnc.sdnc import OranSdnc
12 from oransdk.utils.jinja import jinja_env
14 logging.config.dictConfig(settings.LOG_CONFIG)
15 logger = logging.getLogger("test APEX policy")
19 logger.info("Create new topic")
20 topic = '{ "topicName": "unauthenticated.SEC_FAULT_OUTPUT", "topicDescription": "test topic", "partitionCount": 1, "replicationCnCount": 1, "transactionEnabled": "false"}'
21 response = dmaap.create_topic(topic)
22 logger.info("response is: %s", response)
24 logger.info("Verify topic created")
25 topiclist = dmaap.get_all_topics({})
27 for topic in topiclist:
28 if topic["topicName"] == "unauthenticated.SEC_FAULT_OUTPUT":
33 logger.info("Topic created successfully")
35 logger.info("Topic creation failed")
38 logger.info("Verify policy components are ready")
40 policy_ready = {"api_ready": False, "pap_ready": False, "apex_ready": False}
42 policy_status = policy.get_components_status(settings.POLICY_BASICAUTH)
43 if (policy_status["api"]["healthy"] and policy_ready["api_ready"] == False):
44 logger.info("Policy Api is ready")
45 policy_ready["api_ready"] = True
46 if (policy_status["pap"]["healthy"] and policy_ready["pap_ready"] == False):
47 logger.info("Policy Pap is ready")
48 policy_ready["pap_ready"] = True
49 if (policy_status["pdps"]["apex"][0]["healthy"] == "HEALTHY" and policy_ready["apex_ready"] == False):
50 logger.info("Policy Apex is ready")
51 policy_ready["apex_ready"] = True
52 if (policy_ready["api_ready"] and policy_ready["pap_ready"] and policy_ready["apex_ready"]):
53 logger.info("Policy status all ready")
57 if (policy_ready["api_ready"] == False or policy_ready["pap_ready"] == False or policy_ready["apex_ready"] == False):
58 logger.info("Policy components are not ready. Exit the test.")
61 logger.info("Create policy")
62 policy_data = jinja_env().get_template("ToscaPolicy.json.j2").render()
63 policy.create_policy(PolicyType(type="onap.policies.native.Apex", version="1.0.0"), policy_data, settings.POLICY_BASICAUTH)
65 logger.info("Verify whether policy created successfully")
66 policy_response = policy.get_policy(PolicyType(type="onap.policies.native.Apex", version="1.0.0"),
67 "onap.policies.native.apex.LinkMonitor", "1.0.0", settings.POLICY_BASICAUTH)
69 logger.info("Policy created successfully")
71 logger.info("Policy creation failed")
73 logger.info("Deploy policy")
74 deploy_policy = jinja_env().get_template("DeployPolicyPAP.json.j2").render()
75 policy.deploy_policy(deploy_policy, settings.POLICY_BASICAUTH)
77 logger.info("Verify the policy is deployed")
78 policy_status_list = policy.get_policy_status(settings.POLICY_BASICAUTH)
79 policy_deployed = False
80 for status in policy_status_list:
81 logger.info("the status %s,%s,%s:", status["policy"]["name"], status["policy"]["version"] , status["deploy"] )
82 if (status["policy"]["name"] == "onap.policies.native.apex.LinkMonitor" and status["policy"]["version"] == "1.0.0" and status["deploy"]):
83 policy_deployed = True
87 logger.info("Policy deployed successfully")
89 logger.info("Failed to deploy policy")
91 logger.info("Check O-du/O-ru status")
92 SDNC_BASICAUTH = { 'username': 'admin', 'password': 'Kp8bJ4SXszM0WXlhak3eHlcse2gAw84vaoGGmJvUy2U' }
94 status = sdnc.get_odu_oru_status("o-du-1122", "o-ru-11221", SDNC_BASICAUTH)
95 if status["o-ran-sc-du-hello-world:du-to-ru-connection"][0]["administrative-state"] == "LOCKED":
96 logger.info("The initial state of o-du o-ru connection is LOCKED")
98 logger.info("Wait for a while for Apex engine to be ready before sending Dmaap event")
100 event = jinja_env().get_template("LinkFailureEvent.json.j2").render()
102 dmaap.send_link_failure_event(event)
103 output = subprocess.getoutput('kubectl logs onap-policy-apex-pdp-0 -n onap | grep "Task Selection Execution: \'LinkMonitorPolicy:0.0.1:NULL:LinkFailureOrClearedState\'" | wc -l')
104 if (int(output) > 0):
105 logger.info("Apex engine is ready. LinkFailureEvent sent to Dmaap")
108 logger.info("Apex engine not ready yet, wait for a while and try again")
111 logger.info("Check O-du/O-ru status again")
112 status = sdnc.get_odu_oru_status("o-du-1122", "o-ru-11221", SDNC_BASICAUTH)
113 if status["o-ran-sc-du-hello-world:du-to-ru-connection"][0]["administrative-state"] == "UNLOCKED":
114 logger.info("The updated state of o-du o-ru connection is UNLOCKED")