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

   1 #!/usr/bin/env python3
   2 ###
   3 # ============LICENSE_START=======================================================
   4 # ORAN SMO PACKAGE - PYTHONSDK TESTS
   5 # ================================================================================
   6 # Copyright (C) 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 """Closed Loop Apex usecase tests module."""
  25
  26 import time
  27
  28 import logging
  29 import logging.config
  30 import os
  31 import pytest
  32 from onapsdk.configuration import settings
  33 from onapsdk.exceptions import ResourceNotFound
  34 from waiting import wait
  35 from oransdk.dmaap.dmaap import OranDmaap
  36 from oransdk.policy.clamp import ClampToscaTemplate
  37 from oransdk.policy.policy import OranPolicy, PolicyType
  38 from oransdk.sdnc.sdnc import OranSdnc
  39 from oransdk.utils.jinja import jinja_env
  40 from smo.network_simulators import NetworkSimulators
  41 from smo.dmaap import DmaapUtils
  42 from smo.cl_usecase import ClCommissioningUtils
  43
  44
  45 # Set working dir as python script location
  46 abspath = os.path.abspath(__file__)
  47 dname = os.path.dirname(abspath)
  48 os.chdir(dname)
  49
  50 logging.config.dictConfig(settings.LOG_CONFIG)
  51 logger = logging.getLogger("test Control Loops for O-RU Fronthaul Recovery usecase - Apex policy")
  52 dmaap = OranDmaap()
  53 dmaap_utils = DmaapUtils()
  54 clcommissioning_utils = ClCommissioningUtils()
  55 network_simulators = NetworkSimulators("./resources")
  56 clamp = ClampToscaTemplate(settings.CLAMP_BASICAUTH)
  57 policy = OranPolicy()
  58
  59 @pytest.fixture(scope="module", autouse=True)
  60 def setup_simulators():
  61     """Prepare the test environment before the executing the tests."""
  62     logger.info("Test class setup for Closed Loop Apex test")
  63
  64     dmaap_utils.clean_dmaap(settings.DMAAP_CL_GROUP, settings.DMAAP_CL_USER)
  65
  66     network_simulators.start_and_wait_network_simulators()
  67
  68     # Wait enough time to have at least the SDNR notifications sent
  69     logger.info("Waiting 10s that SDNR sends all registration events to VES...")
  70     time.sleep(10)
  71     logger.info("Test Session setup completed successfully")
  72
  73     ### Cleanup code
  74     yield
  75     # Finish and delete the cl instance
  76     clcommissioning_utils.clean_instance()
  77
  78     try:
  79         policy.undeploy_policy("operational.apex.linkmonitor", "1.0.0", settings.POLICY_BASICAUTH)
  80     except ResourceNotFound:
  81         logger.info("Policy already undeployed")
  82         try:
  83             policy.delete_policy(PolicyType(type="onap.policies.controlloop.operational.common.Apex", version="1.0.0"), "operational.apex.linkmonitor", "1.0.0", settings.POLICY_BASICAUTH)
  84         except ResourceNotFound:
  85             logger.info("Policy already deleted")
  86
  87     network_simulators.stop_network_simulators()
  88     time.sleep(10)
  89     logger.info("Test Session cleanup done")
  90
  91 def verify_apex_policy_created():
  92     """
  93     Verify whether the Apex policy has deployed successfully.
  94
  95     Returns:
  96         the boolean value indicating whether policy deployed successfully
  97     """
  98     logger.info("Verify Apex policy is deployed")
  99     policy_status_list = policy.get_policy_status(settings.POLICY_BASICAUTH)
 100
 101     for status in policy_status_list:
 102         logger.info("the status %s,%s,%s,%s:", status["policy"]["name"], status["policy"]["version"], status["deploy"], status["state"])
 103         if (status["policy"]["name"] == "operational.apex.linkmonitor" and status["policy"]["version"] == "1.0.0" and status["deploy"] and status["state"] == "SUCCESS"):
 104             logger.info("Policy deployement OK")
 105             return True
 106
 107     logger.info("Failed to deploy Apex policy")
 108     return False
 109
 110 def send_dmaap_event():
 111     """Send a event to Dmaap that should trigger the apex policy."""
 112     event = jinja_env().get_template("LinkFailureEvent.json.j2").render()
 113     dmaap.send_link_failure_event(event)
 114
 115 def test_cl_apex():
 116     """The Closed Loop O-RU Fronthaul Recovery usecase Apex version."""
 117     logger.info("Upload tosca to commissioning")
 118     tosca_template = jinja_env().get_template("commission_apex.json.j2").render(dmaapGroup=settings.DMAAP_CL_GROUP, dmaapUser=settings.DMAAP_CL_USER)
 119     assert clcommissioning_utils.create_instance(tosca_template) is True
 120
 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
 125     send_dmaap_event()
 126
 127     wait(lambda: verify_apex_policy_created(), sleep_seconds=10, timeout_seconds=60, waiting_for="Policy Deployment to be OK")
 128
 129     time.sleep(20)
 130     logger.info("Check O-du/O-ru status again")
 131     status = sdnc.get_odu_oru_status("o-du-1122", "rrm-pol-2", settings.SDNC_BASICAUTH)
 132     assert status["o-ran-sc-du-hello-world:radio-resource-management-policy-ratio"][0]["administrative-state"] == "unlocked"