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 ##############################################################################
17 from common import test_utils
20 class TransportPCEFulltesting(unittest.TestCase):
22 WAITING = 20 # nominal value is 300
28 cls.processes = test_utils.start_tpce()
29 cls.processes = test_utils.start_sims(['xpdra', 'roadma-full', 'roadmc-full', 'xpdrc'])
32 def tearDownClass(cls):
33 for process in cls.processes:
34 test_utils.shutdown_process(process)
35 print("all processes killed")
37 def setUp(self): # instruction executed before each test method
38 print("execution of {}".format(self.id().split(".")[-1]))
40 # connect netconf devices
41 def test_01_connect_xpdrA(self):
42 response = test_utils.mount_device("XPDRA01", 'xpdra')
43 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
45 def test_02_connect_xpdrC(self):
46 response = test_utils.mount_device("XPDRC01", 'xpdrc')
47 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
49 def test_03_connect_rdmA(self):
50 response = test_utils.mount_device("ROADMA01", 'roadma-full')
51 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
53 def test_04_connect_rdmC(self):
54 response = test_utils.mount_device("ROADMC01", 'roadmc-full')
55 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
57 def test_05_connect_xprdA_N1_to_roadmA_PP1(self):
58 response = test_utils.connect_xpdr_to_rdm_request("XPDRA01", "1", "1",
59 "ROADMA01", "1", "SRG1-PP1-TXRX")
60 self.assertEqual(response.status_code, requests.codes.ok)
62 self.assertIn('Xponder Roadm Link created successfully',
63 res["output"]["result"])
66 def test_06_connect_roadmA_PP1_to_xpdrA_N1(self):
67 url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links"
69 "networkutils:input": {
70 "networkutils:links-input": {
71 "networkutils:xpdr-node": "XPDRA01",
72 "networkutils:xpdr-num": "1",
73 "networkutils:network-num": "1",
74 "networkutils:rdm-node": "ROADMA01",
75 "networkutils:srg-num": "1",
76 "networkutils:termination-point-num": "SRG1-PP1-TXRX"
80 response = test_utils.post_request(url, data)
81 response = test_utils.connect_rdm_to_xpdr_request("XPDRA01", "1", "1",
82 "ROADMA01", "1", "SRG1-PP1-TXRX")
83 self.assertEqual(response.status_code, requests.codes.ok)
85 self.assertIn('Roadm Xponder links created successfully',
86 res["output"]["result"])
89 def test_07_connect_xprdC_N1_to_roadmC_PP1(self):
90 response = test_utils.connect_xpdr_to_rdm_request("XPDRC01", "1", "1",
91 "ROADMC01", "1", "SRG1-PP1-TXRX")
92 self.assertEqual(response.status_code, requests.codes.ok)
94 self.assertIn('Xponder Roadm Link created successfully',
95 res["output"]["result"])
98 def test_08_connect_roadmC_PP1_to_xpdrC_N1(self):
99 response = test_utils.connect_rdm_to_xpdr_request("XPDRC01", "1", "1",
100 "ROADMC01", "1", "SRG1-PP1-TXRX")
101 self.assertEqual(response.status_code, requests.codes.ok)
102 res = response.json()
103 self.assertIn('Roadm Xponder links created successfully',
104 res["output"]["result"])
107 def test_09_add_omsAttributes_ROADMA_ROADMC(self):
108 # Config ROADMA-ROADMC oms-attributes
111 "auto-spanloss": "true",
112 "spanloss-base": 11.4,
113 "spanloss-current": 12,
114 "engineered-spanloss": 12.2,
115 "link-concatenation": [{
118 "SRLG-length": 100000,
120 response = test_utils.add_oms_attr_request("ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX", data)
121 self.assertEqual(response.status_code, requests.codes.created)
123 def test_10_add_omsAttributes_ROADMC_ROADMA(self):
124 # Config ROADMC-ROADMA oms-attributes
127 "auto-spanloss": "true",
128 "spanloss-base": 11.4,
129 "spanloss-current": 12,
130 "engineered-spanloss": 12.2,
131 "link-concatenation": [{
134 "SRLG-length": 100000,
136 response = test_utils.add_oms_attr_request("ROADMC01-DEG2-DEG2-TTP-TXRXtoROADMA01-DEG1-DEG1-TTP-TXRX", data)
137 self.assertEqual(response.status_code, requests.codes.created)
139 # test service-create for Eth service from xpdr to xpdr
140 def test_11_create_eth_service1(self):
141 url = "{}/operations/org-openroadm-service:service-create"
144 "sdnc-request-header": {
145 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
146 "rpc-action": "service-create",
147 "request-system-id": "appname",
149 "http://localhost:8585/NotificationServer/notify"
151 "service-name": "service1",
152 "common-id": "ASATT1234567",
153 "connection-type": "service",
155 "service-rate": "100",
156 "node-id": "XPDRA01",
157 "service-format": "Ethernet",
158 "clli": "SNJSCAMCJP8",
162 "ROUTER_SNJSCAMCJP8_000000.00_00",
163 "port-type": "router",
164 "port-name": "Gigabit Ethernet_Tx.ge-5/0/0.0",
165 "port-rack": "000000.00",
170 "LGX Panel_SNJSCAMCJP8_000000.00_00",
171 "lgx-port-name": "LGX Back.3",
172 "lgx-port-rack": "000000.00",
173 "lgx-port-shelf": "00"
179 "ROUTER_SNJSCAMCJP8_000000.00_00",
180 "port-type": "router",
181 "port-name": "Gigabit Ethernet_Rx.ge-5/0/0.0",
182 "port-rack": "000000.00",
187 "LGX Panel_SNJSCAMCJP8_000000.00_00",
188 "lgx-port-name": "LGX Back.4",
189 "lgx-port-rack": "000000.00",
190 "lgx-port-shelf": "00"
196 "service-rate": "100",
197 "node-id": "XPDRC01",
198 "service-format": "Ethernet",
199 "clli": "SNJSCAMCJT4",
203 "ROUTER_SNJSCAMCJT4_000000.00_00",
204 "port-type": "router",
205 "port-name": "Gigabit Ethernet_Tx.ge-1/0/0.0",
206 "port-rack": "000000.00",
211 "LGX Panel_SNJSCAMCJT4_000000.00_00",
212 "lgx-port-name": "LGX Back.29",
213 "lgx-port-rack": "000000.00",
214 "lgx-port-shelf": "00"
220 "ROUTER_SNJSCAMCJT4_000000.00_00",
221 "port-type": "router",
222 "port-name": "Gigabit Ethernet_Rx.ge-1/0/0.0",
223 "port-rack": "000000.00",
228 "LGX Panel_SNJSCAMCJT4_000000.00_00",
229 "lgx-port-name": "LGX Back.30",
230 "lgx-port-rack": "000000.00",
231 "lgx-port-shelf": "00"
236 "due-date": "2016-11-28T00:00:01Z",
237 "operator-contact": "pw1234"
240 response = test_utils.post_request(url, data)
241 self.assertEqual(response.status_code, requests.codes.ok)
242 res = response.json()
243 self.assertIn('PCE calculation in progress',
244 res['output']['configuration-response-common'][
246 time.sleep(self.WAITING)
248 def test_12_get_eth_service1(self):
249 url = "{}/operational/org-openroadm-service:service-list/services/service1"
250 response = test_utils.get_request(url)
251 self.assertEqual(response.status_code, requests.codes.ok)
252 res = response.json()
254 res['services'][0]['administrative-state'],
257 res['services'][0]['service-name'], 'service1')
259 res['services'][0]['connection-type'], 'service')
261 res['services'][0]['lifecycle-state'], 'planned')
264 def test_13_check_xc1_ROADMA(self):
265 response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-1")
266 self.assertEqual(response.status_code, requests.codes.ok)
267 res = response.json()
268 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
269 self.assertDictEqual(
271 'connection-number': 'SRG1-PP1-TXRX-DEG1-TTP-TXRX-1',
272 'wavelength-number': 1,
273 'opticalControlMode': 'gainLoss',
274 'target-output-power': -3.0
275 }, **res['roadm-connections'][0]),
276 res['roadm-connections'][0]
278 self.assertDictEqual(
279 {'src-if': 'SRG1-PP1-TXRX-1'},
280 res['roadm-connections'][0]['source'])
281 self.assertDictEqual(
282 {'dst-if': 'DEG1-TTP-TXRX-1'},
283 res['roadm-connections'][0]['destination'])
286 def test_14_check_xc1_ROADMC(self):
287 response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-1")
288 self.assertEqual(response.status_code, requests.codes.ok)
289 res = response.json()
290 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
291 self.assertDictEqual(
293 'connection-number': 'SRG1-PP1-TXRX-DEG2-TTP-TXRX-1',
294 'wavelength-number': 1,
295 'opticalControlMode': 'gainLoss',
296 'target-output-power': 2.0
297 }, **res['roadm-connections'][0]),
298 res['roadm-connections'][0]
300 self.assertDictEqual(
301 {'src-if': 'SRG1-PP1-TXRX-1'},
302 res['roadm-connections'][0]['source'])
303 self.assertDictEqual(
304 {'dst-if': 'DEG2-TTP-TXRX-1'},
305 res['roadm-connections'][0]['destination'])
308 def test_15_check_topo_XPDRA(self):
309 response = test_utils.get_ordm_topo_request("node/XPDRA01-XPDR1")
310 self.assertEqual(response.status_code, requests.codes.ok)
311 res = response.json()
312 liste_tp = res['node'][0]['ietf-network-topology:termination-point']
314 if ele['tp-id'] == 'XPDR1-NETWORK1':
315 self.assertEqual({u'frequency': 196.1, u'width': 40},
316 ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength'])
317 if ele['tp-id'] == 'XPDR1-CLIENT1' or ele['tp-id'] == 'XPDR1-CLIENT3':
319 'org-openroadm-network-topology:xpdr-client-attributes',
321 if ele['tp-id'] == 'XPDR1-NETWORK2':
323 'org-openroadm-network-topology:xpdr-network-attributes',
327 def test_16_check_topo_ROADMA_SRG1(self):
328 response = test_utils.get_ordm_topo_request("node/ROADMA01-SRG1")
329 self.assertEqual(response.status_code, requests.codes.ok)
330 res = response.json()
331 self.assertNotIn({u'index': 1},
333 u'org-openroadm-network-topology:srg-attributes'][
334 'available-wavelengths'])
335 liste_tp = res['node'][0]['ietf-network-topology:termination-point']
337 if ele['tp-id'] == 'SRG1-PP1-TXRX':
338 self.assertIn({u'index': 1, u'frequency': 196.1, u'width': 40},
339 ele['org-openroadm-network-topology:'
340 'pp-attributes']['used-wavelength']
342 if ele['tp-id'] == 'SRG1-PP2-TXRX':
343 self.assertNotIn('used-wavelength', dict.keys(ele))
346 def test_17_check_topo_ROADMA_DEG1(self):
347 response = test_utils.get_ordm_topo_request("node/ROADMA01-DEG1")
348 self.assertEqual(response.status_code, requests.codes.ok)
349 res = response.json()
350 self.assertNotIn({u'index': 1},
352 u'org-openroadm-network-topology:'
353 u'degree-attributes'][
354 'available-wavelengths'])
355 liste_tp = res['node'][0]['ietf-network-topology:termination-point']
357 if ele['tp-id'] == 'DEG1-CTP-TXRX':
358 self.assertIn({u'index': 1, u'frequency': 196.1, u'width': 40},
359 ele['org-openroadm-network-topology:'
362 if ele['tp-id'] == 'DEG1-TTP-TXRX':
363 self.assertIn({u'index': 1, u'frequency': 196.1, u'width': 40},
364 ele['org-openroadm-network-topology:'
365 'tx-ttp-attributes'][
369 def test_18_connect_xprdA_N2_to_roadmA_PP2(self):
370 response = test_utils.connect_xpdr_to_rdm_request("XPDRA01", "1", "2",
371 "ROADMA01", "1", "SRG1-PP2-TXRX")
372 self.assertEqual(response.status_code, requests.codes.ok)
373 res = response.json()
374 self.assertIn('Xponder Roadm Link created successfully',
375 res["output"]["result"])
378 def test_19_connect_roadmA_PP2_to_xpdrA_N2(self):
379 response = test_utils.connect_rdm_to_xpdr_request("XPDRA01", "1", "2",
380 "ROADMA01", "1", "SRG1-PP2-TXRX")
381 self.assertEqual(response.status_code, requests.codes.ok)
382 res = response.json()
383 self.assertIn('Roadm Xponder links created successfully',
384 res["output"]["result"])
387 def test_20_connect_xprdC_N2_to_roadmC_PP2(self):
388 response = test_utils.connect_xpdr_to_rdm_request("XPDRC01", "1", "2",
389 "ROADMC01", "1", "SRG1-PP2-TXRX")
390 self.assertEqual(response.status_code, requests.codes.ok)
391 res = response.json()
392 self.assertIn('Xponder Roadm Link created successfully',
393 res["output"]["result"])
396 def test_21_connect_roadmC_PP2_to_xpdrC_N2(self):
397 response = test_utils.connect_rdm_to_xpdr_request("XPDRC01", "1", "2",
398 "ROADMC01", "1", "SRG1-PP2-TXRX")
399 self.assertEqual(response.status_code, requests.codes.ok)
400 res = response.json()
401 self.assertIn('Roadm Xponder links created successfully',
402 res["output"]["result"])
405 def test_22_create_eth_service2(self):
406 url = "{}/operations/org-openroadm-service:service-create"
409 "sdnc-request-header": {
410 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
411 "rpc-action": "service-create",
412 "request-system-id": "appname",
414 "http://localhost:8585/NotificationServer/notify"
416 "service-name": "service2",
417 "common-id": "ASATT1234567",
418 "connection-type": "service",
420 "service-rate": "100",
421 "node-id": "XPDRA01",
422 "service-format": "Ethernet",
423 "clli": "SNJSCAMCJP8",
427 "ROUTER_SNJSCAMCJP8_000000.00_00",
428 "port-type": "router",
429 "port-name": "Gigabit Ethernet_Tx.ge-5/0/0.0",
430 "port-rack": "000000.00",
435 "LGX Panel_SNJSCAMCJP8_000000.00_00",
436 "lgx-port-name": "LGX Back.3",
437 "lgx-port-rack": "000000.00",
438 "lgx-port-shelf": "00"
444 "ROUTER_SNJSCAMCJP8_000000.00_00",
445 "port-type": "router",
446 "port-name": "Gigabit Ethernet_Rx.ge-5/0/0.0",
447 "port-rack": "000000.00",
452 "LGX Panel_SNJSCAMCJP8_000000.00_00",
453 "lgx-port-name": "LGX Back.4",
454 "lgx-port-rack": "000000.00",
455 "lgx-port-shelf": "00"
461 "service-rate": "100",
462 "node-id": "XPDRC01",
463 "service-format": "Ethernet",
464 "clli": "SNJSCAMCJT4",
468 "ROUTER_SNJSCAMCJT4_000000.00_00",
469 "port-type": "router",
470 "port-name": "Gigabit Ethernet_Tx.ge-1/0/0.0",
471 "port-rack": "000000.00",
476 "LGX Panel_SNJSCAMCJT4_000000.00_00",
477 "lgx-port-name": "LGX Back.29",
478 "lgx-port-rack": "000000.00",
479 "lgx-port-shelf": "00"
485 "ROUTER_SNJSCAMCJT4_000000.00_00",
486 "port-type": "router",
487 "port-name": "Gigabit Ethernet_Rx.ge-1/0/0.0",
488 "port-rack": "000000.00",
493 "LGX Panel_SNJSCAMCJT4_000000.00_00",
494 "lgx-port-name": "LGX Back.30",
495 "lgx-port-rack": "000000.00",
496 "lgx-port-shelf": "00"
501 "due-date": "2016-11-28T00:00:01Z",
502 "operator-contact": "pw1234"
505 response = test_utils.post_request(url, data)
506 self.assertEqual(response.status_code, requests.codes.ok)
507 res = response.json()
508 self.assertIn('PCE calculation in progress',
509 res['output']['configuration-response-common'][
511 time.sleep(self.WAITING)
513 def test_23_get_eth_service2(self):
514 url = "{}/operational/org-openroadm-service:service-list/services/service2"
515 response = test_utils.get_request(url)
516 self.assertEqual(response.status_code, requests.codes.ok)
517 res = response.json()
519 res['services'][0]['administrative-state'],
522 res['services'][0]['service-name'], 'service2')
524 res['services'][0]['connection-type'], 'service')
526 res['services'][0]['lifecycle-state'], 'planned')
529 def test_24_check_xc2_ROADMA(self):
530 response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/DEG1-TTP-TXRX-SRG1-PP2-TXRX-2")
531 self.assertEqual(response.status_code, requests.codes.ok)
532 res = response.json()
533 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
534 self.assertDictEqual(
536 'connection-number': 'DEG1-TTP-TXRX-SRG1-PP2-TXRX-2',
537 'wavelength-number': 2,
538 'opticalControlMode': 'power'
539 }, **res['roadm-connections'][0]),
540 res['roadm-connections'][0]
542 self.assertDictEqual(
543 {'src-if': 'DEG1-TTP-TXRX-2'},
544 res['roadm-connections'][0]['source'])
545 self.assertDictEqual(
546 {'dst-if': 'SRG1-PP2-TXRX-2'},
547 res['roadm-connections'][0]['destination'])
549 def test_25_check_topo_XPDRA(self):
550 response = test_utils.get_ordm_topo_request("node/XPDRA01-XPDR1")
551 self.assertEqual(response.status_code, requests.codes.ok)
552 res = response.json()
553 liste_tp = res['node'][0]['ietf-network-topology:termination-point']
555 if ele['tp-id'] == 'XPDR1-NETWORK1':
556 self.assertEqual({u'frequency': 196.1, u'width': 40},
557 ele['org-openroadm-network-topology:'
558 'xpdr-network-attributes'][
560 if ele['tp-id'] == 'XPDR1-NETWORK2':
561 self.assertEqual({u'frequency': 196.05, u'width': 40},
562 ele['org-openroadm-network-topology:'
563 'xpdr-network-attributes'][
565 if ele['tp-id'] == 'XPDR1-CLIENT1' or \
566 ele['tp-id'] == 'XPDR1-CLIENT3':
568 'org-openroadm-network-topology:xpdr-client-attributes',
572 def test_26_check_topo_ROADMA_SRG1(self):
573 response = test_utils.get_ordm_topo_request("node/ROADMA01-SRG1")
574 self.assertEqual(response.status_code, requests.codes.ok)
575 res = response.json()
576 self.assertNotIn({u'index': 1}, res['node'][0][
577 u'org-openroadm-network-topology:srg-attributes'][
578 'available-wavelengths'])
579 self.assertNotIn({u'index': 2}, res['node'][0][
580 u'org-openroadm-network-topology:srg-attributes'][
581 'available-wavelengths'])
582 liste_tp = res['node'][0]['ietf-network-topology:termination-point']
584 if ele['tp-id'] == 'SRG1-PP1-TXRX':
585 self.assertIn({u'index': 1, u'frequency': 196.1, u'width': 40},
586 ele['org-openroadm-network-topology:'
587 'pp-attributes']['used-wavelength'])
588 self.assertNotIn({u'index': 2, u'frequency': 196.05,
590 ele['org-openroadm-network-topology:'
591 'pp-attributes']['used-wavelength'])
592 if ele['tp-id'] == 'SRG1-PP2-TXRX':
593 self.assertIn({u'index': 2, u'frequency': 196.05, u'width': 40},
594 ele['org-openroadm-network-topology:'
595 'pp-attributes']['used-wavelength'])
596 self.assertNotIn({u'index': 1, u'frequency': 196.1,
598 ele['org-openroadm-network-topology:'
599 'pp-attributes']['used-wavelength'])
600 if ele['tp-id'] == 'SRG1-PP3-TXRX':
601 self.assertNotIn('org-openroadm-network-topology:pp-attributes',
605 def test_27_check_topo_ROADMA_DEG1(self):
606 response = test_utils.get_ordm_topo_request("node/ROADMA01-DEG1")
607 self.assertEqual(response.status_code, requests.codes.ok)
608 res = response.json()
609 self.assertNotIn({u'index': 1}, res['node'][0][
610 u'org-openroadm-network-topology:degree-attributes'][
611 'available-wavelengths'])
612 self.assertNotIn({u'index': 2}, res['node'][0][
613 u'org-openroadm-network-topology:degree-attributes'][
614 'available-wavelengths'])
615 liste_tp = res['node'][0]['ietf-network-topology:termination-point']
617 if ele['tp-id'] == 'DEG1-CTP-TXRX':
618 self.assertIn({u'index': 1, u'frequency': 196.1, u'width': 40},
619 ele['org-openroadm-network-topology:'
620 'ctp-attributes']['used-wavelengths'])
621 self.assertIn({u'index': 2, u'frequency': 196.05, u'width': 40},
622 ele['org-openroadm-network-topology:'
623 'ctp-attributes']['used-wavelengths'])
624 if ele['tp-id'] == 'DEG1-TTP-TXRX':
625 self.assertIn({u'index': 1, u'frequency': 196.1, u'width': 40},
626 ele['org-openroadm-network-topology:'
627 'tx-ttp-attributes']['used-wavelengths'])
628 self.assertIn({u'index': 2, u'frequency': 196.05, u'width': 40},
629 ele['org-openroadm-network-topology:'
630 'tx-ttp-attributes']['used-wavelengths'])
633 # creation service test on a non-available resource
634 def test_28_create_eth_service3(self):
635 url = "{}/operations/org-openroadm-service:service-create"
638 "sdnc-request-header": {
639 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
640 "rpc-action": "service-create",
641 "request-system-id": "appname",
643 "http://localhost:8585/NotificationServer/notify"
645 "service-name": "service3",
646 "common-id": "ASATT1234567",
647 "connection-type": "service",
649 "service-rate": "100",
650 "node-id": "XPDRA01",
651 "service-format": "Ethernet",
652 "clli": "SNJSCAMCJP8",
656 "ROUTER_SNJSCAMCJP8_000000.00_00",
657 "port-type": "router",
658 "port-name": "Gigabit Ethernet_Tx.ge-5/0/0.0",
659 "port-rack": "000000.00",
664 "LGX Panel_SNJSCAMCJP8_000000.00_00",
665 "lgx-port-name": "LGX Back.3",
666 "lgx-port-rack": "000000.00",
667 "lgx-port-shelf": "00"
673 "ROUTER_SNJSCAMCJP8_000000.00_00",
674 "port-type": "router",
675 "port-name": "Gigabit Ethernet_Rx.ge-5/0/0.0",
676 "port-rack": "000000.00",
681 "LGX Panel_SNJSCAMCJP8_000000.00_00",
682 "lgx-port-name": "LGX Back.4",
683 "lgx-port-rack": "000000.00",
684 "lgx-port-shelf": "00"
690 "service-rate": "100",
691 "node-id": "XPDRC01",
692 "service-format": "Ethernet",
693 "clli": "SNJSCAMCJT4",
697 "ROUTER_SNJSCAMCJT4_000000.00_00",
698 "port-type": "router",
699 "port-name": "Gigabit Ethernet_Tx.ge-1/0/0.0",
700 "port-rack": "000000.00",
705 "LGX Panel_SNJSCAMCJT4_000000.00_00",
706 "lgx-port-name": "LGX Back.29",
707 "lgx-port-rack": "000000.00",
708 "lgx-port-shelf": "00"
714 "ROUTER_SNJSCAMCJT4_000000.00_00",
715 "port-type": "router",
716 "port-name": "Gigabit Ethernet_Rx.ge-1/0/0.0",
717 "port-rack": "000000.00",
722 "LGX Panel_SNJSCAMCJT4_000000.00_00",
723 "lgx-port-name": "LGX Back.30",
724 "lgx-port-rack": "000000.00",
725 "lgx-port-shelf": "00"
730 "due-date": "2016-11-28T00:00:01Z",
731 "operator-contact": "pw1234"
734 response = test_utils.post_request(url, data)
735 self.assertEqual(response.status_code, requests.codes.ok)
736 res = response.json()
737 self.assertIn('PCE calculation in progress',
738 res['output']['configuration-response-common'][
740 self.assertIn('200', res['output']['configuration-response-common'][
742 time.sleep(self.WAITING)
744 # add a test that check the openroadm-service-list still only
745 # contains 2 elements
747 def test_29_delete_eth_service3(self):
748 url = "{}/operations/org-openroadm-service:service-delete"
750 "sdnc-request-header": {
751 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
752 "rpc-action": "service-delete",
753 "request-system-id": "appname",
755 "http://localhost:8585/NotificationServer/notify"
757 "service-delete-req-info": {
758 "service-name": "service3",
759 "tail-retention": "no"
763 response = test_utils.post_request(url, data)
764 self.assertEqual(response.status_code, requests.codes.ok)
765 res = response.json()
766 self.assertIn('Service \'service3\' does not exist in datastore',
767 res['output']['configuration-response-common'][
769 self.assertIn('500', res['output']['configuration-response-common'][
773 def test_30_delete_eth_service1(self):
774 url = "{}/operations/org-openroadm-service:service-delete"
776 "sdnc-request-header": {
777 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
778 "rpc-action": "service-delete",
779 "request-system-id": "appname",
781 "http://localhost:8585/NotificationServer/notify"
783 "service-delete-req-info": {
784 "service-name": "service1",
785 "tail-retention": "no"
789 response = test_utils.post_request(url, data)
790 self.assertEqual(response.status_code, requests.codes.ok)
791 res = response.json()
792 self.assertIn('Renderer service delete in progress',
793 res['output']['configuration-response-common'][
797 def test_31_delete_eth_service2(self):
798 url = "{}/operations/org-openroadm-service:service-delete"
800 "sdnc-request-header": {
801 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
802 "rpc-action": "service-delete",
803 "request-system-id": "appname",
805 "http://localhost:8585/NotificationServer/notify"
807 "service-delete-req-info": {
808 "service-name": "service2",
809 "tail-retention": "no"
813 response = test_utils.post_request(url, data)
814 self.assertEqual(response.status_code, requests.codes.ok)
815 res = response.json()
816 self.assertIn('Renderer service delete in progress',
817 res['output']['configuration-response-common'][
821 def test_32_check_no_xc_ROADMA(self):
822 response = test_utils.check_netconf_node_request("ROADMA01", "")
823 res = response.json()
824 self.assertEqual(response.status_code, requests.codes.ok)
825 self.assertNotIn('roadm-connections',
826 dict.keys(res['org-openroadm-device']))
829 def test_33_check_topo_XPDRA(self):
830 response = test_utils.get_ordm_topo_request("node/XPDRA01-XPDR1")
831 self.assertEqual(response.status_code, requests.codes.ok)
832 res = response.json()
833 liste_tp = res['node'][0]['ietf-network-topology:termination-point']
835 if ((ele[u'org-openroadm-common-network:tp-type'] ==
837 and (ele['tp-id'] == 'XPDR1-CLIENT1' or ele[
838 'tp-id'] == 'XPDR1-CLIENT3')):
840 'org-openroadm-network-topology:xpdr-client-attributes',
842 elif (ele[u'org-openroadm-common-network:tp-type'] ==
844 self.assertIn(u'tail-equipment-id', dict.keys(
845 ele[u'org-openroadm-network-topology:'
846 u'xpdr-network-attributes']))
847 self.assertNotIn('wavelength', dict.keys(
848 ele[u'org-openroadm-network-topology:'
849 u'xpdr-network-attributes']))
852 def test_34_check_topo_ROADMA_SRG1(self):
853 response = test_utils.get_ordm_topo_request("node/ROADMA01-SRG1")
854 self.assertEqual(response.status_code, requests.codes.ok)
855 res = response.json()
856 self.assertIn({u'index': 1}, res['node'][0][
857 u'org-openroadm-network-topology:srg-attributes'][
858 'available-wavelengths'])
859 self.assertIn({u'index': 2}, res['node'][0][
860 u'org-openroadm-network-topology:srg-attributes'][
861 'available-wavelengths'])
862 liste_tp = res['node'][0]['ietf-network-topology:termination-point']
864 if ele['tp-id'] == 'SRG1-PP1-TXRX' or \
865 ele['tp-id'] == 'SRG1-PP1-TXRX':
866 self.assertNotIn('org-openroadm-network-topology:pp-attributes',
869 self.assertNotIn('org-openroadm-network-topology:pp-attributes',
873 def test_35_check_topo_ROADMA_DEG1(self):
874 response = test_utils.get_ordm_topo_request("node/ROADMA01-DEG1")
875 self.assertEqual(response.status_code, requests.codes.ok)
876 res = response.json()
877 self.assertIn({u'index': 1}, res['node'][0][
878 u'org-openroadm-network-topology:degree-attributes'][
879 'available-wavelengths'])
880 self.assertIn({u'index': 2}, res['node'][0][
881 u'org-openroadm-network-topology:degree-attributes'][
882 'available-wavelengths'])
883 liste_tp = res['node'][0]['ietf-network-topology:termination-point']
885 if ele['tp-id'] == 'DEG1-CTP-TXRX':
886 self.assertNotIn('org-openroadm-network-topology:'
887 'ctp-attributes', dict.keys(ele))
888 if ele['tp-id'] == 'DEG1-TTP-TXRX':
889 self.assertNotIn('org-openroadm-network-topology:'
890 'tx-ttp-attributes', dict.keys(ele))
893 # test service-create for Optical Channel (OC) service from srg-pp to srg-pp
894 def test_36_create_oc_service1(self):
895 url = "{}/operations/org-openroadm-service:service-create"
898 "sdnc-request-header": {
899 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
900 "rpc-action": "service-create",
901 "request-system-id": "appname",
903 "http://localhost:8585/NotificationServer/notify"
905 "service-name": "service1",
906 "common-id": "ASATT1234567",
907 "connection-type": "roadm-line",
909 "service-rate": "100",
910 "node-id": "ROADMA01",
911 "service-format": "OC",
912 "clli": "SNJSCAMCJP8",
916 "ROUTER_SNJSCAMCJP8_000000.00_00",
917 "port-type": "router",
918 "port-name": "Gigabit Ethernet_Tx.ge-5/0/0.0",
919 "port-rack": "000000.00",
924 "LGX Panel_SNJSCAMCJP8_000000.00_00",
925 "lgx-port-name": "LGX Back.3",
926 "lgx-port-rack": "000000.00",
927 "lgx-port-shelf": "00"
933 "ROUTER_SNJSCAMCJP8_000000.00_00",
934 "port-type": "router",
935 "port-name": "Gigabit Ethernet_Rx.ge-5/0/0.0",
936 "port-rack": "000000.00",
941 "LGX Panel_SNJSCAMCJP8_000000.00_00",
942 "lgx-port-name": "LGX Back.4",
943 "lgx-port-rack": "000000.00",
944 "lgx-port-shelf": "00"
950 "service-rate": "100",
951 "node-id": "ROADMC01",
952 "service-format": "OC",
953 "clli": "SNJSCAMCJT4",
957 "ROUTER_SNJSCAMCJT4_000000.00_00",
958 "port-type": "router",
959 "port-name": "Gigabit Ethernet_Tx.ge-1/0/0.0",
960 "port-rack": "000000.00",
965 "LGX Panel_SNJSCAMCJT4_000000.00_00",
966 "lgx-port-name": "LGX Back.29",
967 "lgx-port-rack": "000000.00",
968 "lgx-port-shelf": "00"
974 "ROUTER_SNJSCAMCJT4_000000.00_00",
975 "port-type": "router",
976 "port-name": "Gigabit Ethernet_Rx.ge-1/0/0.0",
977 "port-rack": "000000.00",
982 "LGX Panel_SNJSCAMCJT4_000000.00_00",
983 "lgx-port-name": "LGX Back.30",
984 "lgx-port-rack": "000000.00",
985 "lgx-port-shelf": "00"
990 "due-date": "2016-11-28T00:00:01Z",
991 "operator-contact": "pw1234"
994 response = test_utils.post_request(url, data)
995 self.assertEqual(response.status_code, requests.codes.ok)
996 res = response.json()
997 self.assertIn('PCE calculation in progress',
998 res['output']['configuration-response-common'][
1000 time.sleep(self.WAITING)
1002 def test_37_get_oc_service1(self):
1003 url = "{}/operational/org-openroadm-service:service-list/services/service1"
1004 response = test_utils.get_request(url)
1005 self.assertEqual(response.status_code, requests.codes.ok)
1006 res = response.json()
1008 res['services'][0]['administrative-state'],
1011 res['services'][0]['service-name'], 'service1')
1013 res['services'][0]['connection-type'], 'roadm-line')
1015 res['services'][0]['lifecycle-state'], 'planned')
1018 def test_38_check_xc1_ROADMA(self):
1019 response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-1")
1020 self.assertEqual(response.status_code, requests.codes.ok)
1021 res = response.json()
1022 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
1023 self.assertDictEqual(
1025 'connection-number': 'SRG1-PP1-TXRX-DEG1-TTP-TXRX-1',
1026 'wavelength-number': 1,
1027 'opticalControlMode': 'gainLoss',
1028 'target-output-power': -3.0
1029 }, **res['roadm-connections'][0]),
1030 res['roadm-connections'][0]
1032 self.assertDictEqual(
1033 {'src-if': 'SRG1-PP1-TXRX-1'},
1034 res['roadm-connections'][0]['source'])
1035 self.assertDictEqual(
1036 {'dst-if': 'DEG1-TTP-TXRX-1'},
1037 res['roadm-connections'][0]['destination'])
1040 def test_39_check_xc1_ROADMC(self):
1041 response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-1")
1042 self.assertEqual(response.status_code, requests.codes.ok)
1043 res = response.json()
1044 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
1045 self.assertDictEqual(
1047 'connection-number': 'SRG1-PP1-TXRX-DEG2-TTP-TXRX-1',
1048 'wavelength-number': 1,
1049 'opticalControlMode': 'gainLoss',
1050 'target-output-power': 2.0
1051 }, **res['roadm-connections'][0]),
1052 res['roadm-connections'][0]
1054 self.assertDictEqual(
1055 {'src-if': 'SRG1-PP1-TXRX-1'},
1056 res['roadm-connections'][0]['source'])
1057 self.assertDictEqual(
1058 {'dst-if': 'DEG2-TTP-TXRX-1'},
1059 res['roadm-connections'][0]['destination'])
1062 def test_40_create_oc_service2(self):
1063 url = "{}/operations/org-openroadm-service:service-create"
1066 "sdnc-request-header": {
1067 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
1068 "rpc-action": "service-create",
1069 "request-system-id": "appname",
1071 "http://localhost:8585/NotificationServer/notify"
1073 "service-name": "service2",
1074 "common-id": "ASATT1234567",
1075 "connection-type": "roadm-line",
1077 "service-rate": "100",
1078 "node-id": "ROADMA01",
1079 "service-format": "OC",
1080 "clli": "SNJSCAMCJP8",
1084 "ROUTER_SNJSCAMCJP8_000000.00_00",
1085 "port-type": "router",
1086 "port-name": "Gigabit Ethernet_Tx.ge-5/0/0.0",
1087 "port-rack": "000000.00",
1092 "LGX Panel_SNJSCAMCJP8_000000.00_00",
1093 "lgx-port-name": "LGX Back.3",
1094 "lgx-port-rack": "000000.00",
1095 "lgx-port-shelf": "00"
1101 "ROUTER_SNJSCAMCJP8_000000.00_00",
1102 "port-type": "router",
1103 "port-name": "Gigabit Ethernet_Rx.ge-5/0/0.0",
1104 "port-rack": "000000.00",
1109 "LGX Panel_SNJSCAMCJP8_000000.00_00",
1110 "lgx-port-name": "LGX Back.4",
1111 "lgx-port-rack": "000000.00",
1112 "lgx-port-shelf": "00"
1115 "optic-type": "gray"
1118 "service-rate": "100",
1119 "node-id": "ROADMC01",
1120 "service-format": "OC",
1121 "clli": "SNJSCAMCJT4",
1125 "ROUTER_SNJSCAMCJT4_000000.00_00",
1126 "port-type": "router",
1127 "port-name": "Gigabit Ethernet_Tx.ge-1/0/0.0",
1128 "port-rack": "000000.00",
1133 "LGX Panel_SNJSCAMCJT4_000000.00_00",
1134 "lgx-port-name": "LGX Back.29",
1135 "lgx-port-rack": "000000.00",
1136 "lgx-port-shelf": "00"
1142 "ROUTER_SNJSCAMCJT4_000000.00_00",
1143 "port-type": "router",
1144 "port-name": "Gigabit Ethernet_Rx.ge-1/0/0.0",
1145 "port-rack": "000000.00",
1150 "LGX Panel_SNJSCAMCJT4_000000.00_00",
1151 "lgx-port-name": "LGX Back.30",
1152 "lgx-port-rack": "000000.00",
1153 "lgx-port-shelf": "00"
1156 "optic-type": "gray"
1158 "due-date": "2016-11-28T00:00:01Z",
1159 "operator-contact": "pw1234"
1162 response = test_utils.post_request(url, data)
1163 self.assertEqual(response.status_code, requests.codes.ok)
1164 res = response.json()
1165 self.assertIn('PCE calculation in progress',
1166 res['output']['configuration-response-common'][
1167 'response-message'])
1168 time.sleep(self.WAITING)
1170 def test_41_get_oc_service2(self):
1171 url = "{}/operational/org-openroadm-service:service-list/services/service2"
1172 response = test_utils.get_request(url)
1173 self.assertEqual(response.status_code, requests.codes.ok)
1174 res = response.json()
1176 res['services'][0]['administrative-state'],
1179 res['services'][0]['service-name'], 'service2')
1181 res['services'][0]['connection-type'], 'roadm-line')
1183 res['services'][0]['lifecycle-state'], 'planned')
1186 def test_42_check_xc2_ROADMA(self):
1187 response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP2-TXRX-DEG1-TTP-TXRX-2")
1188 self.assertEqual(response.status_code, requests.codes.ok)
1189 res = response.json()
1190 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
1191 self.assertDictEqual(
1193 'connection-number': 'SRG1-PP2-TXRX-DEG1-TTP-TXRX-2',
1194 'wavelength-number': 2,
1195 'opticalControlMode': 'gainLoss',
1196 'target-output-power': -3.0
1197 }, **res['roadm-connections'][0]),
1198 res['roadm-connections'][0]
1200 self.assertDictEqual(
1201 {'src-if': 'SRG1-PP2-TXRX-2'},
1202 res['roadm-connections'][0]['source'])
1203 self.assertDictEqual(
1204 {'dst-if': 'DEG1-TTP-TXRX-2'},
1205 res['roadm-connections'][0]['destination'])
1208 def test_43_check_topo_ROADMA(self):
1209 self.test_26_check_topo_ROADMA_SRG1()
1210 self.test_27_check_topo_ROADMA_DEG1()
1213 def test_44_delete_oc_service1(self):
1214 url = "{}/operations/org-openroadm-service:service-delete"
1216 "sdnc-request-header": {
1217 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
1218 "rpc-action": "service-delete",
1219 "request-system-id": "appname",
1221 "http://localhost:8585/NotificationServer/notify"
1223 "service-delete-req-info": {
1224 "service-name": "service1",
1225 "tail-retention": "no"
1229 response = test_utils.post_request(url, data)
1230 self.assertEqual(response.status_code, requests.codes.ok)
1231 res = response.json()
1232 self.assertIn('Renderer service delete in progress',
1233 res['output']['configuration-response-common'][
1234 'response-message'])
1237 def test_45_delete_oc_service2(self):
1238 url = "{}/operations/org-openroadm-service:service-delete"
1240 "sdnc-request-header": {
1241 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
1242 "rpc-action": "service-delete",
1243 "request-system-id": "appname",
1245 "http://localhost:8585/NotificationServer/notify"
1247 "service-delete-req-info": {
1248 "service-name": "service2",
1249 "tail-retention": "no"
1253 response = test_utils.post_request(url, data)
1254 self.assertEqual(response.status_code, requests.codes.ok)
1255 res = response.json()
1256 self.assertIn('Renderer service delete in progress',
1257 res['output']['configuration-response-common'][
1258 'response-message'])
1261 def test_46_get_no_oc_services(self):
1263 url = "{}/operational/org-openroadm-service:service-list"
1264 response = test_utils.get_request(url)
1265 self.assertEqual(response.status_code, requests.codes.not_found)
1266 res = response.json()
1269 "error-type": "application",
1270 "error-tag": "data-missing",
1272 "Request could not be completed because the relevant data "
1273 "model content does not exist"
1275 res['errors']['error'])
1278 def test_47_get_no_xc_ROADMA(self):
1279 response = test_utils.check_netconf_node_request("ROADMA01", "")
1280 self.assertEqual(response.status_code, requests.codes.ok)
1281 res = response.json()
1282 self.assertNotIn(['roadm-connections'][0], res['org-openroadm-device'])
1285 def test_48_check_topo_ROADMA(self):
1286 self.test_34_check_topo_ROADMA_SRG1()
1287 self.test_35_check_topo_ROADMA_DEG1()
1289 def test_49_loop_create_eth_service(self):
1290 for i in range(1, 6):
1291 print("iteration number {}".format(i))
1292 print("eth service creation")
1293 self.test_11_create_eth_service1()
1294 print("check xc in ROADMA01")
1295 self.test_13_check_xc1_ROADMA()
1296 print("check xc in ROADMC01")
1297 self.test_14_check_xc1_ROADMC()
1298 print("eth service deletion\n")
1299 self.test_30_delete_eth_service1()
1301 def test_50_loop_create_oc_service(self):
1302 url = "{}/operational/org-openroadm-service:service-list/services/service1"
1303 response = test_utils.get_request(url)
1304 if response.status_code != 404:
1305 url = "{}/operations/org-openroadm-service:service-delete"
1307 "sdnc-request-header": {
1308 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
1309 "rpc-action": "service-delete",
1310 "request-system-id": "appname",
1312 "http://localhost:8585/NotificationServer/notify"
1314 "service-delete-req-info": {
1315 "service-name": "service1",
1316 "tail-retention": "no"
1320 test_utils.post_request(url, data)
1323 for i in range(1, 6):
1324 print("iteration number {}".format(i))
1325 print("oc service creation")
1326 self.test_36_create_oc_service1()
1327 print("check xc in ROADMA01")
1328 self.test_38_check_xc1_ROADMA()
1329 print("check xc in ROADMC01")
1330 self.test_39_check_xc1_ROADMC()
1331 print("oc service deletion\n")
1332 self.test_44_delete_oc_service1()
1334 def test_51_disconnect_XPDRA(self):
1335 response = test_utils.unmount_device("XPDRA01")
1336 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
1338 def test_52_disconnect_XPDRC(self):
1339 response = test_utils.unmount_device("XPDRC01")
1340 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
1342 def test_53_disconnect_ROADMA(self):
1343 response = test_utils.unmount_device("ROADMA01")
1344 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
1346 def test_54_disconnect_ROADMC(self):
1347 response = test_utils.unmount_device("ROADMC01")
1348 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
1351 if __name__ == "__main__":
1352 unittest.main(verbosity=2)