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."""
30 from waiting import wait
31 from onapsdk.configuration import settings
32 from oransdk.dmaap.dmaap import OranDmaap
33 from oransdk.policy.policy import OranPolicy, PolicyType
34 from oransdk.sdnc.sdnc import OranSdnc
35 from oransdk.utils.jinja import jinja_env
36 from smo.network_simulators import NetworkSimulators
38 # Set working dir as python script location
39 abspath = os.path.abspath(__file__)
40 dname = os.path.dirname(abspath)
44 logging.config.dictConfig(settings.LOG_CONFIG)
45 logger = logging.getLogger("test APEX policy")
48 network_simulators = NetworkSimulators("./resources")
50 policy_id = "onap.policies.native.apex.LinkMonitor"
51 policy_version = "1.0.0"
52 policy_type_id = "onap.policies.native.Apex"
53 policy_type_version = "1.0.0"
54 policy_type = PolicyType(type=policy_type_id, version=policy_type_version)
55 engine_name = "LinkMonitorApexEngine"
56 engine_version = "0.0.1"
58 deployment_port = "12345"
60 @pytest.fixture(scope="module", autouse=True)
61 def setup_simulators():
62 """Setup the simulators before the executing the tests."""
63 logger.info("Test class setup for Apex tests")
65 dmaap.create_topic(settings.DMAAP_TOPIC_FAULT_JSON)
66 dmaap.create_topic(settings.DMAAP_TOPIC_PNFREG_JSON)
67 # Purge the FAULT TOPIC
68 wait(lambda: (dmaap.get_message_from_topic(settings.DMAAP_TOPIC_FAULT, 5000, settings.DMAAP_GROUP, settings.DMAAP_USER).json() == []), sleep_seconds=10, timeout_seconds=60, waiting_for="DMaap topic SEC_FAULT_OUTPUT to be empty")
69 wait(lambda: (dmaap.get_message_from_topic(settings.DMAAP_TOPIC_PNFREG, 5000, settings.DMAAP_GROUP, settings.DMAAP_USER).json() == []), sleep_seconds=10, timeout_seconds=60, waiting_for="DMaap topic unauthenticated.VES_PNFREG_OUTPUT to be empty")
71 network_simulators.start_network_simulators()
72 network_simulators.wait_for_network_simulators_to_be_running()
74 # Wait enough time to have at least the SDNR notifications sent
75 logger.info("Waiting 10s that SDNR sends all registration events to VES...")
77 logger.info("Test Session setup completed successfully")
81 network_simulators.stop_network_simulators()
82 policy.undeploy_policy(policy_id, policy_version, settings.POLICY_BASICAUTH)
83 policy.delete_policy(policy_type, policy_id, policy_version, settings.POLICY_BASICAUTH)
84 logger.info("Test Session cleanup done")
87 """Create the policy."""
88 logger.info("Create policy")
89 policy_data = jinja_env().get_template("ToscaPolicy.json.j2").render(policyId=policy_id, policyVersion=policy_version, policyTypeId=policy_type_id, policyTypeVersion=policy_type_version, engineName=engine_name, engineVersion=engine_version, engineId=engine_id, deploymentPort=deployment_port, dmaapGroup=settings.DMAAP_GROUP, dmaapUser=settings.DMAAP_USER)
90 policy.create_policy(policy_type, policy_data, settings.POLICY_BASICAUTH)
92 logger.info("Verify whether policy created successfully")
93 assert policy.get_policy(policy_type, policy_id, policy_version, settings.POLICY_BASICAUTH).status_code == 200
97 """Deploy the policy."""
98 logger.info("Deploy policy")
99 policy_to_deploy = jinja_env().get_template("DeployPolicyPAP.json.j2").render(policyId=policy_id, policyVersion=policy_version)
100 policy.deploy_policy(policy_to_deploy, settings.POLICY_BASICAUTH)
101 wait(lambda: check_policy_deployment(), sleep_seconds=10, timeout_seconds=60, waiting_for="Policy Deployment to be OK")
103 def check_policy_deployment():
104 """Verify the policy deployment."""
105 logger.info("Verify if the policy is deployed")
106 policy_status_list = policy.get_policy_status(settings.POLICY_BASICAUTH)
108 for status in policy_status_list:
109 logger.info("the status %s,%s,%s:", status["policy"]["name"], status["policy"]["version"], status["deploy"])
110 if (status["policy"]["name"] == policy_id and status["policy"]["version"] == policy_version and status["deploy"] and status["state"] == "SUCCESS"):
111 logger.info("Policy deployement OK")
113 logger.info("Policy deployement not yet OK")
116 def send_dmaap_event():
117 """Send a event to Dmaap that should trigger the apex policy."""
118 event = jinja_env().get_template("LinkFailureEvent.json.j2").render()
119 dmaap.send_link_failure_event(event)
121 def test_apex_policy():
122 """Test the apex policy."""
123 logger.info("Check O-du/O-ru status")
125 status = sdnc.get_odu_oru_status("o-du-1122", "rrm-pol-2", settings.SDNC_BASICAUTH)
126 assert status["o-ran-sc-du-hello-world:radio-resource-management-policy-ratio"][0]["administrative-state"] == "locked"
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"