2 ##############################################################################
3 # Copyright (c) 2020 Orange, Inc. and others. All rights reserved.
5 # All rights reserved. This program and the accompanying materials
6 # are made available under the terms of the Apache License, Version 2.0
7 # which accompanies this distribution, and is available at
8 # http://www.apache.org/licenses/LICENSE-2.0
9 ##############################################################################
11 # pylint: disable=no-member
12 # pylint: disable=too-many-public-methods
15 # pylint: disable=wrong-import-order
20 sys.path.append('transportpce_tests/common/')
21 # pylint: disable=wrong-import-position
22 # pylint: disable=import-error
23 import test_utils # nopep8
26 # pylint: disable=too-few-public-methods
29 # pylint: disable=invalid-name
36 # pylint: disable=too-few-public-methods
37 class UuidSubscriptions:
39 # pylint: disable=invalid-name
46 class TransportNbiNotificationstesting(unittest.TestCase):
47 cr_serv_input_data = {
50 "layer-protocol-name": "DSR",
51 "service-interface-point": {
52 "service-interface-point-uuid": "b1f4bd3b-7fa9-367b-a8ab-6e80293238df"
54 "administrative-state": "UNLOCKED",
55 "operational-state": "ENABLED",
56 "direction": "BIDIRECTIONAL",
58 "protection-role": "WORK",
59 "local-id": "XPDR-C1-XPDR1",
62 "value-name": "OpenROADM node id",
63 "value": "XPDR-C1-XPDR1"
68 "layer-protocol-name": "DSR",
69 "service-interface-point": {
70 "service-interface-point-uuid": "b5964ce9-274c-3f68-b4d1-83c0b61bc74e"
72 "administrative-state": "UNLOCKED",
73 "operational-state": "ENABLED",
74 "direction": "BIDIRECTIONAL",
76 "protection-role": "WORK",
77 "local-id": "XPDR-A1-XPDR1",
80 "value-name": "OpenROADM node id",
81 "value": "XPDR-A1-XPDR1"
86 "connectivity-constraint": {
87 "service-layer": "ETH",
88 "service-type": "POINT_TO_POINT_CONNECTIVITY",
89 "service-level": "Some service-level",
90 "requested-capacity": {
100 tapi_serv_details = {"service-id-or-name": "TBD"}
102 cr_notif_subs_input_data = {
103 "subscription-filter": {
104 "requested-notification-types": [
107 "requested-object-types": [
108 "CONNECTIVITY_SERVICE"
110 "requested-layer-protocols": [
113 "requested-object-identifier": [
114 "76d8f07b-ead5-4132-8eb8-cf3fdef7e079"
116 "include-content": True,
117 "local-id": "localId",
120 "value-name": "Subscription name",
121 "value": "test subscription"
125 "subscription-state": "ACTIVE"
128 cr_get_notif_list_input_data = {
129 "subscription-id-or-name": "c07e7fd1-0377-4fbf-8928-36c17b0d0d68",
130 "time-period": "time-period"
134 uuid_services = UuidServices()
135 uuid_subscriptions = UuidSubscriptions()
136 WAITING = 25 # nominal value is 300
137 NODE_VERSION_221 = '2.2.1'
141 # pylint: disable=unsubscriptable-object
142 # TODO: for lighty manage the activation of NBI notification feature
143 cls.init_failed_nbi = False
144 cls.init_failed_tapi = False
145 os.environ['JAVA_MIN_MEM'] = '1024M'
146 os.environ['JAVA_MAX_MEM'] = '4096M'
147 cls.processes = test_utils.start_tpce()
148 # NBI notification feature is not installed by default in Karaf
149 if "NO_ODL_STARTUP" not in os.environ or "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
150 print("installing NBI notification feature...")
151 result = test_utils.install_karaf_feature("odl-transportpce-nbinotifications")
152 if result.returncode != 0:
153 cls.init_failed_nbi = True
154 print("installing tapi feature...")
155 result = test_utils.install_karaf_feature("odl-transportpce-tapi")
156 if result.returncode != 0:
157 cls.init_failed_tapi = True
158 if cls.init_failed_nbi:
159 print("NBI notification installation feature failed...")
160 test_utils.shutdown_process(cls.processes[0])
162 if cls.init_failed_tapi:
163 print("tapi installation feature failed...")
164 test_utils.shutdown_process(cls.processes[0])
166 cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION_221),
167 ('roadma', cls.NODE_VERSION_221),
168 ('roadmc', cls.NODE_VERSION_221),
169 ('xpdrc', cls.NODE_VERSION_221)])
172 def tearDownClass(cls):
173 # pylint: disable=not-an-iterable
174 for process in cls.processes:
175 test_utils.shutdown_process(process)
176 print("all processes killed")
178 def setUp(self): # instruction executed before each test method
179 # pylint: disable=consider-using-f-string
180 print("execution of {}".format(self.id().split(".")[-1]))
182 def test_01_connect_xpdrA(self):
183 response = test_utils.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION_221))
184 self.assertEqual(response.status_code,
185 requests.codes.created, test_utils.CODE_SHOULD_BE_201)
187 def test_02_connect_xpdrC(self):
188 response = test_utils.mount_device("XPDR-C1", ('xpdrc', self.NODE_VERSION_221))
189 self.assertEqual(response.status_code,
190 requests.codes.created, test_utils.CODE_SHOULD_BE_201)
192 def test_03_connect_rdmA(self):
193 response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION_221))
194 self.assertEqual(response.status_code,
195 requests.codes.created, test_utils.CODE_SHOULD_BE_201)
197 def test_04_connect_rdmC(self):
198 response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION_221))
199 self.assertEqual(response.status_code,
200 requests.codes.created, test_utils.CODE_SHOULD_BE_201)
202 def test_05_connect_xpdrA_N1_to_roadmA_PP1(self):
203 response = test_utils.transportpce_api_rpc_request(
204 'transportpce-networkutils', 'init-xpdr-rdm-links',
205 {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1',
206 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
207 self.assertEqual(response['status_code'], requests.codes.ok)
208 self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
211 def test_06_connect_roadmA_PP1_to_xpdrA_N1(self):
212 response = test_utils.transportpce_api_rpc_request(
213 'transportpce-networkutils', 'init-rdm-xpdr-links',
214 {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1',
215 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
216 self.assertEqual(response['status_code'], requests.codes.ok)
217 self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
220 def test_07_connect_xpdrC_N1_to_roadmC_PP1(self):
221 response = test_utils.transportpce_api_rpc_request(
222 'transportpce-networkutils', 'init-xpdr-rdm-links',
223 {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1',
224 'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
225 self.assertEqual(response['status_code'], requests.codes.ok)
226 self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
229 def test_08_connect_roadmC_PP1_to_xpdrC_N1(self):
230 response = test_utils.transportpce_api_rpc_request(
231 'transportpce-networkutils', 'init-rdm-xpdr-links',
232 {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1',
233 'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
234 self.assertEqual(response['status_code'], requests.codes.ok)
235 self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
238 def test_09_add_omsAttributes_ROADMA_ROADMC(self):
239 # Config ROADMA-ROADMC oms-attributes
241 "auto-spanloss": "true",
242 "spanloss-base": 11.4,
243 "spanloss-current": 12,
244 "engineered-spanloss": 12.2,
245 "link-concatenation": [{
248 "SRLG-length": 100000,
250 response = test_utils.add_oms_attr_request(
251 "ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", data)
252 self.assertEqual(response.status_code, requests.codes.created)
254 def test_10_add_omsAttributes_ROADMC_ROADMA(self):
255 # Config ROADMC-ROADMA oms-attributes
257 "auto-spanloss": "true",
258 "spanloss-base": 11.4,
259 "spanloss-current": 12,
260 "engineered-spanloss": 12.2,
261 "link-concatenation": [{
264 "SRLG-length": 100000,
266 response = test_utils.add_oms_attr_request(
267 "ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX", data)
268 self.assertEqual(response.status_code, requests.codes.created)
270 # test service-create for Eth service from xpdr to xpdr
271 def test_11_create_connectivity_service_Ethernet(self):
272 response = test_utils.transportpce_api_rpc_request(
273 'tapi-connectivity', 'create-connectivity-service', self.cr_serv_input_data)
274 time.sleep(self.WAITING)
275 self.assertEqual(response['status_code'], requests.codes.ok)
276 self.uuid_services.eth = response['output']['service']['uuid']
277 # pylint: disable=consider-using-f-string
279 input_dict_1 = {'administrative-state': 'LOCKED',
280 'lifecycle-state': 'PLANNED',
281 'operational-state': 'DISABLED',
282 'service-type': 'POINT_TO_POINT_CONNECTIVITY',
283 'service-layer': 'ETH',
284 'connectivity-direction': 'BIDIRECTIONAL'
286 input_dict_2 = {'value-name': 'OpenROADM node id',
287 'value': 'XPDR-C1-XPDR1'}
288 input_dict_3 = {'value-name': 'OpenROADM node id',
289 'value': 'XPDR-A1-XPDR1'}
291 self.assertDictEqual(dict(input_dict_1, **response['output']['service']),
292 response['output']['service'])
293 self.assertDictEqual(dict(input_dict_2, **response['output']['service']['end-point'][0]['name'][0]),
294 response['output']['service']['end-point'][0]['name'][0])
295 self.assertDictEqual(dict(input_dict_3, **response['output']['service']['end-point'][1]['name'][0]),
296 response['output']['service']['end-point'][1]['name'][0])
298 def test_12_get_service_Ethernet(self):
299 response = test_utils.get_ordm_serv_list_attr_request("services", str(self.uuid_services.eth))
300 self.assertEqual(response['status_code'], requests.codes.ok)
301 self.assertEqual(response['services'][0]['administrative-state'], 'inService')
302 self.assertEqual(response['services'][0]['service-name'], str(self.uuid_services.eth))
303 self.assertEqual(response['services'][0]['connection-type'], 'service')
304 self.assertEqual(response['services'][0]['lifecycle-state'], 'planned')
307 def test_13_get_connectivity_service_Ethernet(self):
308 self.tapi_serv_details["service-id-or-name"] = str(self.uuid_services.eth)
309 response = test_utils.transportpce_api_rpc_request(
310 'tapi-connectivity', 'get-connectivity-service-details', self.tapi_serv_details)
311 self.assertEqual(response['status_code'], requests.codes.ok)
312 self.assertEqual(response['output']['service']['operational-state'], 'ENABLED')
313 self.assertEqual(response['output']['service']['name'][0]['value'], self.uuid_services.eth)
314 self.assertEqual(response['output']['service']['administrative-state'], 'UNLOCKED')
315 self.assertEqual(response['output']['service']['lifecycle-state'], 'INSTALLED')
317 def test_14_create_notifications_subscription_service(self):
318 self.cr_notif_subs_input_data["subscription-filter"]["requested-object-identifier"][0] =\
319 str(self.uuid_services.eth)
320 response = test_utils.transportpce_api_rpc_request(
321 'tapi-notification', 'create-notification-subscription-service', self.cr_notif_subs_input_data)
322 self.assertEqual(response['status_code'], requests.codes.ok)
323 self.uuid_subscriptions.eth = response['output']['subscription-service']['uuid']
324 self.assertEqual(response['output']['subscription-service']['subscription-filter']
325 ['requested-object-types'][0], 'CONNECTIVITY_SERVICE')
326 self.assertEqual(response['output']['subscription-service']['subscription-filter']
327 ['requested-notification-types'][0], 'ALARM_EVENT')
328 self.assertEqual(response['output']['subscription-service']['subscription-filter']
329 ['requested-object-identifier'][0], str(self.uuid_services.eth))
332 def test_15_change_status_port_roadma_srg(self):
333 self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3/0', 'C1',
336 "logical-connection-point": "SRG1-PP1",
337 "port-type": "client",
338 "circuit-id": "SRG1",
339 "administrative-state": "outOfService",
340 "port-qual": "roadm-external"
343 def test_16_get_tapi_notifications_connectivity_service_Ethernet(self):
344 self.cr_get_notif_list_input_data["subscription-id-or-name"] = str(self.uuid_subscriptions.eth)
345 response = test_utils.transportpce_api_rpc_request(
346 'tapi-notification', 'get-notification-list', self.cr_get_notif_list_input_data)
347 self.assertEqual(response['status_code'], requests.codes.ok)
348 self.assertEqual(response['output']['notification'][0]['target-object-identifier'], str(self.uuid_services.eth))
349 self.assertEqual(response['output']['notification'][0]['target-object-type'], 'CONNECTIVITY_SERVICE')
350 self.assertEqual(response['output']['notification'][0]['changed-attributes'][0]['new-value'], 'LOCKED')
351 self.assertEqual(response['output']['notification'][0]['changed-attributes'][1]['new-value'], 'DISABLED')
354 def test_17_restore_status_port_roadma_srg(self):
355 self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3/0', 'C1',
358 "logical-connection-point": "SRG1-PP1",
359 "port-type": "client",
360 "circuit-id": "SRG1",
361 "administrative-state": "inService",
362 "port-qual": "roadm-external"
365 def test_18_get_tapi_notifications_connectivity_service_Ethernet(self):
366 self.cr_get_notif_list_input_data["subscription-id-or-name"] = str(self.uuid_subscriptions.eth)
367 response = test_utils.transportpce_api_rpc_request(
368 'tapi-notification', 'get-notification-list', self.cr_get_notif_list_input_data)
369 self.assertEqual(response['status_code'], requests.codes.ok)
370 self.assertEqual(response['output']['notification'][0]['target-object-identifier'], str(self.uuid_services.eth))
371 self.assertEqual(response['output']['notification'][0]['target-object-type'], 'CONNECTIVITY_SERVICE')
372 self.assertEqual(response['output']['notification'][0]['changed-attributes'][0]['new-value'], 'UNLOCKED')
373 self.assertEqual(response['output']['notification'][0]['changed-attributes'][1]['new-value'], 'ENABLED')
376 def test_19_delete_connectivity_service_Ethernet(self):
377 self.tapi_serv_details["service-id-or-name"] = str(self.uuid_services.eth)
378 response = test_utils.transportpce_api_rpc_request(
379 'tapi-connectivity', 'delete-connectivity-service', self.tapi_serv_details)
380 self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content))
381 time.sleep(self.WAITING)
383 def test_20_disconnect_XPDRA(self):
384 response = test_utils.unmount_device("XPDR-A1")
385 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
387 def test_21_disconnect_XPDRC(self):
388 response = test_utils.unmount_device("XPDR-C1")
389 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
391 def test_22_disconnect_ROADMA(self):
392 response = test_utils.unmount_device("ROADM-A1")
393 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
395 def test_23_disconnect_ROADMC(self):
396 response = test_utils.unmount_device("ROADM-C1")
397 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
400 if __name__ == "__main__":
401 unittest.main(verbosity=2)