smo-install/test/pythonsdk/src/orantests/oran_tests/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 usecase_name = "apex_usecase"
  59
  60 @pytest.fixture(scope="module", autouse=True)
  61 def setup_simulators():
  62     """Prepare the test environment before the executing the tests."""
  63     logger.info("Test class setup for Closed Loop Apex test")
  64
  65     dmaap_utils.clean_dmaap(settings.DMAAP_CL_GROUP, settings.DMAAP_CL_USER)
  66
  67     network_simulators.start_and_wait_network_simulators()
  68
  69     # Wait enough time to have at least the SDNR notifications sent
  70     logger.info("Waiting 10s that SDNR sends all registration events to VES...")
  71     time.sleep(10)
  72     logger.info("Test Session setup completed successfully")
  73
  74     ### Cleanup code
  75     yield
  76     # Finish and delete the cl instance
  77     clcommissioning_utils.clean_instance(usecase_name)
  78
  79     try:
  80         policy.undeploy_policy("operational.apex.linkmonitor", "1.0.0", settings.POLICY_BASICAUTH)
  81     except ResourceNotFound:
  82         logger.info("Policy already undeployed")
  83         try:
  84             policy.delete_policy(PolicyType(type="onap.policies.controlloop.operational.common.Apex", version="1.0.0"), "operational.apex.linkmonitor", "1.0.0", settings.POLICY_BASICAUTH)
  85         except ResourceNotFound:
  86             logger.info("Policy already deleted")
  87
  88     network_simulators.stop_network_simulators()
  89     time.sleep(10)
  90     logger.info("Test Session cleanup done")
  91
  92 def verify_apex_policy_created():
  93     """
  94     Verify whether the Apex policy has deployed successfully.
  95
  96     Returns:
  97         the boolean value indicating whether policy deployed successfully
  98     """
  99     logger.info("Verify Apex policy is deployed")
 100     policy_status_list = policy.get_policy_status(settings.POLICY_BASICAUTH)
 101     logger.info("policy_list: %s", policy_status_list)
 102     for status in policy_status_list:
 103         logger.info("the status %s,%s,%s,%s:", status["policy"]["name"], status["policy"]["version"], status["deploy"], status["state"])
 104         if (status["policy"]["name"] == "operational.apex.linkmonitor" and status["policy"]["version"] == "1.0.0" and status["deploy"] and status["state"] == "SUCCESS"):
 105             logger.info("Policy deployement OK")
 106             return True
 107
 108     logger.info("Failed to deploy Apex policy")
 109     return False
 110
 111 def send_dmaap_event():
 112     """Send a event to Dmaap that should trigger the apex policy."""
 113     event = jinja_env().get_template("LinkFailureEvent.json.j2").render()
 114     dmaap.send_link_failure_event(event)
 115
 116 def test_cl_apex():
 117     """The Closed Loop O-RU Fronthaul Recovery usecase Apex version."""
 118     logger.info("Upload tosca to commissioning")
 119     commissioning_payload = jinja_env().get_template("commission_apex.json.j2").render(dmaapGroup=settings.DMAAP_CL_GROUP, dmaapUser=settings.DMAAP_CL_USER)
 120     instance_payload = jinja_env().get_template("create_instance_apex.json.j2").render(dmaapGroup=settings.DMAAP_CL_GROUP, dmaapUser=settings.DMAAP_CL_USER, instanceName=usecase_name)
 121     assert clcommissioning_utils.create_instance(usecase_name, commissioning_payload, instance_payload) is True
 122
 123     sdnc = OranSdnc()
 124     status = sdnc.get_odu_oru_status("o-du-1122", "rrm-pol-2", settings.SDNC_BASICAUTH)
 125     assert status["o-ran-sc-du-hello-world:radio-resource-management-policy-ratio"][0]["administrative-state"] == "locked"
 126
 127     send_dmaap_event()
 128
 129     wait(lambda: verify_apex_policy_created(), sleep_seconds=10, timeout_seconds=60, waiting_for="Policy Deployment to be OK")
 130
 131     time.sleep(20)
 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"