2 ##############################################################################
3 # Copyright (c) 2017 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
18 # 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 class TransportPCEFulltesting(unittest.TestCase):
29 cr_serv_input_data = {
30 "sdnc-request-header": {
31 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
32 "rpc-action": "service-create",
33 "request-system-id": "appname",
34 "notification-url": "http://localhost:8585/NotificationServer/notify"
36 "service-name": "service1",
37 "common-id": "ASATT1234567",
38 "connection-type": "service",
40 "service-rate": "100",
42 "service-format": "Ethernet",
46 "port-device-name": "XPDR-A1-XPDR1",
48 "port-name": "1/0/2-PLUG-CLIENT",
49 "port-rack": "000000.00",
53 "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00",
54 "lgx-port-name": "LGX Back.3",
55 "lgx-port-rack": "000000.00",
56 "lgx-port-shelf": "00"
62 "port-device-name": "XPDR-A1-XPDR1",
64 "port-name": "1/0/2-PLUG-CLIENT",
65 "port-rack": "000000.00",
69 "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00",
70 "lgx-port-name": "LGX Back.4",
71 "lgx-port-rack": "000000.00",
72 "lgx-port-shelf": "00"
79 "service-rate": "100",
81 "service-format": "Ethernet",
85 "port-device-name": "XPDR-C1-XPDR2",
86 "port-type": "router",
87 "port-name": "1/0/2-PLUG-CLIENT",
88 "port-rack": "000000.00",
92 "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00",
93 "lgx-port-name": "LGX Back.29",
94 "lgx-port-rack": "000000.00",
95 "lgx-port-shelf": "00"
101 "port-device-name": "XPDR-C1-XPDR2",
102 "port-type": "router",
103 "port-name": "1/0/2-PLUG-CLIENT",
104 "port-rack": "000000.00",
108 "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00",
109 "lgx-port-name": "LGX Back.29",
110 "lgx-port-rack": "000000.00",
111 "lgx-port-shelf": "00"
117 "due-date": "2016-11-28T00:00:01Z",
118 "operator-contact": "pw1234"
121 del_serv_input_data = {
122 "sdnc-request-header": {
123 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
124 "rpc-action": "service-delete",
125 "request-system-id": "appname",
126 "notification-url": "http://localhost:8585/NotificationServer/notify"},
127 "service-delete-req-info": {
128 "service-name": "TBD",
129 "tail-retention": "no"}
132 WAITING = 20 # nominal value is 300
133 NODE_VERSION = '2.2.1'
137 cls.processes = test_utils.start_tpce()
138 cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION),
139 ('roadma', cls.NODE_VERSION),
140 ('roadmc', cls.NODE_VERSION),
141 ('xpdrc', cls.NODE_VERSION)])
144 def tearDownClass(cls):
145 # pylint: disable=not-an-iterable
146 for process in cls.processes:
147 test_utils.shutdown_process(process)
148 print("all processes killed")
150 def setUp(self): # instruction executed before each test method
151 # pylint: disable=consider-using-f-string
152 print("execution of {}".format(self.id().split(".")[-1]))
154 def test_01_connect_xpdrA(self):
155 response = test_utils.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION))
156 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
158 def test_02_connect_xpdrC(self):
159 response = test_utils.mount_device("XPDR-C1", ('xpdrc', self.NODE_VERSION))
160 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
162 def test_03_connect_rdmA(self):
163 response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION))
164 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
166 def test_04_connect_rdmC(self):
167 response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION))
168 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
170 def test_05_connect_xprdA_N1_to_roadmA_PP1(self):
171 response = test_utils.transportpce_api_rpc_request(
172 'transportpce-networkutils', 'init-xpdr-rdm-links',
173 {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1',
174 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
175 self.assertEqual(response['status_code'], requests.codes.ok)
176 self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
179 def test_06_connect_roadmA_PP1_to_xpdrA_N1(self):
180 response = test_utils.transportpce_api_rpc_request(
181 'transportpce-networkutils', 'init-rdm-xpdr-links',
182 {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1',
183 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
184 self.assertEqual(response['status_code'], requests.codes.ok)
185 self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
188 def test_07_connect_xprdC_xpdr1_N1_to_roadmC_PP1(self):
189 response = test_utils.transportpce_api_rpc_request(
190 'transportpce-networkutils', 'init-xpdr-rdm-links',
191 {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1',
192 'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
193 self.assertEqual(response['status_code'], requests.codes.ok)
194 self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
197 def test_08_connect_roadmC_PP1_to_xpdrC_xprd1_N1(self):
198 response = test_utils.transportpce_api_rpc_request(
199 'transportpce-networkutils', 'init-rdm-xpdr-links',
200 {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1',
201 'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
202 self.assertEqual(response['status_code'], requests.codes.ok)
203 self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
206 def test_09_connect_xprdA_N2_to_roadmA_PP2(self):
207 response = test_utils.transportpce_api_rpc_request(
208 'transportpce-networkutils', 'init-xpdr-rdm-links',
209 {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '2',
210 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}})
211 self.assertEqual(response['status_code'], requests.codes.ok)
212 self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
215 def test_10_connect_roadmA_PP2_to_xpdrA_N2(self):
216 response = test_utils.transportpce_api_rpc_request(
217 'transportpce-networkutils', 'init-rdm-xpdr-links',
218 {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '2',
219 'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}})
220 self.assertEqual(response['status_code'], requests.codes.ok)
221 self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
224 def test_11_connect_xprdC_xpdr2_N1_to_roadmC_PP2(self):
225 response = test_utils.transportpce_api_rpc_request(
226 'transportpce-networkutils', 'init-xpdr-rdm-links',
227 {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '2', 'network-num': '1',
228 'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}})
229 self.assertEqual(response['status_code'], requests.codes.ok)
230 self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
233 def test_12_connect_roadmC_PP2_to_xpdrC_xpdr2_N1(self):
234 response = test_utils.transportpce_api_rpc_request(
235 'transportpce-networkutils', 'init-rdm-xpdr-links',
236 {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '2', 'network-num': '1',
237 'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}})
238 self.assertEqual(response['status_code'], requests.codes.ok)
239 self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
242 def test_13_add_omsAttributes_ROADMA_ROADMC(self):
243 # Config ROADMA-ROADMC oms-attributes
245 "auto-spanloss": "true",
246 "spanloss-base": 11.4,
247 "spanloss-current": 12,
248 "engineered-spanloss": 12.2,
249 "link-concatenation": [{
252 "SRLG-length": 100000,
254 response = test_utils.add_oms_attr_request("ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX",
256 self.assertEqual(response.status_code, requests.codes.created)
258 def test_14_add_omsAttributes_ROADMC_ROADMA(self):
259 # Config ROADMC-ROADMA oms-attributes
261 "auto-spanloss": "true",
262 "spanloss-base": 11.4,
263 "spanloss-current": 12,
264 "engineered-spanloss": 12.2,
265 "link-concatenation": [{
268 "SRLG-length": 100000,
270 response = test_utils.add_oms_attr_request("ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX",
272 self.assertEqual(response.status_code, requests.codes.created)
274 # test service-create for Eth service from xpdr to xpdr
275 def test_15_create_eth_service2(self):
276 self.cr_serv_input_data["service-name"] = "service2"
277 response = test_utils.transportpce_api_rpc_request(
278 'org-openroadm-service', 'service-create',
279 self.cr_serv_input_data)
280 self.assertEqual(response['status_code'], requests.codes.ok)
281 self.assertIn('PCE calculation in progress',
282 response['output']['configuration-response-common']['response-message'])
283 time.sleep(self.WAITING)
285 def test_16_get_eth_service2(self):
286 response = test_utils.get_ordm_serv_list_attr_request("services", "service2")
287 self.assertEqual(response['status_code'], requests.codes.ok)
288 self.assertEqual(response['services'][0]['administrative-state'], 'inService')
289 self.assertEqual(response['services'][0]['service-name'], 'service2')
290 self.assertEqual(response['services'][0]['connection-type'], 'service')
291 self.assertEqual(response['services'][0]['lifecycle-state'], 'planned')
294 def test_17_check_xc1_ROADMA(self):
295 response = test_utils.check_node_attribute_request(
296 "ROADM-A1", "roadm-connections", "SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
297 self.assertEqual(response['status_code'], requests.codes.ok)
298 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
299 self.assertDictEqual(
301 'connection-name': 'SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768',
302 'opticalControlMode': 'gainLoss',
303 'target-output-power': -3.0
304 }, **response['roadm-connections'][0]), response['roadm-connections'][0])
305 self.assertDictEqual({'src-if': 'SRG1-PP1-TXRX-nmc-761:768'}, response['roadm-connections'][0]['source'])
306 self.assertDictEqual({'dst-if': 'DEG2-TTP-TXRX-nmc-761:768'}, response['roadm-connections'][0]['destination'])
309 def test_18_check_xc1_ROADMC(self):
310 response = test_utils.check_node_attribute_request(
311 "ROADM-C1", "roadm-connections", "SRG1-PP2-TXRX-DEG1-TTP-TXRX-761:768")
312 self.assertEqual(response['status_code'], requests.codes.ok)
313 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
314 self.assertDictEqual(
316 'connection-name': 'SRG1-PP2-TXRX-DEG1-TTP-TXRX-761:768',
317 'opticalControlMode': 'gainLoss',
318 'target-output-power': -3.0
319 }, **response['roadm-connections'][0]), response['roadm-connections'][0])
320 self.assertDictEqual({'src-if': 'SRG1-PP2-TXRX-nmc-761:768'}, response['roadm-connections'][0]['source'])
321 self.assertDictEqual({'dst-if': 'DEG1-TTP-TXRX-nmc-761:768'}, response['roadm-connections'][0]['destination'])
324 def test_19_check_topo_XPDRA(self):
325 response = test_utils.get_ietf_network_node_request('openroadm-topology', 'XPDR-A1-XPDR1', 'config')
326 self.assertEqual(response['status_code'], requests.codes.ok)
327 liste_tp = response['node']['ietf-network-topology:termination-point']
329 if ele['tp-id'] == 'XPDR1-NETWORK1':
332 float(ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength']['frequency']))
335 float(ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength']['width']))
336 elif ele['tp-id'] in ('XPDR1-CLIENT1', 'XPDR1-CLIENT2'):
337 self.assertNotIn('org-openroadm-network-topology:xpdr-client-attributes', dict.keys(ele))
338 elif ele['tp-id'] == 'XPDR1-NETWORK2':
339 self.assertIn('org-openroadm-network-topology:xpdr-network-attributes', dict.keys(ele))
342 def test_20_check_topo_ROADMA_SRG1(self):
343 response = test_utils.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-SRG1', 'config')
344 self.assertEqual(response['status_code'], requests.codes.ok)
345 freq_map = base64.b64decode(
346 response['node']['org-openroadm-network-topology:srg-attributes']['avail-freq-maps'][0]['freq-map'])
347 freq_map_array = [int(x) for x in freq_map]
348 self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
349 liste_tp = response['node']['ietf-network-topology:termination-point']
351 if ele['tp-id'] == 'SRG1-PP1-TXRX':
352 freq_map = base64.b64decode(
353 ele['org-openroadm-network-topology:pp-attributes']['avail-freq-maps'][0]['freq-map'])
354 freq_map_array = [int(x) for x in freq_map]
355 self.assertEqual(freq_map_array[95], 0, "Index 1 should not be available")
356 elif ele['tp-id'] == 'SRG1-PP2-TXRX':
357 self.assertNotIn('avail-freq-maps', dict.keys(ele))
360 def test_21_check_topo_ROADMA_DEG2(self):
361 response = test_utils.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-DEG2', 'config')
362 self.assertEqual(response['status_code'], requests.codes.ok)
363 freq_map = base64.b64decode(
364 response['node']['org-openroadm-network-topology:degree-attributes']['avail-freq-maps'][0]['freq-map'])
365 freq_map_array = [int(x) for x in freq_map]
366 self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
367 liste_tp = response['node']['ietf-network-topology:termination-point']
369 if ele['tp-id'] == 'DEG2-CTP-TXRX':
370 freq_map = base64.b64decode(
371 ele['org-openroadm-network-topology:ctp-attributes']['avail-freq-maps'][0]['freq-map'])
372 freq_map_array = [int(x) for x in freq_map]
373 self.assertEqual(freq_map_array[95], 0, "Index 1 should not be available")
374 elif ele['tp-id'] == 'DEG2-TTP-TXRX':
375 freq_map = base64.b64decode(
376 ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map'])
377 freq_map_array = [int(x) for x in freq_map]
378 self.assertEqual(freq_map_array[95], 0, "Index 1 should not be available")
381 def test_22_create_eth_service1(self):
382 self.cr_serv_input_data["service-name"] = "service1"
383 del self.cr_serv_input_data["service-a-end"]["tx-direction"][0]["port"]["port-device-name"]
384 del self.cr_serv_input_data["service-a-end"]["tx-direction"][0]["port"]["port-name"]
385 del self.cr_serv_input_data["service-a-end"]["rx-direction"][0]["port"]["port-device-name"]
386 del self.cr_serv_input_data["service-a-end"]["rx-direction"][0]["port"]["port-name"]
387 del self.cr_serv_input_data["service-z-end"]["tx-direction"][0]["port"]["port-device-name"]
388 del self.cr_serv_input_data["service-z-end"]["tx-direction"][0]["port"]["port-name"]
389 del self.cr_serv_input_data["service-z-end"]["rx-direction"][0]["port"]["port-device-name"]
390 del self.cr_serv_input_data["service-z-end"]["rx-direction"][0]["port"]["port-name"]
391 response = test_utils.transportpce_api_rpc_request(
392 'org-openroadm-service', 'service-create',
393 self.cr_serv_input_data)
394 self.assertEqual(response['status_code'], requests.codes.ok)
395 self.assertIn('PCE calculation in progress',
396 response['output']['configuration-response-common']['response-message'])
397 time.sleep(self.WAITING)
399 def test_23_get_eth_service1(self):
400 response = test_utils.get_ordm_serv_list_attr_request("services", "service1")
401 self.assertEqual(response['status_code'], requests.codes.ok)
402 self.assertEqual(response['services'][0]['administrative-state'], 'inService')
403 self.assertEqual(response['services'][0]['service-name'], 'service1')
404 self.assertEqual(response['services'][0]['connection-type'], 'service')
405 self.assertEqual(response['services'][0]['lifecycle-state'], 'planned')
408 def test_24_check_xc1_ROADMA(self):
409 response = test_utils.check_node_attribute_request(
410 "ROADM-A1", "roadm-connections", "DEG2-TTP-TXRX-SRG1-PP2-TXRX-753:760")
411 self.assertEqual(response['status_code'], requests.codes.ok)
412 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
413 self.assertDictEqual(
415 'connection-name': 'DEG2-TTP-TXRX-SRG1-PP2-TXRX-753:760',
416 'opticalControlMode': 'power'
417 }, **response['roadm-connections'][0]), response['roadm-connections'][0])
418 self.assertDictEqual({'src-if': 'DEG2-TTP-TXRX-nmc-753:760'}, response['roadm-connections'][0]['source'])
419 self.assertDictEqual({'dst-if': 'SRG1-PP2-TXRX-nmc-753:760'}, response['roadm-connections'][0]['destination'])
422 def test_25_check_topo_XPDRA(self):
423 response = test_utils.get_ietf_network_node_request('openroadm-topology', 'XPDR-A1-XPDR1', 'config')
424 self.assertEqual(response['status_code'], requests.codes.ok)
425 liste_tp = response['node']['ietf-network-topology:termination-point']
427 if ele['tp-id'] == 'XPDR1-NETWORK2':
430 float(ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength']['frequency']))
433 float(ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength']['width']))
434 elif ele['tp-id'] in ('XPDR1-CLIENT1'):
435 self.assertNotIn('org-openroadm-network-topology:xpdr-client-attributes', dict.keys(ele))
438 def test_26_check_topo_ROADMA_SRG1(self):
439 response = test_utils.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-SRG1', 'config')
440 self.assertEqual(response['status_code'], requests.codes.ok)
441 freq_map = base64.b64decode(
442 response['node']['org-openroadm-network-topology:srg-attributes']['avail-freq-maps'][0]['freq-map'])
443 freq_map_array = [int(x) for x in freq_map]
444 self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
445 self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available")
446 liste_tp = response['node']['ietf-network-topology:termination-point']
448 if ele['tp-id'] == 'SRG1-PP1-TXRX':
449 freq_map = base64.b64decode(
450 ele['org-openroadm-network-topology:pp-attributes']['avail-freq-maps'][0]['freq-map'])
451 freq_map_array = [int(x) for x in freq_map]
452 self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
453 self.assertEqual(freq_map_array[94], 255, "Lambda 2 should be available")
454 elif ele['tp-id'] == 'SRG1-PP2-TXRX':
455 freq_map = base64.b64decode(
456 ele['org-openroadm-network-topology:pp-attributes']['avail-freq-maps'][0]['freq-map'])
457 freq_map_array = [int(x) for x in freq_map]
458 self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available")
459 self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available")
460 elif ele['tp-id'] == 'SRG1-PP3-TXRX':
461 self.assertNotIn('org-openroadm-network-topology:pp-attributes', dict.keys(ele))
464 def test_27_check_topo_ROADMA_DEG2(self):
465 response = test_utils.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-DEG2', 'config')
466 self.assertEqual(response['status_code'], requests.codes.ok)
467 freq_map = base64.b64decode(
468 response['node']['org-openroadm-network-topology:degree-attributes']['avail-freq-maps'][0]['freq-map'])
469 freq_map_array = [int(x) for x in freq_map]
470 self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
471 self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available")
472 liste_tp = response['node']['ietf-network-topology:termination-point']
474 if ele['tp-id'] == 'DEG2-CTP-TXRX':
475 freq_map = base64.b64decode(
476 ele['org-openroadm-network-topology:ctp-attributes']['avail-freq-maps'][0]['freq-map'])
477 freq_map_array = [int(x) for x in freq_map]
478 self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
479 self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available")
480 elif ele['tp-id'] == 'DEG2-TTP-TXRX':
481 freq_map = base64.b64decode(
482 ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map'])
483 freq_map_array = [int(x) for x in freq_map]
484 self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
485 self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available")
488 # creation service test on a non-available resource
489 def test_28_create_eth_service3(self):
490 self.cr_serv_input_data["service-name"] = "service3"
491 response = test_utils.transportpce_api_rpc_request(
492 'org-openroadm-service', 'service-create',
493 self.cr_serv_input_data)
494 self.assertEqual(response['status_code'], requests.codes.ok)
495 self.assertIn('PCE calculation in progress',
496 response['output']['configuration-response-common']['response-message'])
497 self.assertIn('200', response['output']['configuration-response-common']['response-code'])
498 time.sleep(self.WAITING)
500 # add a test that check the openroadm-service-list still only contains 2 elements
501 def test_29_delete_eth_service3(self):
502 self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service3"
503 response = test_utils.transportpce_api_rpc_request(
504 'org-openroadm-service', 'service-delete',
505 self.del_serv_input_data)
506 self.assertEqual(response['status_code'], requests.codes.ok)
507 self.assertIn('Service \'service3\' does not exist in datastore',
508 response['output']['configuration-response-common']['response-message'])
509 self.assertIn('500', response['output']['configuration-response-common']['response-code'])
512 def test_30_delete_eth_service1(self):
513 self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service1"
514 response = test_utils.transportpce_api_rpc_request(
515 'org-openroadm-service', 'service-delete',
516 self.del_serv_input_data)
517 self.assertEqual(response['status_code'], requests.codes.ok)
518 self.assertIn('Renderer service delete in progress',
519 response['output']['configuration-response-common']['response-message'])
520 time.sleep(self.WAITING)
522 def test_31_delete_eth_service2(self):
523 self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service2"
524 response = test_utils.transportpce_api_rpc_request(
525 'org-openroadm-service', 'service-delete',
526 self.del_serv_input_data)
527 self.assertEqual(response['status_code'], requests.codes.ok)
528 self.assertIn('Renderer service delete in progress',
529 response['output']['configuration-response-common']['response-message'])
530 time.sleep(self.WAITING)
532 def test_32_check_no_xc_ROADMA(self):
533 response = test_utils.check_node_request("ROADM-A1")
534 self.assertEqual(response['status_code'], requests.codes.ok)
535 self.assertNotIn('roadm-connections',
536 dict.keys(response['org-openroadm-device']))
539 def test_33_check_topo_XPDRA(self):
540 response = test_utils.get_ietf_network_node_request('openroadm-topology', 'XPDR-A1-XPDR1', 'config')
541 self.assertEqual(response['status_code'], requests.codes.ok)
542 liste_tp = response['node']['ietf-network-topology:termination-point']
544 if ele['org-openroadm-common-network:tp-type'] == 'XPONDER-CLIENT':
545 self.assertNotIn('org-openroadm-network-topology:xpdr-client-attributes', dict.keys(ele))
546 elif ele['org-openroadm-common-network:tp-type'] == 'XPONDER-NETWORK':
547 self.assertIn('tail-equipment-id',
548 dict.keys(ele['org-openroadm-network-topology:xpdr-network-attributes']))
549 self.assertNotIn('wavelength',
550 dict.keys(ele['org-openroadm-network-topology:xpdr-network-attributes']))
553 def test_34_check_topo_ROADMA_SRG1(self):
554 response = test_utils.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-SRG1', 'config')
555 self.assertEqual(response['status_code'], requests.codes.ok)
556 freq_map = base64.b64decode(
557 response['node']['org-openroadm-network-topology:srg-attributes']['avail-freq-maps'][0]['freq-map'])
558 freq_map_array = [int(x) for x in freq_map]
559 self.assertEqual(freq_map_array[95], 255, "Index 1 should be available")
560 self.assertEqual(freq_map_array[94], 255, "Index 2 should be available")
561 liste_tp = response['node']['ietf-network-topology:termination-point']
563 if ele['tp-id'] in ('SRG1-PP1-TXRX', 'SRG1-PP2-TXRX'):
564 freq_map = base64.b64decode(
565 ele['org-openroadm-network-topology:pp-attributes']['avail-freq-maps'][0]['freq-map'])
566 freq_map_array = [int(x) for x in freq_map]
567 self.assertEqual(freq_map_array[95], 255, "Index 1 should be available")
568 self.assertEqual(freq_map_array[94], 255, "Index 2 should be available")
569 elif ele['tp-id'] == 'SRG1-CP-TXRX':
570 freq_map = base64.b64decode(
571 ele['org-openroadm-network-topology:cp-attributes']['avail-freq-maps'][0]['freq-map'])
572 freq_map_array = [int(x) for x in freq_map]
573 self.assertEqual(freq_map_array[95], 255, "Index 1 should be available")
574 self.assertEqual(freq_map_array[94], 255, "Index 2 should be available")
576 self.assertNotIn('org-openroadm-network-topology:pp-attributes', dict.keys(ele))
579 def test_35_check_topo_ROADMA_DEG2(self):
580 response = test_utils.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-DEG2', 'config')
581 self.assertEqual(response['status_code'], requests.codes.ok)
582 freq_map = base64.b64decode(
583 response['node']['org-openroadm-network-topology:degree-attributes']['avail-freq-maps'][0]['freq-map'])
584 freq_map_array = [int(x) for x in freq_map]
585 self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available")
586 self.assertEqual(freq_map_array[94], 255, "Lambda 2 should be available")
587 liste_tp = response['node']['ietf-network-topology:termination-point']
589 if ele['tp-id'] == 'DEG2-CTP-TXRX':
590 freq_map = base64.b64decode(
591 ele['org-openroadm-network-topology:ctp-attributes']['avail-freq-maps'][0]['freq-map'])
592 freq_map_array = [int(x) for x in freq_map]
593 self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available")
594 self.assertEqual(freq_map_array[94], 255, "Lambda 2 should be available")
595 elif ele['tp-id'] == 'DEG2-TTP-TXRX':
596 freq_map = base64.b64decode(
597 ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map'])
598 freq_map_array = [int(x) for x in freq_map]
599 self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available")
600 self.assertEqual(freq_map_array[94], 255, "Lambda 2 should be available")
603 # test service-create for Optical Channel (OC) service from srg-pp to srg-pp
604 def test_36_create_oc_service1(self):
605 self.cr_serv_input_data["service-name"] = "service1"
606 self.cr_serv_input_data["connection-type"] = "roadm-line"
607 self.cr_serv_input_data["service-a-end"]["node-id"] = "ROADM-A1"
608 self.cr_serv_input_data["service-a-end"]["service-format"] = "OC"
609 self.cr_serv_input_data["service-z-end"]["node-id"] = "ROADM-C1"
610 self.cr_serv_input_data["service-z-end"]["service-format"] = "OC"
611 response = test_utils.transportpce_api_rpc_request(
612 'org-openroadm-service', 'service-create',
613 self.cr_serv_input_data)
614 self.assertEqual(response['status_code'], requests.codes.ok)
615 self.assertIn('PCE calculation in progress',
616 response['output']['configuration-response-common']['response-message'])
617 time.sleep(self.WAITING)
619 def test_37_get_oc_service1(self):
620 response = test_utils.get_ordm_serv_list_attr_request("services", "service1")
621 self.assertEqual(response['status_code'], requests.codes.ok)
622 self.assertEqual(response['services'][0]['administrative-state'], 'inService')
623 self.assertEqual(response['services'][0]['service-name'], 'service1')
624 self.assertEqual(response['services'][0]['connection-type'], 'roadm-line')
625 self.assertEqual(response['services'][0]['lifecycle-state'], 'planned')
628 def test_38_check_xc1_ROADMA(self):
629 response = test_utils.check_node_attribute_request(
630 "ROADM-A1", "roadm-connections", "SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
631 self.assertEqual(response['status_code'], requests.codes.ok)
632 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
633 self.assertDictEqual(
635 'connection-name': 'SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768',
636 'opticalControlMode': 'gainLoss',
637 'target-output-power': -3.0
638 }, **response['roadm-connections'][0]), response['roadm-connections'][0])
639 self.assertDictEqual({'src-if': 'SRG1-PP1-TXRX-nmc-761:768'}, response['roadm-connections'][0]['source'])
640 self.assertDictEqual({'dst-if': 'DEG2-TTP-TXRX-nmc-761:768'}, response['roadm-connections'][0]['destination'])
643 def test_39_check_xc1_ROADMC(self):
644 response = test_utils.check_node_attribute_request(
645 "ROADM-C1", "roadm-connections", "SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768")
646 self.assertEqual(response['status_code'], requests.codes.ok)
647 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
648 self.assertDictEqual(
650 'connection-name': 'SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768',
651 'opticalControlMode': 'gainLoss',
652 'target-output-power': -3.0
653 }, **response['roadm-connections'][0]), response['roadm-connections'][0])
654 self.assertDictEqual({'src-if': 'SRG1-PP1-TXRX-nmc-761:768'}, response['roadm-connections'][0]['source'])
655 self.assertDictEqual({'dst-if': 'DEG1-TTP-TXRX-nmc-761:768'}, response['roadm-connections'][0]['destination'])
658 def test_40_create_oc_service2(self):
659 self.cr_serv_input_data["service-name"] = "service2"
660 self.cr_serv_input_data["connection-type"] = "roadm-line"
661 self.cr_serv_input_data["service-a-end"]["node-id"] = "ROADM-A1"
662 self.cr_serv_input_data["service-a-end"]["service-format"] = "OC"
663 self.cr_serv_input_data["service-z-end"]["node-id"] = "ROADM-C1"
664 self.cr_serv_input_data["service-z-end"]["service-format"] = "OC"
665 response = test_utils.transportpce_api_rpc_request(
666 'org-openroadm-service', 'service-create',
667 self.cr_serv_input_data)
668 self.assertEqual(response['status_code'], requests.codes.ok)
669 self.assertIn('PCE calculation in progress',
670 response['output']['configuration-response-common']['response-message'])
671 time.sleep(self.WAITING)
673 def test_41_get_oc_service2(self):
674 response = test_utils.get_ordm_serv_list_attr_request("services", "service2")
675 self.assertEqual(response['status_code'], requests.codes.ok)
676 self.assertEqual(response['services'][0]['administrative-state'], 'inService')
677 self.assertEqual(response['services'][0]['service-name'], 'service2')
678 self.assertEqual(response['services'][0]['connection-type'], 'roadm-line')
679 self.assertEqual(response['services'][0]['lifecycle-state'], 'planned')
682 def test_42_check_xc2_ROADMA(self):
683 response = test_utils.check_node_attribute_request(
684 "ROADM-A1", "roadm-connections", "SRG1-PP2-TXRX-DEG2-TTP-TXRX-753:760")
685 self.assertEqual(response['status_code'], requests.codes.ok)
686 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
687 self.assertDictEqual(
689 'connection-name': 'SRG1-PP2-TXRX-DEG2-TTP-TXRX-753:760',
690 'opticalControlMode': 'gainLoss',
691 'target-output-power': -3.0
692 }, **response['roadm-connections'][0]), response['roadm-connections'][0])
693 self.assertDictEqual({'src-if': 'SRG1-PP2-TXRX-nmc-753:760'}, response['roadm-connections'][0]['source'])
694 self.assertDictEqual({'dst-if': 'DEG2-TTP-TXRX-nmc-753:760'}, response['roadm-connections'][0]['destination'])
697 def test_43_check_topo_ROADMA(self):
698 response = test_utils.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-SRG1', 'config')
699 self.assertEqual(response['status_code'], requests.codes.ok)
700 freq_map = base64.b64decode(
701 response['node']['org-openroadm-network-topology:srg-attributes']['avail-freq-maps'][0]['freq-map'])
702 freq_map_array = [int(x) for x in freq_map]
703 self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
704 self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available")
705 liste_tp = response['node']['ietf-network-topology:termination-point']
707 if ele['tp-id'] == 'SRG1-PP1-TXRX':
708 freq_map = base64.b64decode(
709 ele['org-openroadm-network-topology:pp-attributes']['avail-freq-maps'][0]['freq-map'])
710 freq_map_array = [int(x) for x in freq_map]
711 self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
712 self.assertEqual(freq_map_array[94], 255, "Lambda 2 should be available")
713 elif ele['tp-id'] == 'SRG1-PP2-TXRX':
714 freq_map = base64.b64decode(
715 ele['org-openroadm-network-topology:pp-attributes']['avail-freq-maps'][0]['freq-map'])
716 freq_map_array = [int(x) for x in freq_map]
717 self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available")
718 self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available")
719 elif ele['tp-id'] == 'SRG1-PP3-TXRX':
720 self.assertNotIn('org-openroadm-network-topology:pp-attributes', dict.keys(ele))
722 self.test_27_check_topo_ROADMA_DEG2()
725 def test_44_delete_oc_service1(self):
726 self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service1"
727 response = test_utils.transportpce_api_rpc_request(
728 'org-openroadm-service', 'service-delete',
729 self.del_serv_input_data)
730 self.assertEqual(response['status_code'], requests.codes.ok)
731 self.assertIn('Renderer service delete in progress',
732 response['output']['configuration-response-common']['response-message'])
733 time.sleep(self.WAITING)
735 def test_45_delete_oc_service2(self):
736 self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service2"
737 response = test_utils.transportpce_api_rpc_request(
738 'org-openroadm-service', 'service-delete',
739 self.del_serv_input_data)
740 self.assertEqual(response['status_code'], requests.codes.ok)
741 self.assertIn('Renderer service delete in progress',
742 response['output']['configuration-response-common']['response-message'])
743 time.sleep(self.WAITING)
745 def test_46_get_no_oc_services(self):
746 response = test_utils.get_ordm_serv_list_request()
747 self.assertEqual(response['status_code'], requests.codes.conflict)
748 self.assertIn(response['service-list'], (
750 "error-type": "protocol",
751 "error-tag": "data-missing",
753 "Request could not be completed because the relevant data "
754 "model content does not exist"
756 "error-type": "application",
757 "error-tag": "data-missing",
759 "Request could not be completed because the relevant data "
760 "model content does not exist"
764 def test_47_get_no_xc_ROADMA(self):
765 response = test_utils.check_node_request("ROADM-A1")
766 self.assertEqual(response['status_code'], requests.codes.ok)
767 self.assertNotIn('roadm-connections', dict.keys(response['org-openroadm-device']))
770 def test_48_check_topo_ROADMA(self):
771 self.test_34_check_topo_ROADMA_SRG1()
772 self.test_35_check_topo_ROADMA_DEG2()
774 def test_49_loop_create_oc_service(self):
775 for i in range(1, 3):
776 # pylint: disable=consider-using-f-string
777 print("iteration number {}".format(i))
778 print("oc service creation")
779 self.test_36_create_oc_service1()
780 print("check xc in ROADM-A1")
781 self.test_38_check_xc1_ROADMA()
782 print("check xc in ROADM-C1")
783 self.test_39_check_xc1_ROADMC()
784 print("oc service deletion\n")
785 self.test_44_delete_oc_service1()
787 def test_50_loop_create_eth_service(self):
788 response = test_utils.get_ordm_serv_list_attr_request("services", "service1")
789 if response['status_code'] != 404:
790 self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service1"
791 response = test_utils.transportpce_api_rpc_request(
792 'org-openroadm-service', 'service-delete',
793 self.del_serv_input_data)
795 self.cr_serv_input_data["connection-type"] = "service"
796 self.cr_serv_input_data["service-a-end"]["node-id"] = "XPDR-A1"
797 self.cr_serv_input_data["service-a-end"]["service-format"] = "Ethernet"
798 self.cr_serv_input_data["service-z-end"]["node-id"] = "XPDR-C1"
799 self.cr_serv_input_data["service-z-end"]["service-format"] = "Ethernet"
800 for i in range(1, 3):
801 # pylint: disable=consider-using-f-string
802 print("iteration number {}".format(i))
803 print("eth service creation")
804 self.test_15_create_eth_service2()
805 print("check xc in ROADM-A1")
806 self.test_17_check_xc1_ROADMA()
807 print("eth service deletion\n")
808 self.test_31_delete_eth_service2()
810 def test_51_disconnect_XPDRA(self):
811 response = test_utils.unmount_device("XPDR-A1")
812 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
814 def test_52_disconnect_XPDRC(self):
815 response = test_utils.unmount_device("XPDR-C1")
816 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
818 def test_53_disconnect_ROADMA(self):
819 response = test_utils.unmount_device("ROADM-A1")
820 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
822 def test_54_disconnect_ROADMC(self):
823 response = test_utils.unmount_device("ROADM-C1")
824 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
827 if __name__ == "__main__":
828 unittest.main(verbosity=2)