--- /dev/null
+#!/usr/bin/env python3
+###
+# ============LICENSE_START=======================================================
+# ORAN SMO PACKAGE - PYTHONSDK TESTS
+# ================================================================================
+# Copyright (C) 2022 AT&T Intellectual Property. All rights
+# reserved.
+# ================================================================================
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ============LICENSE_END============================================
+# ===================================================================
+#
+###
+"""Closed Loop Apex usecase tests module."""
+# This usecase has limitations due to Clamp issue.
+# 1. manually change clamp-be settings before running the test:
+# - run command "kubectl -n onap edit cm onap-policy-clamp-be-configmap"
+# find variable clamp.config.controlloop.runtime.url and change http into https
+# - run command "kubectl rollout restart deployment onap-policy-clamp-be -n onap"
+# and wait until policy-clamp-be pod restarted successfully
+# 2. make sure using the policy-clamp-be version 6.2.0-snapshot-latest at this the moment
+
+import time
+import logging
+import logging.config
+from onapsdk.configuration import settings
+from onapsdk.exceptions import RequestError
+from waiting import wait
+from oransdk.dmaap.dmaap import OranDmaap
+from oransdk.policy.policy import OranPolicy
+from oransdk.policy.clamp import ClampToscaTemplate
+from oransdk.sdnc.sdnc import OranSdnc
+from oransdk.utils.jinja import jinja_env
+
+logging.config.dictConfig(settings.LOG_CONFIG)
+logger = logging.getLogger("test Control Loops for O-RU Fronthaul Recovery usecase - Apex policy")
+dmaap = OranDmaap()
+clamp = ClampToscaTemplate(settings.CLAMP_BASICAUTH)
+
+def create_topic():
+ """Create the topic in Dmaap."""
+ logger.info("Create new topic")
+ topic = '{ "topicName": "unauthenticated.SEC_FAULT_OUTPUT", "topicDescription": "test topic", "partitionCount": 1, "replicationCnCount": 1, "transactionEnabled": "false"}'
+ response = dmaap.create_topic(topic)
+ logger.info("response is: %s", response)
+
+def verify_topic_created():
+ """Verify whether needed topic created."""
+ logger.info("Verify topic created")
+ topiclist = dmaap.get_all_topics({})
+ topic_created = False
+ for topic in topiclist:
+ if topic["topicName"] == "unauthenticated.SEC_FAULT_OUTPUT":
+ topic_created = True
+ break
+
+ if topic_created:
+ logger.info("Topic created successfully")
+ else:
+ logger.info("Topic creation failed")
+
+def upload_commission(tosca_template):
+ """
+ Upload the tosca to commissioning.
+
+ Args:
+ tosca_template : the tosca template to upload in json format
+ Returns:
+ the response from the upload action
+ """
+ logger.info("Upload tosca to commissioning")
+ return clamp.upload_commission(tosca_template)
+
+def create_instance(tosca_template):
+ """
+ Create a instance.
+
+ Args:
+ tosca_template : the tosca template to create in json format
+ Returns:
+ the response from the creation action
+ """
+ logger.info("Create Instance")
+ return clamp.create_instance(tosca_template)
+
+def change_instance_status(new_status) -> str:
+ """
+ Change the instance statue.
+
+ Args:
+ new_status : the new instance to be changed to
+ Returns:
+ the new status to be changed to
+ """
+ logger.info("Change Instance Status to %s", new_status)
+ try:
+ clamp.change_instance_status(new_status, "PMSH_Instance1", "1.2.3")
+ except RequestError:
+ logger.info("Change Instance Status request returned failed. Will query the instance status to double check whether the request is successful or not.")
+
+ # There's a bug in Clamp code, sometimes it returned 500, but actually the status has been changed successfully
+ # Thus we verify the status to determine whether it was successful or not
+ time.sleep(2)
+ response = clamp.get_template_instance()
+ return response["controlLoopList"][0]["orderedState"]
+
+def verify_instance_status(new_status):
+ """
+ Verify whether the instance changed to the new status.
+
+ Args:
+ new_status : the new status of the instance
+ Returns:
+ the boolean value indicating whether status changed successfully
+ """
+ logger.info("Verify the Instance Status is updated to the expected status %s", new_status)
+ response = clamp.get_template_instance()
+ if response["controlLoopList"][0]["state"] == new_status:
+ return True
+ return False
+
+def verify_apex_policy_created():
+ """
+ Verify whether the Apex policy has deployed successfully.
+
+ Returns:
+ the boolean value indicating whether policy deployed successfully
+ """
+ logger.info("Verify Apex policy is deployed")
+ policy = OranPolicy()
+ policy_status_list = policy.get_policy_status(settings.POLICY_BASICAUTH)
+
+ 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"] == "operational.apex.linkmonitor" and status["policy"]["version"] == "1.0.0" and status["deploy"]):
+ return True
+
+ logger.info("Failed to deploy Apex policy")
+ return False
+
+def delete_template_instance():
+ """
+ Delete the template instance.
+
+ Returns:
+ the response from the deletion action
+ """
+ logger.info("Delete Instance")
+ return clamp.delete_template_instance("PMSH_Instance1", "1.2.3")
+
+def decommission_tosca():
+ """
+ Decommission the tosca template.
+
+ Returns:
+ the response from the decommission action
+ """
+ logger.info("Decommission tosca")
+ return clamp.decommission_template("ToscaServiceTemplateSimple", "1.0.0")
+
+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)
+
+def test_cl_oru_recovery():
+ """The Closed Loop O-RU Fronthaul Recovery usecase Apex version."""
+ create_topic()
+ verify_topic_created()
+
+ tosca_template = jinja_env().get_template("commission_apex.json.j2").render()
+
+ response = upload_commission(tosca_template)
+ assert response["errorDetails"] is None
+
+ response = create_instance(tosca_template)
+ assert response["errorDetails"] is None
+
+ response = change_instance_status("PASSIVE")
+ assert response == "PASSIVE"
+ wait(lambda: verify_instance_status("PASSIVE"), sleep_seconds=5, timeout_seconds=60, waiting_for="Clamp instance switches to PASSIVE")
+
+ response = change_instance_status("RUNNING")
+ assert response == "RUNNING"
+ wait(lambda: verify_instance_status("RUNNING"), sleep_seconds=5, timeout_seconds=60, waiting_for="Clamp instance switches to RUNNING")
+
+ 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"
+
+ send_dmaap_event()
+
+ assert verify_apex_policy_created()
+
+ time.sleep(20)
+ 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"
+
+ response = change_instance_status("PASSIVE")
+ assert response == "PASSIVE"
+ wait(lambda: verify_instance_status("PASSIVE"), sleep_seconds=5, timeout_seconds=60, waiting_for="Clamp instance switches to PASSIVE")
+
+ response = change_instance_status("UNINITIALISED")
+ assert response == "UNINITIALISED"
+ wait(lambda: verify_instance_status("UNINITIALISED"), sleep_seconds=5, timeout_seconds=60, waiting_for="Clamp instance switches to UNINITIALISED")
+
+ response = delete_template_instance()
+ assert response["errorDetails"] is None
+
+ response = decommission_tosca()
+ assert response["errorDetails"] is None
Code Review / it / dep.git / blobdiff