smo-install/test/pythonsdk/src/orantests/test_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 logging
  27 import logging.config
  28 import os
  29 import pytest
  30 from waiting import wait
  31 from onapsdk.configuration import settings
  32 from oransdk.dmaap.dmaap import OranDmaap
  33 from oransdk.policy.policy import OranPolicy, PolicyType
  34 from oransdk.sdnc.sdnc import OranSdnc
  35 from oransdk.utils.jinja import jinja_env
  36 from smo.network_simulators import NetworkSimulators
  37
  38 # Set working dir as python script location
  39 abspath = os.path.abspath(__file__)
  40 dname = os.path.dirname(abspath)
  41 os.chdir(dname)
  42
  43
  44 logging.config.dictConfig(settings.LOG_CONFIG)
  45 logger = logging.getLogger("test APEX policy")
  46 dmaap = OranDmaap()
  47 policy = OranPolicy()
  48 network_simulators = NetworkSimulators("./resources")
  49
  50 policy_id = "onap.policies.native.apex.LinkMonitor"
  51 policy_version = "1.0.0"
  52 policy_type_id = "onap.policies.native.Apex"
  53 policy_type_version = "1.0.0"
  54 policy_type = PolicyType(type=policy_type_id, version=policy_type_version)
  55 engine_name = "LinkMonitorApexEngine"
  56 engine_version = "0.0.1"
  57 engine_id = "101"
  58 deployment_port = "12345"
  59
  60 @pytest.fixture(scope="module", autouse=True)
  61 def setup_simulators():
  62     """Setup the simulators before the executing the tests."""
  63     logger.info("Test class setup for Apex tests")
  64
  65     dmaap.create_topic(settings.DMAAP_TOPIC_FAULT_JSON)
  66     dmaap.create_topic(settings.DMAAP_TOPIC_PNFREG_JSON)
  67     # Purge the FAULT TOPIC
  68     wait(lambda: (dmaap.get_message_from_topic(settings.DMAAP_TOPIC_FAULT, 5000, settings.DMAAP_GROUP, settings.DMAAP_USER).json() == []), sleep_seconds=10, timeout_seconds=60, waiting_for="DMaap topic SEC_FAULT_OUTPUT to be empty")
  69     wait(lambda: (dmaap.get_message_from_topic(settings.DMAAP_TOPIC_PNFREG, 5000, settings.DMAAP_GROUP, settings.DMAAP_USER).json() == []), sleep_seconds=10, timeout_seconds=60, waiting_for="DMaap topic unauthenticated.VES_PNFREG_OUTPUT to be empty")
  70
  71     network_simulators.start_network_simulators()
  72     network_simulators.wait_for_network_simulators_to_be_running()
  73
  74     # Wait enough time to have at least the SDNR notifications sent
  75     logger.info("Waiting 10s that SDNR sends all registration events to VES...")
  76     time.sleep(10)
  77     logger.info("Test Session setup completed successfully")
  78
  79     ### Cleanup code
  80     yield
  81     network_simulators.stop_network_simulators()
  82     policy.undeploy_policy(policy_id, policy_version, settings.POLICY_BASICAUTH)
  83     policy.delete_policy(policy_type, policy_id, policy_version, settings.POLICY_BASICAUTH)
  84     logger.info("Test Session cleanup done")
  85
  86 def create_policy():
  87     """Create the policy."""
  88     logger.info("Create policy")
  89     policy_data = jinja_env().get_template("ToscaPolicy.json.j2").render(policyId=policy_id, policyVersion=policy_version, policyTypeId=policy_type_id, policyTypeVersion=policy_type_version, engineName=engine_name, engineVersion=engine_version, engineId=engine_id, deploymentPort=deployment_port, dmaapGroup=settings.DMAAP_GROUP, dmaapUser=settings.DMAAP_USER)
  90     policy.create_policy(policy_type, policy_data, settings.POLICY_BASICAUTH)
  91
  92     logger.info("Verify whether policy created successfully")
  93     assert policy.get_policy(policy_type, policy_id, policy_version, settings.POLICY_BASICAUTH).status_code == 200
  94
  95
  96 def deploy_policy():
  97     """Deploy the policy."""
  98     logger.info("Deploy policy")
  99     policy_to_deploy = jinja_env().get_template("DeployPolicyPAP.json.j2").render(policyId=policy_id, policyVersion=policy_version)
 100     policy.deploy_policy(policy_to_deploy, settings.POLICY_BASICAUTH)
 101     wait(lambda: check_policy_deployment(), sleep_seconds=10, timeout_seconds=60, waiting_for="Policy Deployment to be OK")
 102
 103 def check_policy_deployment():
 104     """Verify the policy deployment."""
 105     logger.info("Verify if the policy is deployed")
 106     policy_status_list = policy.get_policy_status(settings.POLICY_BASICAUTH)
 107
 108     for status in policy_status_list:
 109         logger.info("the status %s,%s,%s:", status["policy"]["name"], status["policy"]["version"], status["deploy"])
 110         if (status["policy"]["name"] == policy_id and status["policy"]["version"] == policy_version and status["deploy"] and status["state"] == "SUCCESS"):
 111             logger.info("Policy deployement OK")
 112             return True
 113     logger.info("Policy deployement not yet OK")
 114     return False
 115
 116 def send_dmaap_event():
 117     """Send a event to Dmaap that should trigger the apex policy."""
 118     event = jinja_env().get_template("LinkFailureEvent.json.j2").render()
 119     dmaap.send_link_failure_event(event)
 120
 121 def test_apex_policy():
 122     """Test the apex policy."""
 123     logger.info("Check O-du/O-ru status")
 124     sdnc = OranSdnc()
 125     status = sdnc.get_odu_oru_status("o-du-1122", "rrm-pol-2", settings.SDNC_BASICAUTH)
 126     assert status["o-ran-sc-du-hello-world:radio-resource-management-policy-ratio"][0]["administrative-state"] == "locked"
 127     send_dmaap_event()
 128     create_policy()
 129     deploy_policy()
 130     time.sleep(10)
 131
 132     logger.info("Check O-du/O-ru status again")
 133     status = sdnc.get_odu_oru_status("o-du-1122", "rrm-pol-2", settings.SDNC_BASICAUTH)
 134     assert status["o-ran-sc-du-hello-world:radio-resource-management-policy-ratio"][0]["administrative-state"] == "unlocked"