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