3 # ============LICENSE_START=======================================================
4 # ORAN SMO PACKAGE - PYTHONSDK TESTS
5 # ================================================================================
6 # Copyright (C) 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 """Closed Loop Apex usecase tests module."""
25 # This usecase has limitations due to Clamp issue.
26 # 1. manually change clamp-be settings before running the test:
27 # - run command "kubectl -n onap edit cm onap-policy-clamp-be-configmap"
28 # find variable clamp.config.controlloop.runtime.url and change http into https
29 # - run command "kubectl rollout restart deployment onap-policy-clamp-be -n onap"
30 # and wait until policy-clamp-be pod restarted successfully
31 # 2. make sure using the policy-clamp-be version 6.2.0-snapshot-latest at this the moment
36 from onapsdk.configuration import settings
37 from onapsdk.exceptions import RequestError
38 from waiting import wait
39 from oransdk.dmaap.dmaap import OranDmaap
40 from oransdk.policy.policy import OranPolicy
41 from oransdk.policy.clamp import ClampToscaTemplate
42 from oransdk.sdnc.sdnc import OranSdnc
43 from oransdk.utils.jinja import jinja_env
45 logging.config.dictConfig(settings.LOG_CONFIG)
46 logger = logging.getLogger("test Control Loops for O-RU Fronthaul Recovery usecase - Apex policy")
48 clamp = ClampToscaTemplate(settings.CLAMP_BASICAUTH)
51 """Create the topic in Dmaap."""
52 logger.info("Create new topic")
53 topic = '{ "topicName": "unauthenticated.SEC_FAULT_OUTPUT", "topicDescription": "test topic", "partitionCount": 1, "replicationCnCount": 1, "transactionEnabled": "false"}'
54 response = dmaap.create_topic(topic)
55 logger.info("response is: %s", response)
57 def verify_topic_created():
58 """Verify whether needed topic created."""
59 logger.info("Verify topic created")
60 topiclist = dmaap.get_all_topics({})
62 for topic in topiclist:
63 if topic["topicName"] == "unauthenticated.SEC_FAULT_OUTPUT":
68 logger.info("Topic created successfully")
70 logger.info("Topic creation failed")
72 def upload_commission(tosca_template):
74 Upload the tosca to commissioning.
77 tosca_template : the tosca template to upload in json format
79 the response from the upload action
81 logger.info("Upload tosca to commissioning")
82 return clamp.upload_commission(tosca_template)
84 def create_instance(tosca_template):
89 tosca_template : the tosca template to create in json format
91 the response from the creation action
93 logger.info("Create Instance")
94 return clamp.create_instance(tosca_template)
96 def change_instance_status(new_status) -> str:
98 Change the instance statue.
101 new_status : the new instance to be changed to
103 the new status to be changed to
105 logger.info("Change Instance Status to %s", new_status)
107 clamp.change_instance_status(new_status, "PMSH_Instance1", "1.2.3")
109 logger.info("Change Instance Status request returned failed. Will query the instance status to double check whether the request is successful or not.")
111 # There's a bug in Clamp code, sometimes it returned 500, but actually the status has been changed successfully
112 # Thus we verify the status to determine whether it was successful or not
114 response = clamp.get_template_instance()
115 return response["controlLoopList"][0]["orderedState"]
117 def verify_instance_status(new_status):
119 Verify whether the instance changed to the new status.
122 new_status : the new status of the instance
124 the boolean value indicating whether status changed successfully
126 logger.info("Verify the Instance Status is updated to the expected status %s", new_status)
127 response = clamp.get_template_instance()
128 if response["controlLoopList"][0]["state"] == new_status:
132 def verify_apex_policy_created():
134 Verify whether the Apex policy has deployed successfully.
137 the boolean value indicating whether policy deployed successfully
139 logger.info("Verify Apex policy is deployed")
140 policy = OranPolicy()
141 policy_status_list = policy.get_policy_status(settings.POLICY_BASICAUTH)
143 for status in policy_status_list:
144 logger.info("the status %s,%s,%s:", status["policy"]["name"], status["policy"]["version"], status["deploy"])
145 if (status["policy"]["name"] == "operational.apex.linkmonitor" and status["policy"]["version"] == "1.0.0" and status["deploy"]):
148 logger.info("Failed to deploy Apex policy")
151 def delete_template_instance():
153 Delete the template instance.
156 the response from the deletion action
158 logger.info("Delete Instance")
159 return clamp.delete_template_instance("PMSH_Instance1", "1.2.3")
161 def decommission_tosca():
163 Decommission the tosca template.
166 the response from the decommission action
168 logger.info("Decommission tosca")
169 return clamp.decommission_template("ToscaServiceTemplateSimple", "1.0.0")
171 def send_dmaap_event():
172 """Send a event to Dmaap that should trigger the apex policy."""
173 event = jinja_env().get_template("LinkFailureEvent.json.j2").render()
174 dmaap.send_link_failure_event(event)
176 def test_cl_oru_recovery():
177 """The Closed Loop O-RU Fronthaul Recovery usecase Apex version."""
179 verify_topic_created()
181 tosca_template = jinja_env().get_template("commission_apex.json.j2").render()
183 response = upload_commission(tosca_template)
184 assert response["errorDetails"] is None
186 response = create_instance(tosca_template)
187 assert response["errorDetails"] is None
189 response = change_instance_status("PASSIVE")
190 assert response == "PASSIVE"
191 wait(lambda: verify_instance_status("PASSIVE"), sleep_seconds=5, timeout_seconds=60, waiting_for="Clamp instance switches to PASSIVE")
193 response = change_instance_status("RUNNING")
194 assert response == "RUNNING"
195 wait(lambda: verify_instance_status("RUNNING"), sleep_seconds=5, timeout_seconds=60, waiting_for="Clamp instance switches to RUNNING")
198 status = sdnc.get_odu_oru_status("o-du-1122", "rrm-pol-2", settings.SDNC_BASICAUTH)
199 assert status["o-ran-sc-du-hello-world:radio-resource-management-policy-ratio"][0]["administrative-state"] == "locked"
203 assert verify_apex_policy_created()
206 logger.info("Check O-du/O-ru status again")
207 status = sdnc.get_odu_oru_status("o-du-1122", "rrm-pol-2", settings.SDNC_BASICAUTH)
208 assert status["o-ran-sc-du-hello-world:radio-resource-management-policy-ratio"][0]["administrative-state"] == "unlocked"
210 response = change_instance_status("PASSIVE")
211 assert response == "PASSIVE"
212 wait(lambda: verify_instance_status("PASSIVE"), sleep_seconds=5, timeout_seconds=60, waiting_for="Clamp instance switches to PASSIVE")
214 response = change_instance_status("UNINITIALISED")
215 assert response == "UNINITIALISED"
216 wait(lambda: verify_instance_status("UNINITIALISED"), sleep_seconds=5, timeout_seconds=60, waiting_for="Clamp instance switches to UNINITIALISED")
218 response = delete_template_instance()
219 assert response["errorDetails"] is None
221 response = decommission_tosca()
222 assert response["errorDetails"] is None