smo-install/test/pythonsdk/src/orantests/disabled_test_cl_apex.py

   1 #!/usr/bin/env python3
   2 ###
   3 # ============LICENSE_START=======================================================
   4 # ORAN SMO PACKAGE - PYTHONSDK TESTS
   5 # ================================================================================
   6 # Copyright (C) 2022 AT&T Intellectual Property. All rights
   7 #                             reserved.
   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
  12 #
  13 # http://www.apache.org/licenses/LICENSE-2.0
  14 #
  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 # ===================================================================
  22 #
  23 ###
  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
  32
  33 import time
  34 import logging
  35 import logging.config
  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
  44
  45 logging.config.dictConfig(settings.LOG_CONFIG)
  46 logger = logging.getLogger("test Control Loops for O-RU Fronthaul Recovery usecase - Apex policy")
  47 dmaap = OranDmaap()
  48 clamp = ClampToscaTemplate(settings.CLAMP_BASICAUTH)
  49
  50 def create_topic():
  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)
  56
  57 def verify_topic_created():
  58     """Verify whether needed topic created."""
  59     logger.info("Verify topic created")
  60     topiclist = dmaap.get_all_topics({})
  61     topic_created = False
  62     for topic in topiclist:
  63         if topic["topicName"] == "unauthenticated.SEC_FAULT_OUTPUT":
  64             topic_created = True
  65             break
  66
  67     if topic_created:
  68         logger.info("Topic created successfully")
  69     else:
  70         logger.info("Topic creation failed")
  71
  72 def upload_commission(tosca_template):
  73     """
  74     Upload the tosca to commissioning.
  75
  76     Args:
  77         tosca_template : the tosca template to upload in json format
  78     Returns:
  79         the response from the upload action
  80     """
  81     logger.info("Upload tosca to commissioning")
  82     return clamp.upload_commission(tosca_template)
  83
  84 def create_instance(tosca_template):
  85     """
  86     Create a instance.
  87
  88         Args:
  89             tosca_template : the tosca template to create in json format
  90         Returns:
  91             the response from the creation action
  92     """
  93     logger.info("Create Instance")
  94     return clamp.create_instance(tosca_template)
  95
  96 def change_instance_status(new_status) -> str:
  97     """
  98     Change the instance statue.
  99
 100     Args:
 101         new_status : the new instance to be changed to
 102     Returns:
 103         the new status to be changed to
 104     """
 105     logger.info("Change Instance Status to %s", new_status)
 106     try:
 107         clamp.change_instance_status(new_status, "PMSH_Instance1", "1.2.3")
 108     except RequestError:
 109         logger.info("Change Instance Status request returned failed. Will query the instance status to double check whether the request is successful or not.")
 110
 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
 113     time.sleep(2)
 114     response = clamp.get_template_instance()
 115     return response["controlLoopList"][0]["orderedState"]
 116
 117 def verify_instance_status(new_status):
 118     """
 119     Verify whether the instance changed to the new status.
 120
 121     Args:
 122         new_status : the new status of the instance
 123     Returns:
 124         the boolean value indicating whether status changed successfully
 125     """
 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:
 129         return True
 130     return False
 131
 132 def verify_apex_policy_created():
 133     """
 134     Verify whether the Apex policy has deployed successfully.
 135
 136     Returns:
 137         the boolean value indicating whether policy deployed successfully
 138     """
 139     logger.info("Verify Apex policy is deployed")
 140     policy = OranPolicy()
 141     policy_status_list = policy.get_policy_status(settings.POLICY_BASICAUTH)
 142
 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"]):
 146             return True
 147
 148     logger.info("Failed to deploy Apex policy")
 149     return False
 150
 151 def delete_template_instance():
 152     """
 153     Delete the template instance.
 154
 155     Returns:
 156         the response from the deletion action
 157     """
 158     logger.info("Delete Instance")
 159     return clamp.delete_template_instance("PMSH_Instance1", "1.2.3")
 160
 161 def decommission_tosca():
 162     """
 163     Decommission the tosca template.
 164
 165     Returns:
 166         the response from the decommission action
 167     """
 168     logger.info("Decommission tosca")
 169     return clamp.decommission_template("ToscaServiceTemplateSimple", "1.0.0")
 170
 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)
 175
 176 def test_cl_oru_recovery():
 177     """The Closed Loop O-RU Fronthaul Recovery usecase Apex version."""
 178     create_topic()
 179     verify_topic_created()
 180
 181     tosca_template = jinja_env().get_template("commission_apex.json.j2").render()
 182
 183     response = upload_commission(tosca_template)
 184     assert response["errorDetails"] is None
 185
 186     response = create_instance(tosca_template)
 187     assert response["errorDetails"] is None
 188
 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")
 192
 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")
 196
 197     sdnc = OranSdnc()
 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"
 200
 201     send_dmaap_event()
 202
 203     assert verify_apex_policy_created()
 204
 205     time.sleep(20)
 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"
 209
 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")
 213
 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")
 217
 218     response = delete_template_instance()
 219     assert response["errorDetails"] is None
 220
 221     response = decommission_tosca()
 222     assert response["errorDetails"] is None