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

   1 #!/usr/bin/env python3
   2 ###
   3 # ============LICENSE_START=======================================================
   4 # ORAN SMO PACKAGE - PYTHONSDK TESTS
   5 # ================================================================================
   6 # Copyright (C) 2021-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 """Apex policy tests module."""
  25 import time
  26 import subprocess
  27 import logging
  28 import logging.config
  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
  34
  35 logging.config.dictConfig(settings.LOG_CONFIG)
  36 logger = logging.getLogger("test APEX policy")
  37 dmaap = OranDmaap()
  38 policy = OranPolicy()
  39
  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)
  46
  47     logger.info("Verify topic created")
  48     topiclist = dmaap.get_all_topics({})
  49     topic_created = False
  50     for topic in topiclist:
  51         if topic["topicName"] == "unauthenticated.SEC_FAULT_OUTPUT":
  52             topic_created = True
  53             break
  54
  55     if topic_created:
  56         logger.info("Topic created successfully")
  57     else:
  58         logger.info("Topic creation failed")
  59
  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}
  64     for x in range(60):
  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")
  78             break
  79
  80
  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.")
  83
  84 def create_policy():
  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)
  89
  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)
  92     if policy_response:
  93         logger.info("Policy created successfully")
  94     else:
  95         logger.info("Policy creation failed")
  96
  97 def deploy_policy():
  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)
 102
 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
 110             break
 111
 112     if policy_deployed:
 113         logger.info("Policy deployed successfully")
 114     else:
 115         logger.info("Failed to deploy policy")
 116
 117 def check_sdnc():
 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'}
 121     sdnc = OranSdnc()
 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")
 125
 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()
 128     for _ in range(60):
 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")
 132             break
 133         logger.info("Apex engine not ready yet, wait for a while and try again")
 134         time.sleep(2)
 135
 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")
 140
 141 def test_apex_policy():
 142     """Test the apex policy."""
 143     create_new_topic()
 144     policy_component_ready()
 145     create_policy()
 146     deploy_policy()
 147     check_sdnc()