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.dmaap import DmaapUtils
37 from smo.network_simulators import NetworkSimulators
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")
48 dmaap_utils = DmaapUtils()
50 network_simulators = NetworkSimulators("./resources")
52 policy_id = "onap.policies.native.apex.LinkMonitor"
53 policy_version = "1.0.0"
54 policy_type_id = "onap.policies.native.Apex"
55 policy_type_version = "1.0.0"
56 policy_type = PolicyType(type=policy_type_id, version=policy_type_version)
57 engine_name = "LinkMonitorApexEngine"
58 engine_version = "0.0.1"
60 deployment_port = "12345"
62 @pytest.fixture(scope="module", autouse=True)
63 def setup_simulators():
64 """Setup the simulators before the executing the tests."""
65 logger.info("Test class setup for Apex tests")
66 dmaap_utils.clean_dmaap(settings.DMAAP_GROUP, settings.DMAAP_USER)
67 network_simulators.start_and_wait_network_simulators()
69 # Wait enough time to have at least the SDNR notifications sent
71 logger.info("Waiting 10s that SDNR sends all registration events to VES...")
73 logger.info("Test Session setup completed successfully")
77 network_simulators.stop_network_simulators()
78 policy.undeploy_policy(policy_id, policy_version, settings.POLICY_BASICAUTH)
79 policy.delete_policy(policy_type, policy_id, policy_version, settings.POLICY_BASICAUTH)
81 logger.info("Test Session cleanup done")
84 """Create the policy."""
85 logger.info("Create policy")
86 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)
87 policy.create_policy(policy_type, policy_data, settings.POLICY_BASICAUTH)
89 logger.info("Verify whether policy created successfully")
90 assert policy.get_policy(policy_type, policy_id, policy_version, settings.POLICY_BASICAUTH).status_code == 200
94 """Deploy the policy."""
95 logger.info("Deploy policy")
96 policy_to_deploy = jinja_env().get_template("DeployPolicyPAP.json.j2").render(policyId=policy_id, policyVersion=policy_version)
97 policy.deploy_policy(policy_to_deploy, settings.POLICY_BASICAUTH)
98 wait(lambda: check_policy_deployment(), sleep_seconds=10, timeout_seconds=60, waiting_for="Policy Deployment to be OK")
100 def check_policy_deployment():
101 """Verify the policy deployment."""
102 logger.info("Verify if the policy is deployed")
103 policy_status_list = policy.get_policy_status(settings.POLICY_BASICAUTH)
105 for status in policy_status_list:
106 logger.info("the status %s,%s,%s:", status["policy"]["name"], status["policy"]["version"], status["deploy"])
107 if (status["policy"]["name"] == policy_id and status["policy"]["version"] == policy_version and status["deploy"] and status["state"] == "SUCCESS"):
108 logger.info("Policy deployement OK")
110 logger.info("Policy deployement not yet OK")
113 def send_dmaap_event():
114 """Send a event to Dmaap that should trigger the apex policy."""
115 event = jinja_env().get_template("LinkFailureEvent.json.j2").render()
116 dmaap.send_link_failure_event(event)
118 def test_apex_policy():
119 """Test the apex policy."""
120 logger.info("Check O-du/O-ru status")
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"
129 logger.info("Check O-du/O-ru status again")
130 status = sdnc.get_odu_oru_status("o-du-1122", "rrm-pol-2", settings.SDNC_BASICAUTH)
131 assert status["o-ran-sc-du-hello-world:radio-resource-management-policy-ratio"][0]["administrative-state"] == "unlocked"