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 """O1 tests module."""
31 from onapsdk.configuration import settings
32 from waiting import wait
33 from smo.network_simulators import NetworkSimulators
34 from oransdk.dmaap.dmaap import OranDmaap
35 from oransdk.sdnc.sdnc import OranSdnc
37 # Set working dir as python script location
38 abspath = os.path.abspath(__file__)
39 dname = os.path.dirname(abspath)
42 logging.config.dictConfig(settings.LOG_CONFIG)
43 logger = logging.getLogger("Test O1")
45 network_simulators = NetworkSimulators("./resources")
47 test_session_timestamp = datetime.datetime.now()
50 @pytest.fixture(scope="module", autouse=True)
51 def setup_simulators():
52 """Setup the simulators before the executing the tests."""
53 logger.info("Test class setup for O1 tests")
55 # Do a first get to register the o1test/o1test user in DMAAP
56 # all registration messages will then be stored for the registration tests.
57 # If it exists already it clears all cached events.
58 dmaap.create_topic(settings.DMAAP_TOPIC_PNFREG_JSON)
59 dmaap.create_topic(settings.DMAAP_TOPIC_FAULT_JSON)
60 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")
61 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 unauthenticated.SEC_FAULT_OUTPUT to be empty")
62 network_simulators.start_network_simulators()
63 network_simulators.wait_for_network_simulators_to_be_running()
64 # ADD DU RESTART just in case
65 # Wait enough time to have at least the SDNR notifications sent
66 logger.info("Waiting 20s that SDNR sends all registration events to VES...")
68 logger.info("Enabling faults/events reporting on SDNR")
69 network_simulators.enable_events_for_all_simulators()
70 # logger.info("Waiting 20s that the Dmaap faults topic is created...")
72 # Preparing the DMaap to cache all the events for the fault topics.
73 # If it exists already it clears all cached events.
74 logger.info("Waiting 120s to have registration and faults events in DMaap")
76 logger.info("Test Session setup completed successfully")
80 network_simulators.stop_network_simulators()
81 logger.info("Test Session cleanup done")
83 def create_registration_structure(events):
84 """Decode the registration events list."""
85 devices_found_in_events = dict()
87 event_json = json.loads(event)
88 logger.info("Registration json decoded: %s", str(event_json))
89 devices_found_in_events[event_json["event"]["commonEventHeader"]["sourceName"]] = event_json["event"]["commonEventHeader"]["reportingEntityName"]
91 logger.info("Devices found in events:%s", devices_found_in_events)
92 return devices_found_in_events
94 def create_faults_structure(events):
95 """Decode the fault events list."""
96 faults_found_in_events = dict()
98 event_json = json.loads(event)
99 logger.info("Fault json decoded: %s", str(event_json))
100 if event_json["event"]["commonEventHeader"]["sourceName"] in faults_found_in_events:
101 faults_found_in_events[event_json["event"]["commonEventHeader"]["sourceName"]] += 1
103 faults_found_in_events[event_json["event"]["commonEventHeader"]["sourceName"]] = 1
104 logger.info("Faults found in events: %s", faults_found_in_events)
105 return faults_found_in_events
107 def test_devices_in_sdnc():
108 """Verify that the devices are well defined in SDNC."""
109 logger.info("Verify if devices are well in SDNC")
110 for device in settings.NETWORK_SIMULATORS_LIST:
111 logger.info("Verify if %s is well in SDNR", device)
112 assert OranSdnc.get_devices(device, settings.SDNC_BASICAUTH) == 200
114 def validate_faults_timestamp(faults):
115 """Extract only the faults returned by SDNC that have been raised during this test."""
117 for fault in faults['data-provider:output']['data']:
118 converted_fault_timestamp = datetime.datetime.strptime(fault['timestamp'], "%Y-%m-%dT%H:%M:%S.%fZ")
119 logger.info("Comparing fault timestamp %s (%s) to session test timestamp %s", converted_fault_timestamp, fault['timestamp'], test_session_timestamp)
120 if converted_fault_timestamp > test_session_timestamp:
121 valid_faults.append(fault)
122 logger.info("Valid faults array: %s", valid_faults)
125 def test_device_faults_in_sdnc():
126 """Verify that the device faults are well defined in SDNC."""
127 logger.info("Verify is there is any events")
128 for device in settings.NETWORK_SIMULATORS_DU_RU_LIST:
129 faults = OranSdnc.get_events(settings.SDNC_BASICAUTH, device).json()
130 logger.info("Verify if %s has events", device)
131 assert len(validate_faults_timestamp(faults)) >= 3
133 def test_network_devices_registration_in_dmaap():
134 """Validate that the devices are well registered in SDNR and forwarded to VES."""
135 logger.info("Verify if SDNR sends well the RU registration to VES by checking in DMAAP")
136 # As the user has been registered in DMAAP during test session init,
137 # that call should return all sims registered by SDNR
138 all_registrations = []
141 while (events := dmaap.get_message_from_topic(settings.DMAAP_TOPIC_PNFREG, 30000, settings.DMAAP_GROUP, settings.DMAAP_USER).json()) != []:
142 logger.info("Getting a first set of event: %s", events)
143 all_registrations += events
145 logger.info("ALl registration events received: %s", all_registrations)
146 # events should be a list of messages
147 logger.info("Verify if the number of events is well >= to the number of expected devices")
148 # The DU can send multiple times message to VES and SDNR can send multiple time event for RU
149 assert len(all_registrations) >= (len(settings.NETWORK_SIMULATORS_DU_RU_LIST))
150 devices_registered = create_registration_structure(all_registrations)
152 # Each device must be at least one time in the event list
153 for sim_name in settings.NETWORK_SIMULATORS_DU_RU_LIST:
154 logger.info("Checking if %s is in events list", sim_name)
155 assert sim_name in devices_registered
156 if "o-ru" in sim_name:
157 logger.info("RU event detected checking SDNR has well registered it")
158 assert "ONAP SDN-R" in devices_registered[sim_name]
159 elif "o-du" in sim_name:
160 logger.info("DU detected checking it has well registered itself")
161 assert "o-du" in devices_registered[sim_name]
163 def test_device_faults_in_dmaap():
164 """Verify that device faults are well sent to DMAAP by SDNR."""
165 logger.info("Verify if SDNR forwards well the faults sent by the simulators to DMAAP")
166 events = dmaap.get_message_from_topic(settings.DMAAP_TOPIC_FAULT, 30000, settings.DMAAP_GROUP, settings.DMAAP_USER).json()
167 logger.info("Verify if faults have well been received for each device")
168 assert len(events) > 0
169 faults_received = create_faults_structure(events)
171 # Each device must have some faults
172 for sim_name in settings.NETWORK_SIMULATORS_DU_RU_LIST:
173 logger.info("Check if %s has at least >=3 faults", sim_name)
174 assert sim_name in faults_received and faults_received[sim_name] >= 3