3 # ============LICENSE_START=======================================================
4 # ORAN SMO PACKAGE - PYTHONSDK TESTS
5 # ================================================================================
6 # Copyright (C) 2021-2022 AT&T Intellectual Property. All rights
8 # ================================================================================
9 # Licensed under the Apache License, Version 2.0 (the "License");
10 # you may not use this file except in compliance with the License.
11 # You may obtain a copy of the License at
13 # http://www.apache.org/licenses/LICENSE-2.0
15 # Unless required by applicable law or agreed to in writing, software
16 # distributed under the License is distributed on an "AS IS" BASIS,
17 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
18 # See the License for the specific language governing permissions and
19 # limitations under the License.
20 # ============LICENSE_END============================================
21 # ===================================================================
24 """Apex policy tests module."""
29 from onapsdk.configuration import settings
30 from oransdk.dmaap.dmaap import OranDmaap
31 from oransdk.policy.policy import OranPolicy, PolicyType
32 from oransdk.sdnc.sdnc import OranSdnc
33 from oransdk.utils.jinja import jinja_env
35 logging.config.dictConfig(settings.LOG_CONFIG)
36 logger = logging.getLogger("test APEX policy")
40 def create_new_topic():
41 """Create new topic."""
42 logger.info("Create new topic")
43 topic = '{ "topicName": "unauthenticated.SEC_FAULT_OUTPUT", "topicDescription": "test topic", "partitionCount": 1,"replicationCnCount": 1, "transactionEnabled": "false"}'
44 response = dmaap.create_topic(topic)
45 logger.info("response is: %s", response)
47 logger.info("Verify topic created")
48 topiclist = dmaap.get_all_topics({})
50 for topic in topiclist:
51 if topic["topicName"] == "unauthenticated.SEC_FAULT_OUTPUT":
56 logger.info("Topic created successfully")
58 logger.info("Topic creation failed")
60 def policy_component_ready():
61 """Check if components are ready."""
62 logger.info("Verify policy components are ready")
63 policy_ready = {"api_ready": False, "pap_ready": False, "apex_ready": False}
65 logger.info("Iteration %s", x)
66 policy_status = policy.get_components_status(settings.POLICY_BASICAUTH)
67 if (policy_status["api"]["healthy"] and not policy_ready["api_ready"]):
68 logger.info("Policy Api is ready")
69 policy_ready["api_ready"] = True
70 if (policy_status["pap"]["healthy"] and not policy_ready["pap_ready"]):
71 logger.info("Policy Pap is ready")
72 policy_ready["pap_ready"] = True
73 if (policy_status["pdps"]["apex"][0]["healthy"] == "HEALTHY" and not policy_ready["apex_ready"]):
74 logger.info("Policy Apex is ready")
75 policy_ready["apex_ready"] = True
76 if (policy_ready["api_ready"] and policy_ready["pap_ready"] and policy_ready["apex_ready"]):
77 logger.info("Policy status all ready")
81 if (not policy_ready["api_ready"] or not policy_ready["pap_ready"] or not policy_ready["apex_ready"]):
82 logger.info("Policy components are not ready. Exit the test.")
85 """Create the policy."""
86 logger.info("Create policy")
87 policy_data = jinja_env().get_template("ToscaPolicy.json.j2").render()
88 policy.create_policy(PolicyType(type="onap.policies.native.Apex", version="1.0.0"), policy_data, settings.POLICY_BASICAUTH)
90 logger.info("Verify whether policy created successfully")
91 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)
93 logger.info("Policy created successfully")
95 logger.info("Policy creation failed")
98 """Deploy the policy."""
99 logger.info("Deploy policy")
100 policy_to_deploy = jinja_env().get_template("DeployPolicyPAP.json.j2").render()
101 policy.deploy_policy(policy_to_deploy, settings.POLICY_BASICAUTH)
103 logger.info("Verify the policy is deployed")
104 policy_status_list = policy.get_policy_status(settings.POLICY_BASICAUTH)
105 policy_deployed = False
106 for status in policy_status_list:
107 logger.info("the status %s,%s,%s:", status["policy"]["name"], status["policy"]["version"], status["deploy"])
108 if (status["policy"]["name"] == "onap.policies.native.apex.LinkMonitor" and status["policy"]["version"] == "1.0.0" and status["deploy"]):
109 policy_deployed = True
113 logger.info("Policy deployed successfully")
115 logger.info("Failed to deploy policy")
118 """Verify that apex has changed the sdnc status."""
119 logger.info("Check O-du/O-ru status")
120 SDNC_BASICAUTH = {'username': 'admin', 'password': 'Kp8bJ4SXszM0WXlhak3eHlcse2gAw84vaoGGmJvUy2U'}
122 status = sdnc.get_odu_oru_status("o-du-1122", "o-ru-11221", SDNC_BASICAUTH)
123 if status["o-ran-sc-du-hello-world:du-to-ru-connection"][0]["administrative-state"] == "LOCKED":
124 logger.info("The initial state of o-du o-ru connection is LOCKED")
126 logger.info("Wait for a while for Apex engine to be ready before sending Dmaap event")
127 event = jinja_env().get_template("LinkFailureEvent.json.j2").render()
129 dmaap.send_link_failure_event(event)
130 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:
131 logger.info("Apex engine is ready. LinkFailureEvent sent to Dmaap")
133 logger.info("Apex engine not ready yet, wait for a while and try again")
136 logger.info("Check O-du/O-ru status again")
137 status = sdnc.get_odu_oru_status("o-du-1122", "o-ru-11221", SDNC_BASICAUTH)
138 if status["o-ran-sc-du-hello-world:du-to-ru-connection"][0]["administrative-state"] == "UNLOCKED":
139 logger.info("The updated state of o-du o-ru connection is UNLOCKED")
141 def test_apex_policy():
142 """Test the apex policy."""
144 policy_component_ready()