#!/usr/bin/env python3 ### # ============LICENSE_START======================================================= # ORAN SMO PACKAGE - PYTHONSDK TESTS # ================================================================================ # Copyright (C) 2021-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============================================ # =================================================================== # ### """Apex policy tests module.""" import time import subprocess import logging import logging.config from onapsdk.configuration import settings from oransdk.dmaap.dmaap import OranDmaap from oransdk.policy.policy import OranPolicy, PolicyType from oransdk.sdnc.sdnc import OranSdnc from oransdk.utils.jinja import jinja_env logging.config.dictConfig(settings.LOG_CONFIG) logger = logging.getLogger("test APEX policy") dmaap = OranDmaap() policy = OranPolicy() def create_new_topic(): """Create new topic.""" 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) 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 policy_component_ready(): """Check if components are ready.""" logger.info("Verify policy components are ready") policy_ready = {"api_ready": False, "pap_ready": False, "apex_ready": False} for x in range(60): logger.info("Iteration %s", x) policy_status = policy.get_components_status(settings.POLICY_BASICAUTH) if (policy_status["api"]["healthy"] and not policy_ready["api_ready"]): logger.info("Policy Api is ready") policy_ready["api_ready"] = True if (policy_status["pap"]["healthy"] and not policy_ready["pap_ready"]): logger.info("Policy Pap is ready") policy_ready["pap_ready"] = True if (policy_status["pdps"]["apex"][0]["healthy"] == "HEALTHY" and not policy_ready["apex_ready"]): logger.info("Policy Apex is ready") policy_ready["apex_ready"] = True if (policy_ready["api_ready"] and policy_ready["pap_ready"] and policy_ready["apex_ready"]): logger.info("Policy status all ready") break if (not policy_ready["api_ready"] or not policy_ready["pap_ready"] or not policy_ready["apex_ready"]): logger.info("Policy components are not ready. Exit the test.") def create_policy(): """Create the policy.""" logger.info("Create policy") policy_data = jinja_env().get_template("ToscaPolicy.json.j2").render() policy.create_policy(PolicyType(type="onap.policies.native.Apex", version="1.0.0"), policy_data, settings.POLICY_BASICAUTH) logger.info("Verify whether policy created successfully") 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) if policy_response: logger.info("Policy created successfully") else: logger.info("Policy creation failed") def deploy_policy(): """Deploy the policy.""" logger.info("Deploy policy") policy_to_deploy = jinja_env().get_template("DeployPolicyPAP.json.j2").render() policy.deploy_policy(policy_to_deploy, settings.POLICY_BASICAUTH) logger.info("Verify the policy is deployed") policy_status_list = policy.get_policy_status(settings.POLICY_BASICAUTH) policy_deployed = False 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"] == "onap.policies.native.apex.LinkMonitor" and status["policy"]["version"] == "1.0.0" and status["deploy"]): policy_deployed = True break if policy_deployed: logger.info("Policy deployed successfully") else: logger.info("Failed to deploy policy") def check_sdnc(): """Verify that apex has changed the sdnc status.""" logger.info("Check O-du/O-ru status") SDNC_BASICAUTH = {'username': 'admin', 'password': 'Kp8bJ4SXszM0WXlhak3eHlcse2gAw84vaoGGmJvUy2U'} sdnc = OranSdnc() status = sdnc.get_odu_oru_status("o-du-1122", "o-ru-11221", SDNC_BASICAUTH) if status["o-ran-sc-du-hello-world:du-to-ru-connection"][0]["administrative-state"] == "LOCKED": logger.info("The initial state of o-du o-ru connection is LOCKED") logger.info("Wait for a while for Apex engine to be ready before sending Dmaap event") event = jinja_env().get_template("LinkFailureEvent.json.j2").render() for _ in range(60): dmaap.send_link_failure_event(event) 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: logger.info("Apex engine is ready. LinkFailureEvent sent to Dmaap") break logger.info("Apex engine not ready yet, wait for a while and try again") time.sleep(2) logger.info("Check O-du/O-ru status again") status = sdnc.get_odu_oru_status("o-du-1122", "o-ru-11221", SDNC_BASICAUTH) if status["o-ran-sc-du-hello-world:du-to-ru-connection"][0]["administrative-state"] == "UNLOCKED": logger.info("The updated state of o-du o-ru connection is UNLOCKED") def test_apex_policy(): """Test the apex policy.""" create_new_topic() policy_component_ready() create_policy() deploy_policy() check_sdnc()