3 # ============LICENSE_START=======================================================
4 # ORAN SMO PACKAGE - PYTHONSDK TESTS
5 # ================================================================================
6 # Copyright (C) 2022 AT&T Intellectual Property. All rights
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
13 # http://www.apache.org/licenses/LICENSE-2.0
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 # ===================================================================
24 """Closed Loop Apex usecase tests module."""
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
45 # Set working dir as python script location
46 abspath = os.path.abspath(__file__)
47 dname = os.path.dirname(abspath)
50 logging.config.dictConfig(settings.LOG_CONFIG)
51 logger = logging.getLogger("test Control Loops for O-RU Fronthaul Recovery usecase - Apex policy")
53 dmaap_utils = DmaapUtils()
54 clcommissioning_utils = ClCommissioningUtils()
55 network_simulators = NetworkSimulators("./resources")
56 clamp = ClampToscaTemplate(settings.CLAMP_BASICAUTH)
58 usecase_name = "apex_usecase"
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")
65 dmaap_utils.clean_dmaap(settings.DMAAP_CL_GROUP, settings.DMAAP_CL_USER)
67 network_simulators.start_and_wait_network_simulators()
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...")
72 logger.info("Test Session setup completed successfully")
76 # Finish and delete the cl instance
77 clcommissioning_utils.clean_instance(usecase_name)
80 policy.undeploy_policy("operational.apex.linkmonitor", "1.0.0", settings.POLICY_BASICAUTH)
81 except ResourceNotFound:
82 logger.info("Policy already undeployed")
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")
88 network_simulators.stop_network_simulators()
90 logger.info("Test Session cleanup done")
92 def verify_apex_policy_created():
94 Verify whether the Apex policy has deployed successfully.
97 the boolean value indicating whether policy deployed successfully
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")
108 logger.info("Failed to deploy Apex policy")
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)
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
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"
129 wait(lambda: verify_apex_policy_created(), sleep_seconds=10, timeout_seconds=60, waiting_for="Policy Deployment to be OK")
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"