3 # ============LICENSE_START=======================================================
4 # ORAN SMO PACKAGE - PYTHONSDK TESTS
5 # ================================================================================
6 # Copyright (C) 2021-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 """Apex policy tests module."""
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
39 # Set working dir as python script location
40 abspath = os.path.abspath(__file__)
41 dname = os.path.dirname(abspath)
45 logging.config.dictConfig(settings.LOG_CONFIG)
46 logger = logging.getLogger("test APEX policy")
49 network_simulators = NetworkSimulators("./resources")
51 TOPIC_FAULT = '{"topicName": "unauthenticated.SEC_FAULT_OUTPUT"}'
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")
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")
62 network_simulators.start_network_simulators()
63 network_simulators.wait_for_network_simulators_to_be_running()
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...")
68 logger.info("Test Session setup completed successfully")
72 network_simulators.stop_network_simulators()
73 logger.info("Test Session cleanup done")
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)
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)
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)
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
99 assert policy_deployed
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")
111 """Verify that apex has changed the sdnc status."""
112 logger.info("Check O-du/O-ru status")
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"
117 wait(lambda: policy_log_detected(), sleep_seconds=10, timeout_seconds=60, waiting_for="Policy apex log")
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"
123 def test_apex_policy():
124 """Test the apex policy."""