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 import os
  30 import pytest
  31 from waiting import wait
  32 from smo.network_simulators import NetworkSimulators
  33 from onapsdk.configuration import settings
  34 from oransdk.dmaap.dmaap import OranDmaap
  35 from oransdk.policy.policy import OranPolicy, PolicyType
  36 from oransdk.sdnc.sdnc import OranSdnc
  37 from oransdk.utils.jinja import jinja_env
  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 policy = OranPolicy()
  49 network_simulators = NetworkSimulators("./resources")
  50
  51 TOPIC_FAULT = '{"topicName": "unauthenticated.SEC_FAULT_OUTPUT"}'
  52
  53 @pytest.fixture(scope="module", autouse=True)
  54 def setup_simulators():
  55     """Setup the simulators before the executing the tests."""
  56     logger.info("Test class setup for Apex tests")
  57
  58     dmaap.create_topic(TOPIC_FAULT)
  59     # Purge the FAULT TOPIC
  60     wait(lambda: (dmaap.get_message_from_topic("unauthenticated.SEC_FAULT_OUTPUT", 5000, settings.DMAAP_GROUP, settings.DMAAP_USER).json() == []), sleep_seconds=10, timeout_seconds=60, waiting_for="DMaap topic unauthenticated.SEC_FAULT_OUTPUT to be empty")
  61
  62     network_simulators.start_network_simulators()
  63     network_simulators.wait_for_network_simulators_to_be_running()
  64
  65     # Wait enough time to have at least the SDNR notifications sent
  66     logger.info("Waiting 10s that SDNR sends all registration events to VES...")
  67     time.sleep(10)
  68     logger.info("Test Session setup completed successfully")
  69
  70     ### Cleanup code
  71     yield
  72     network_simulators.stop_network_simulators()
  73     logger.info("Test Session cleanup done")
  74
  75 def create_policy():
  76     """Create the policy."""
  77     logger.info("Create policy")
  78     policy_data = jinja_env().get_template("ToscaPolicy.json.j2").render()
  79     policy.create_policy(PolicyType(type="onap.policies.native.Apex", version="1.0.0"), policy_data, settings.POLICY_BASICAUTH)
  80
  81     logger.info("Verify whether policy created successfully")
  82     assert policy.get_policy(PolicyType(type="onap.policies.native.Apex", version="1.0.0"), "onap.policies.native.apex.LinkMonitor", "1.0.0", settings.POLICY_BASICAUTH)
  83
  84 def deploy_policy():
  85     """Deploy the policy."""
  86     logger.info("Deploy policy")
  87     policy_to_deploy = jinja_env().get_template("DeployPolicyPAP.json.j2").render()
  88     policy.deploy_policy(policy_to_deploy, settings.POLICY_BASICAUTH)
  89
  90     logger.info("Verify the policy is deployed")
  91     policy_status_list = policy.get_policy_status(settings.POLICY_BASICAUTH)
  92     policy_deployed = False
  93     for status in policy_status_list:
  94         logger.info("the status %s,%s,%s:", status["policy"]["name"], status["policy"]["version"], status["deploy"])
  95         if (status["policy"]["name"] == "onap.policies.native.apex.LinkMonitor" and status["policy"]["version"] == "1.0.0" and status["deploy"]):
  96             policy_deployed = True
  97             break
  98
  99     assert policy_deployed
 100
 101 def policy_log_detected():
 102     logger.info("Wait for a while for Apex engine to be ready before sending Dmaap event")
 103     event = jinja_env().get_template("LinkFailureEvent.json.j2").render()
 104     dmaap.send_link_failure_event(event)
 105     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:
 106         logger.info("Apex engine is ready. LinkFailureEvent sent to Dmaap")
 107         return True
 108     return False
 109
 110 def check_sdnc():
 111     """Verify that apex has changed the sdnc status."""
 112     logger.info("Check O-du/O-ru status")
 113     sdnc = OranSdnc()
 114     status = sdnc.get_odu_oru_status("o-du-1122", "rrm-pol-2", settings.SDNC_BASICAUTH)
 115     assert status["o-ran-sc-du-hello-world:radio-resource-management-policy-ratio"][0]["administrative-state"] == "locked"
 116
 117     wait(lambda: policy_log_detected(), sleep_seconds=10, timeout_seconds=60, waiting_for="Policy apex log")
 118
 119     logger.info("Check O-du/O-ru status again")
 120     status = sdnc.get_odu_oru_status("o-du-1122", "rrm-pol-2", settings.SDNC_BASICAUTH)
 121     assert status["o-ran-sc-du-hello-world:radio-resource-management-policy-ratio"][0]["administrative-state"] == "unlocked"
 122
 123 def test_apex_policy():
 124     """Test the apex policy."""
 125     create_policy()
 126     deploy_policy()
 127     check_sdnc()