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 ##############################################################################
16 from common import test_utils
19 class TransportPCEFulltesting(unittest.TestCase):
22 WAITING = 20 # nominal value is 300
26 cls.processes = test_utils.start_tpce()
27 cls.processes = test_utils.start_sims(['xpdra', 'roadma', 'roadmc', 'xpdrc'])
30 def tearDownClass(cls):
31 for process in cls.processes:
32 test_utils.shutdown_process(process)
33 print("all processes killed")
35 def setUp(self): # instruction executed before each test method
36 print("execution of {}".format(self.id().split(".")[-1]))
38 def test_01_connect_xpdrA(self):
39 response = test_utils.mount_device("XPDR-A1", 'xpdra')
40 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
42 def test_02_connect_xpdrC(self):
43 response = test_utils.mount_device("XPDR-C1", 'xpdrc')
44 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
46 def test_03_connect_rdmA(self):
47 response = test_utils.mount_device("ROADM-A1", 'roadma')
48 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
50 def test_04_connect_rdmC(self):
51 response = test_utils.mount_device("ROADM-C1", 'roadmc')
52 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
54 def test_05_connect_xprdA_N1_to_roadmA_PP1(self):
55 response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "1",
56 "ROADM-A1", "1", "SRG1-PP1-TXRX")
57 self.assertEqual(response.status_code, requests.codes.ok)
59 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
62 def test_06_connect_roadmA_PP1_to_xpdrA_N1(self):
63 response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "1",
64 "ROADM-A1", "1", "SRG1-PP1-TXRX")
65 self.assertEqual(response.status_code, requests.codes.ok)
67 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
70 def test_07_connect_xprdC_N1_to_roadmC_PP1(self):
71 response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "1",
72 "ROADM-C1", "1", "SRG1-PP1-TXRX")
73 self.assertEqual(response.status_code, requests.codes.ok)
75 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
78 def test_08_connect_roadmC_PP1_to_xpdrC_N1(self):
79 response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "1",
80 "ROADM-C1", "1", "SRG1-PP1-TXRX")
81 self.assertEqual(response.status_code, requests.codes.ok)
83 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
86 def test_09_add_omsAttributes_ROADMA_ROADMC(self):
87 # Config ROADMA-ROADMC oms-attributes
88 url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
89 "link/ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX/org-openroadm-network-topology:"
90 "OMS-attributes/span")
92 "auto-spanloss": "true",
93 "spanloss-base": 11.4,
94 "spanloss-current": 12,
95 "engineered-spanloss": 12.2,
96 "link-concatenation": [{
99 "SRLG-length": 100000,
101 response = test_utils.put_request(url, data)
102 self.assertEqual(response.status_code, requests.codes.created)
104 def test_10_add_omsAttributes_ROADMC_ROADMA(self):
105 # Config ROADMC-ROADMA oms-attributes
106 url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
107 "link/ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX/org-openroadm-network-topology:"
108 "OMS-attributes/span")
110 "auto-spanloss": "true",
111 "spanloss-base": 11.4,
112 "spanloss-current": 12,
113 "engineered-spanloss": 12.2,
114 "link-concatenation": [{
117 "SRLG-length": 100000,
119 response = test_utils.put_request(url, data)
120 self.assertEqual(response.status_code, requests.codes.created)
122 # test service-create for Eth service from xpdr to xpdr
123 def test_11_create_eth_service1(self):
124 url = "{}/operations/org-openroadm-service:service-create"
126 "sdnc-request-header": {
127 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
128 "rpc-action": "service-create",
129 "request-system-id": "appname",
130 "notification-url": "http://localhost:8585/NotificationServer/notify"
132 "service-name": "service1",
133 "common-id": "ASATT1234567",
134 "connection-type": "service",
136 "service-rate": "100",
137 "node-id": "XPDR-A1",
138 "service-format": "Ethernet",
139 "clli": "SNJSCAMCJP8",
142 "port-device-name": "ROUTER_SNJSCAMCJP8_000000.00_00",
143 "port-type": "router",
144 "port-name": "Gigabit Ethernet_Tx.ge-5/0/0.0",
145 "port-rack": "000000.00",
149 "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00",
150 "lgx-port-name": "LGX Back.3",
151 "lgx-port-rack": "000000.00",
152 "lgx-port-shelf": "00"
157 "port-device-name": "ROUTER_SNJSCAMCJP8_000000.00_00",
158 "port-type": "router",
159 "port-name": "Gigabit Ethernet_Rx.ge-5/0/0.0",
160 "port-rack": "000000.00",
164 "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00",
165 "lgx-port-name": "LGX Back.4",
166 "lgx-port-rack": "000000.00",
167 "lgx-port-shelf": "00"
173 "service-rate": "100",
174 "node-id": "XPDR-C1",
175 "service-format": "Ethernet",
176 "clli": "SNJSCAMCJT4",
179 "port-device-name": "ROUTER_SNJSCAMCJT4_000000.00_00",
180 "port-type": "router",
181 "port-name": "Gigabit Ethernet_Tx.ge-1/0/0.0",
182 "port-rack": "000000.00",
186 "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00",
187 "lgx-port-name": "LGX Back.29",
188 "lgx-port-rack": "000000.00",
189 "lgx-port-shelf": "00"
194 "port-device-name": "ROUTER_SNJSCAMCJT4_000000.00_00",
195 "port-type": "router",
196 "port-name": "Gigabit Ethernet_Rx.ge-1/0/0.0",
197 "port-rack": "000000.00",
201 "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00",
202 "lgx-port-name": "LGX Back.30",
203 "lgx-port-rack": "000000.00",
204 "lgx-port-shelf": "00"
209 "due-date": "2016-11-28T00:00:01Z",
210 "operator-contact": "pw1234"
213 response = test_utils.post_request(url, data)
214 self.assertEqual(response.status_code, requests.codes.ok)
215 res = response.json()
216 self.assertIn('PCE calculation in progress',
217 res['output']['configuration-response-common']['response-message'])
218 time.sleep(self.WAITING)
220 def test_12_get_eth_service1(self):
221 url = "{}/operational/org-openroadm-service:service-list/services/service1"
222 response = test_utils.get_request(url)
223 self.assertEqual(response.status_code, requests.codes.ok)
224 res = response.json()
226 res['services'][0]['administrative-state'], 'inService')
228 res['services'][0]['service-name'], 'service1')
230 res['services'][0]['connection-type'], 'service')
232 res['services'][0]['lifecycle-state'], 'planned')
235 def test_13_check_xc1_ROADMA(self):
236 response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-1")
237 self.assertEqual(response.status_code, requests.codes.ok)
238 res = response.json()
239 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
240 self.assertDictEqual(
242 'connection-name': 'SRG1-PP1-TXRX-DEG2-TTP-TXRX-1',
243 'opticalControlMode': 'gainLoss',
244 'target-output-power': -3.0
245 }, **res['roadm-connections'][0]),
246 res['roadm-connections'][0]
248 self.assertDictEqual(
249 {'src-if': 'SRG1-PP1-TXRX-nmc-1'},
250 res['roadm-connections'][0]['source'])
251 self.assertDictEqual(
252 {'dst-if': 'DEG2-TTP-TXRX-nmc-1'},
253 res['roadm-connections'][0]['destination'])
256 def test_14_check_xc1_ROADMC(self):
257 response = test_utils.check_netconf_node_request("ROADM-C1", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-1")
258 self.assertEqual(response.status_code, requests.codes.ok)
259 res = response.json()
260 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
261 self.assertDictEqual(
263 'connection-name': 'SRG1-PP1-TXRX-DEG1-TTP-TXRX-1',
264 'opticalControlMode': 'gainLoss',
265 'target-output-power': -3.0
266 }, **res['roadm-connections'][0]),
267 res['roadm-connections'][0]
269 self.assertDictEqual(
270 {'src-if': 'SRG1-PP1-TXRX-nmc-1'},
271 res['roadm-connections'][0]['source'])
272 self.assertDictEqual(
273 {'dst-if': 'DEG1-TTP-TXRX-nmc-1'},
274 res['roadm-connections'][0]['destination'])
277 def test_15_check_topo_XPDRA(self):
278 url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPDR-A1-XPDR1"
279 response = test_utils.get_request(url)
280 self.assertEqual(response.status_code, requests.codes.ok)
281 res = response.json()
282 liste_tp = res['node'][0]['ietf-network-topology:termination-point']
284 if ele['tp-id'] == 'XPDR1-NETWORK1':
285 self.assertEqual({u'frequency': 196.1,
287 ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength'])
288 if ele['tp-id'] == 'XPDR1-CLIENT2' or ele['tp-id'] == 'XPDR1-CLIENT1':
289 self.assertNotIn('org-openroadm-network-topology:xpdr-client-attributes', dict.keys(ele))
290 if ele['tp-id'] == 'XPDR1-NETWORK2':
291 self.assertNotIn('org-openroadm-network-topology:xpdr-network-attributes', dict.keys(ele))
294 def test_16_check_topo_ROADMA_SRG1(self):
295 url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADM-A1-SRG1"
296 response = test_utils.get_request(url)
297 self.assertEqual(response.status_code, requests.codes.ok)
298 res = response.json()
299 self.assertNotIn({u'index': 1},
300 res['node'][0][u'org-openroadm-network-topology:srg-attributes']['available-wavelengths'])
301 liste_tp = res['node'][0]['ietf-network-topology:termination-point']
303 if ele['tp-id'] == 'SRG1-PP1-TXRX':
304 self.assertIn({u'index': 1, u'frequency': 196.1,
306 ele['org-openroadm-network-topology:pp-attributes']['used-wavelength'])
307 if ele['tp-id'] == 'SRG1-PP2-TXRX':
308 self.assertNotIn('used-wavelength', dict.keys(ele))
311 def test_17_check_topo_ROADMA_DEG1(self):
312 url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADM-A1-DEG2"
313 response = test_utils.get_request(url)
314 self.assertEqual(response.status_code, requests.codes.ok)
315 res = response.json()
316 self.assertNotIn({u'index': 1},
317 res['node'][0][u'org-openroadm-network-topology:degree-attributes']['available-wavelengths'])
318 liste_tp = res['node'][0]['ietf-network-topology:termination-point']
320 if ele['tp-id'] == 'DEG2-CTP-TXRX':
321 self.assertIn({u'index': 1, u'frequency': 196.1,
323 ele['org-openroadm-network-topology:ctp-attributes']['used-wavelengths'])
324 if ele['tp-id'] == 'DEG2-TTP-TXRX':
325 self.assertIn({u'index': 1, u'frequency': 196.1,
327 ele['org-openroadm-network-topology:tx-ttp-attributes']['used-wavelengths'])
330 def test_18_connect_xprdA_N2_to_roadmA_PP2(self):
331 response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "2",
332 "ROADM-A1", "1", "SRG1-PP2-TXRX")
333 self.assertEqual(response.status_code, requests.codes.ok)
334 res = response.json()
335 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
338 def test_19_connect_roadmA_PP2_to_xpdrA_N2(self):
339 response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "2",
340 "ROADM-A1", "1", "SRG1-PP2-TXRX")
341 self.assertEqual(response.status_code, requests.codes.ok)
342 res = response.json()
343 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
346 def test_20_connect_xprdC_N2_to_roadmC_PP2(self):
347 response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "2",
348 "ROADM-C1", "1", "SRG1-PP2-TXRX")
349 self.assertEqual(response.status_code, requests.codes.ok)
350 res = response.json()
351 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
354 def test_21_connect_roadmC_PP2_to_xpdrC_N2(self):
355 response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "2",
356 "ROADM-C1", "1", "SRG1-PP2-TXRX")
357 self.assertEqual(response.status_code, requests.codes.ok)
358 res = response.json()
359 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
362 def test_22_create_eth_service2(self):
363 url = "{}/operations/org-openroadm-service:service-create"
365 "sdnc-request-header": {
366 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
367 "rpc-action": "service-create",
368 "request-system-id": "appname",
369 "notification-url": "http://localhost:8585/NotificationServer/notify"
371 "service-name": "service2",
372 "common-id": "ASATT1234567",
373 "connection-type": "service",
375 "service-rate": "100",
376 "node-id": "XPDR-A1",
377 "service-format": "Ethernet",
378 "clli": "SNJSCAMCJP8",
381 "port-device-name": "ROUTER_SNJSCAMCJP8_000000.00_00",
382 "port-type": "router",
383 "port-name": "Gigabit Ethernet_Tx.ge-5/0/0.0",
384 "port-rack": "000000.00",
388 "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00",
389 "lgx-port-name": "LGX Back.3",
390 "lgx-port-rack": "000000.00",
391 "lgx-port-shelf": "00"
396 "port-device-name": "ROUTER_SNJSCAMCJP8_000000.00_00",
397 "port-type": "router",
398 "port-name": "Gigabit Ethernet_Rx.ge-5/0/0.0",
399 "port-rack": "000000.00",
403 "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00",
404 "lgx-port-name": "LGX Back.4",
405 "lgx-port-rack": "000000.00",
406 "lgx-port-shelf": "00"
412 "service-rate": "100",
413 "node-id": "XPDR-C1",
414 "service-format": "Ethernet",
415 "clli": "SNJSCAMCJT4",
418 "port-device-name": "ROUTER_SNJSCAMCJT4_000000.00_00",
419 "port-type": "router",
420 "port-name": "Gigabit Ethernet_Tx.ge-1/0/0.0",
421 "port-rack": "000000.00",
425 "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00",
426 "lgx-port-name": "LGX Back.29",
427 "lgx-port-rack": "000000.00",
428 "lgx-port-shelf": "00"
433 "port-device-name": "ROUTER_SNJSCAMCJT4_000000.00_00",
434 "port-type": "router",
435 "port-name": "Gigabit Ethernet_Rx.ge-1/0/0.0",
436 "port-rack": "000000.00",
440 "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00",
441 "lgx-port-name": "LGX Back.30",
442 "lgx-port-rack": "000000.00",
443 "lgx-port-shelf": "00"
448 "due-date": "2016-11-28T00:00:01Z",
449 "operator-contact": "pw1234"
452 response = test_utils.post_request(url, data)
453 self.assertEqual(response.status_code, requests.codes.ok)
454 res = response.json()
455 self.assertIn('PCE calculation in progress',
456 res['output']['configuration-response-common']['response-message'])
457 time.sleep(self.WAITING)
459 def test_23_get_eth_service2(self):
460 url = "{}/operational/org-openroadm-service:service-list/services/service2"
461 response = test_utils.get_request(url)
462 self.assertEqual(response.status_code, requests.codes.ok)
463 res = response.json()
465 res['services'][0]['administrative-state'],
468 res['services'][0]['service-name'], 'service2')
470 res['services'][0]['connection-type'], 'service')
472 res['services'][0]['lifecycle-state'], 'planned')
475 def test_24_check_xc2_ROADMA(self):
476 response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/DEG2-TTP-TXRX-SRG1-PP2-TXRX-2")
477 self.assertEqual(response.status_code, requests.codes.ok)
478 res = response.json()
479 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
480 self.assertDictEqual(
482 'connection-name': 'DEG2-TTP-TXRX-SRG1-PP2-TXRX-2',
483 'opticalControlMode': 'power'
484 }, **res['roadm-connections'][0]),
485 res['roadm-connections'][0]
487 self.assertDictEqual(
488 {'src-if': 'DEG2-TTP-TXRX-nmc-2'},
489 res['roadm-connections'][0]['source'])
490 self.assertDictEqual(
491 {'dst-if': 'SRG1-PP2-TXRX-nmc-2'},
492 res['roadm-connections'][0]['destination'])
494 def test_25_check_topo_XPDRA(self):
495 url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPDR-A1-XPDR1"
496 response = test_utils.get_request(url)
497 self.assertEqual(response.status_code, requests.codes.ok)
498 res = response.json()
499 liste_tp = res['node'][0]['ietf-network-topology:termination-point']
501 if ele['tp-id'] == 'XPDR1-NETWORK1':
502 self.assertEqual({u'frequency': 196.1,
504 ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength'])
505 if ele['tp-id'] == 'XPDR1-NETWORK2':
506 self.assertEqual({u'frequency': 196.05,
508 ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength'])
509 if ele['tp-id'] == 'XPDR1-CLIENT1' or ele['tp-id'] == 'XPDR1-CLIENT2':
510 self.assertNotIn('org-openroadm-network-topology:xpdr-client-attributes', dict.keys(ele))
513 def test_26_check_topo_ROADMA_SRG1(self):
514 url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADM-A1-SRG1"
515 response = test_utils.get_request(url)
516 self.assertEqual(response.status_code, requests.codes.ok)
517 res = response.json()
518 self.assertNotIn({u'index': 1}, res['node'][0]
519 [u'org-openroadm-network-topology:srg-attributes']['available-wavelengths'])
520 self.assertNotIn({u'index': 2}, res['node'][0]
521 [u'org-openroadm-network-topology:srg-attributes']['available-wavelengths'])
522 liste_tp = res['node'][0]['ietf-network-topology:termination-point']
524 if ele['tp-id'] == 'SRG1-PP1-TXRX':
525 self.assertIn({u'index': 1, u'frequency': 196.1,
527 ele['org-openroadm-network-topology:pp-attributes']['used-wavelength'])
528 self.assertNotIn({u'index': 2, u'frequency': 196.05,
530 ele['org-openroadm-network-topology:pp-attributes']['used-wavelength'])
531 if ele['tp-id'] == 'SRG1-PP2-TXRX':
532 self.assertIn({u'index': 2, u'frequency': 196.05, u'width': 40},
533 ele['org-openroadm-network-topology:pp-attributes']['used-wavelength'])
534 self.assertNotIn({u'index': 1, u'frequency': 196.1,
536 ele['org-openroadm-network-topology:pp-attributes']['used-wavelength'])
537 if ele['tp-id'] == 'SRG1-PP3-TXRX':
538 self.assertNotIn('org-openroadm-network-topology:pp-attributes', dict.keys(ele))
541 def test_27_check_topo_ROADMA_DEG2(self):
542 url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADM-A1-DEG2"
543 response = test_utils.get_request(url)
544 self.assertEqual(response.status_code, requests.codes.ok)
545 res = response.json()
546 self.assertNotIn({u'index': 1}, res['node'][0]
547 [u'org-openroadm-network-topology:degree-attributes']['available-wavelengths'])
548 self.assertNotIn({u'index': 2}, res['node'][0]
549 [u'org-openroadm-network-topology:degree-attributes']['available-wavelengths'])
550 liste_tp = res['node'][0]['ietf-network-topology:termination-point']
552 if ele['tp-id'] == 'DEG2-CTP-TXRX':
553 self.assertIn({u'index': 1, u'frequency': 196.1,
555 ele['org-openroadm-network-topology:ctp-attributes']['used-wavelengths'])
556 self.assertIn({u'index': 2, u'frequency': 196.05, u'width': 40},
557 ele['org-openroadm-network-topology:ctp-attributes']['used-wavelengths'])
558 if ele['tp-id'] == 'DEG2-TTP-TXRX':
559 self.assertIn({u'index': 1, u'frequency': 196.1,
561 ele['org-openroadm-network-topology:tx-ttp-attributes']['used-wavelengths'])
562 self.assertIn({u'index': 2, u'frequency': 196.05, u'width': 40},
563 ele['org-openroadm-network-topology:tx-ttp-attributes']['used-wavelengths'])
566 # creation service test on a non-available resource
567 def test_28_create_eth_service3(self):
568 url = "{}/operations/org-openroadm-service:service-create"
570 "sdnc-request-header": {
571 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
572 "rpc-action": "service-create",
573 "request-system-id": "appname",
574 "notification-url": "http://localhost:8585/NotificationServer/notify"
576 "service-name": "service3",
577 "common-id": "ASATT1234567",
578 "connection-type": "service",
580 "service-rate": "100",
581 "node-id": "XPDR-A1",
582 "service-format": "Ethernet",
583 "clli": "SNJSCAMCJP8",
586 "port-device-name": "ROUTER_SNJSCAMCJP8_000000.00_00",
587 "port-type": "router",
588 "port-name": "Gigabit Ethernet_Tx.ge-5/0/0.0",
589 "port-rack": "000000.00",
593 "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00",
594 "lgx-port-name": "LGX Back.3",
595 "lgx-port-rack": "000000.00",
596 "lgx-port-shelf": "00"
601 "port-device-name": "ROUTER_SNJSCAMCJP8_000000.00_00",
602 "port-type": "router",
603 "port-name": "Gigabit Ethernet_Rx.ge-5/0/0.0",
604 "port-rack": "000000.00",
608 "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00",
609 "lgx-port-name": "LGX Back.4",
610 "lgx-port-rack": "000000.00",
611 "lgx-port-shelf": "00"
617 "service-rate": "100",
618 "node-id": "XPDR-C1",
619 "service-format": "Ethernet",
620 "clli": "SNJSCAMCJT4",
623 "port-device-name": "ROUTER_SNJSCAMCJT4_000000.00_00",
624 "port-type": "router",
625 "port-name": "Gigabit Ethernet_Tx.ge-1/0/0.0",
626 "port-rack": "000000.00",
630 "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00",
631 "lgx-port-name": "LGX Back.29",
632 "lgx-port-rack": "000000.00",
633 "lgx-port-shelf": "00"
638 "port-device-name": "ROUTER_SNJSCAMCJT4_000000.00_00",
639 "port-type": "router",
640 "port-name": "Gigabit Ethernet_Rx.ge-1/0/0.0",
641 "port-rack": "000000.00",
645 "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00",
646 "lgx-port-name": "LGX Back.30",
647 "lgx-port-rack": "000000.00",
648 "lgx-port-shelf": "00"
653 "due-date": "2016-11-28T00:00:01Z",
654 "operator-contact": "pw1234"
657 response = test_utils.post_request(url, data)
658 self.assertEqual(response.status_code, requests.codes.ok)
659 res = response.json()
660 self.assertIn('PCE calculation in progress',
661 res['output']['configuration-response-common']['response-message'])
662 self.assertIn('200', res['output']['configuration-response-common']['response-code'])
663 time.sleep(self.WAITING)
665 # add a test that check the openroadm-service-list still only contains 2 elements
666 def test_29_delete_eth_service3(self):
667 url = "{}/operations/org-openroadm-service:service-delete"
669 "sdnc-request-header": {
670 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
671 "rpc-action": "service-delete",
672 "request-system-id": "appname",
673 "notification-url": "http://localhost:8585/NotificationServer/notify"
675 "service-delete-req-info": {
676 "service-name": "service3",
677 "tail-retention": "no"
681 response = test_utils.post_request(url, data)
682 self.assertEqual(response.status_code, requests.codes.ok)
683 res = response.json()
684 self.assertIn('Service \'service3\' does not exist in datastore',
685 res['output']['configuration-response-common']['response-message'])
686 self.assertIn('500', res['output']['configuration-response-common']['response-code'])
689 def test_30_delete_eth_service1(self):
690 url = "{}/operations/org-openroadm-service:service-delete"
692 "sdnc-request-header": {
693 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
694 "rpc-action": "service-delete",
695 "request-system-id": "appname",
696 "notification-url": "http://localhost:8585/NotificationServer/notify"
698 "service-delete-req-info": {
699 "service-name": "service1",
700 "tail-retention": "no"
704 response = test_utils.post_request(url, data)
705 self.assertEqual(response.status_code, requests.codes.ok)
706 res = response.json()
707 self.assertIn('Renderer service delete in progress',
708 res['output']['configuration-response-common']['response-message'])
711 def test_31_delete_eth_service2(self):
712 url = "{}/operations/org-openroadm-service:service-delete"
714 "sdnc-request-header": {
715 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
716 "rpc-action": "service-delete",
717 "request-system-id": "appname",
718 "notification-url": "http://localhost:8585/NotificationServer/notify"
720 "service-delete-req-info": {
721 "service-name": "service2",
722 "tail-retention": "no"
726 response = test_utils.post_request(url, data)
727 self.assertEqual(response.status_code, requests.codes.ok)
728 res = response.json()
729 self.assertIn('Renderer service delete in progress',
730 res['output']['configuration-response-common']['response-message'])
733 def test_32_check_no_xc_ROADMA(self):
734 response = test_utils.check_netconf_node_request("ROADM-A1", "")
735 res = response.json()
736 self.assertEqual(response.status_code, requests.codes.ok)
737 self.assertNotIn('roadm-connections', dict.keys(res['org-openroadm-device']))
740 def test_33_check_topo_XPDRA(self):
741 url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPDR-A1-XPDR1"
742 response = test_utils.get_request(url)
743 self.assertEqual(response.status_code, requests.codes.ok)
744 res = response.json()
745 liste_tp = res['node'][0]['ietf-network-topology:termination-point']
747 if ele[u'org-openroadm-common-network:tp-type'] == 'XPONDER-CLIENT':
748 self.assertNotIn('org-openroadm-network-topology:xpdr-client-attributes', dict.keys(ele))
749 elif (ele[u'org-openroadm-common-network:tp-type'] == 'XPONDER-NETWORK'):
750 self.assertIn(u'tail-equipment-id',
751 dict.keys(ele[u'org-openroadm-network-topology:xpdr-network-attributes']))
752 self.assertNotIn('wavelength', dict.keys(
753 ele[u'org-openroadm-network-topology:xpdr-network-attributes']))
756 def test_34_check_topo_ROADMA_SRG1(self):
757 url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADM-A1-SRG1"
758 response = test_utils.get_request(url)
759 self.assertEqual(response.status_code, requests.codes.ok)
760 res = response.json()
761 self.assertIn({u'index': 1}, res['node'][0]
762 [u'org-openroadm-network-topology:srg-attributes']['available-wavelengths'])
763 self.assertIn({u'index': 2}, res['node'][0]
764 [u'org-openroadm-network-topology:srg-attributes']['available-wavelengths'])
765 liste_tp = res['node'][0]['ietf-network-topology:termination-point']
767 if ele['tp-id'] == 'SRG1-PP1-TXRX' or ele['tp-id'] == 'SRG1-PP1-TXRX':
768 self.assertNotIn('org-openroadm-network-topology:pp-attributes', dict.keys(ele))
770 self.assertNotIn('org-openroadm-network-topology:pp-attributes', dict.keys(ele))
773 def test_35_check_topo_ROADMA_DEG2(self):
774 url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADM-A1-DEG2"
775 response = test_utils.get_request(url)
776 self.assertEqual(response.status_code, requests.codes.ok)
777 res = response.json()
778 self.assertIn({u'index': 1}, res['node'][0]
779 [u'org-openroadm-network-topology:degree-attributes']['available-wavelengths'])
780 self.assertIn({u'index': 2}, res['node'][0]
781 [u'org-openroadm-network-topology:degree-attributes']['available-wavelengths'])
782 liste_tp = res['node'][0]['ietf-network-topology:termination-point']
784 if ele['tp-id'] == 'DEG2-CTP-TXRX':
785 self.assertNotIn('org-openroadm-network-topology:ctp-attributes', dict.keys(ele))
786 if ele['tp-id'] == 'DEG2-TTP-TXRX':
787 self.assertNotIn('org-openroadm-network-topology:tx-ttp-attributes', dict.keys(ele))
790 # test service-create for Optical Channel (OC) service from srg-pp to srg-pp
791 def test_36_create_oc_service1(self):
792 url = "{}/operations/org-openroadm-service:service-create"
794 "sdnc-request-header": {
795 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
796 "rpc-action": "service-create",
797 "request-system-id": "appname",
798 "notification-url": "http://localhost:8585/NotificationServer/notify"
800 "service-name": "service1",
801 "common-id": "ASATT1234567",
802 "connection-type": "roadm-line",
804 "service-rate": "100",
805 "node-id": "ROADM-A1",
806 "service-format": "OC",
807 "clli": "SNJSCAMCJP8",
810 "port-device-name": "ROUTER_SNJSCAMCJP8_000000.00_00",
811 "port-type": "router",
812 "port-name": "Gigabit Ethernet_Tx.ge-5/0/0.0",
813 "port-rack": "000000.00",
817 "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00",
818 "lgx-port-name": "LGX Back.3",
819 "lgx-port-rack": "000000.00",
820 "lgx-port-shelf": "00"
825 "port-device-name": "ROUTER_SNJSCAMCJP8_000000.00_00",
826 "port-type": "router",
827 "port-name": "Gigabit Ethernet_Rx.ge-5/0/0.0",
828 "port-rack": "000000.00",
832 "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00",
833 "lgx-port-name": "LGX Back.4",
834 "lgx-port-rack": "000000.00",
835 "lgx-port-shelf": "00"
841 "service-rate": "100",
842 "node-id": "ROADM-C1",
843 "service-format": "OC",
844 "clli": "SNJSCAMCJT4",
847 "port-device-name": "ROUTER_SNJSCAMCJT4_000000.00_00",
848 "port-type": "router",
849 "port-name": "Gigabit Ethernet_Tx.ge-1/0/0.0",
850 "port-rack": "000000.00",
854 "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00",
855 "lgx-port-name": "LGX Back.29",
856 "lgx-port-rack": "000000.00",
857 "lgx-port-shelf": "00"
862 "port-device-name": "ROUTER_SNJSCAMCJT4_000000.00_00",
863 "port-type": "router",
864 "port-name": "Gigabit Ethernet_Rx.ge-1/0/0.0",
865 "port-rack": "000000.00",
869 "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00",
870 "lgx-port-name": "LGX Back.30",
871 "lgx-port-rack": "000000.00",
872 "lgx-port-shelf": "00"
877 "due-date": "2016-11-28T00:00:01Z",
878 "operator-contact": "pw1234"
881 response = test_utils.post_request(url, data)
882 self.assertEqual(response.status_code, requests.codes.ok)
883 res = response.json()
884 self.assertIn('PCE calculation in progress',
885 res['output']['configuration-response-common']['response-message'])
886 time.sleep(self.WAITING)
888 def test_37_get_oc_service1(self):
889 url = "{}/operational/org-openroadm-service:service-list/services/service1"
890 response = test_utils.get_request(url)
891 self.assertEqual(response.status_code, requests.codes.ok)
892 res = response.json()
894 res['services'][0]['administrative-state'],
897 res['services'][0]['service-name'], 'service1')
899 res['services'][0]['connection-type'], 'roadm-line')
901 res['services'][0]['lifecycle-state'], 'planned')
904 def test_38_check_xc1_ROADMA(self):
905 response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-1")
906 self.assertEqual(response.status_code, requests.codes.ok)
907 res = response.json()
908 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
909 self.assertDictEqual(
911 'connection-name': 'SRG1-PP1-TXRX-DEG2-TTP-TXRX-1',
912 'opticalControlMode': 'gainLoss',
913 'target-output-power': -3.0
914 }, **res['roadm-connections'][0]),
915 res['roadm-connections'][0]
917 self.assertDictEqual(
918 {'src-if': 'SRG1-PP1-TXRX-nmc-1'},
919 res['roadm-connections'][0]['source'])
920 self.assertDictEqual(
921 {'dst-if': 'DEG2-TTP-TXRX-nmc-1'},
922 res['roadm-connections'][0]['destination'])
925 def test_39_check_xc1_ROADMC(self):
926 response = test_utils.check_netconf_node_request("ROADM-C1", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-1")
927 self.assertEqual(response.status_code, requests.codes.ok)
928 res = response.json()
929 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
930 self.assertDictEqual(
932 'connection-name': 'SRG1-PP1-TXRX-DEG1-TTP-TXRX-1',
933 'opticalControlMode': 'gainLoss',
934 'target-output-power': -3.0
935 }, **res['roadm-connections'][0]),
936 res['roadm-connections'][0]
938 self.assertDictEqual(
939 {'src-if': 'SRG1-PP1-TXRX-nmc-1'},
940 res['roadm-connections'][0]['source'])
941 self.assertDictEqual(
942 {'dst-if': 'DEG1-TTP-TXRX-nmc-1'},
943 res['roadm-connections'][0]['destination'])
946 def test_40_create_oc_service2(self):
947 url = "{}/operations/org-openroadm-service:service-create"
949 "sdnc-request-header": {
950 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
951 "rpc-action": "service-create",
952 "request-system-id": "appname",
953 "notification-url": "http://localhost:8585/NotificationServer/notify"
955 "service-name": "service2",
956 "common-id": "ASATT1234567",
957 "connection-type": "roadm-line",
959 "service-rate": "100",
960 "node-id": "ROADM-A1",
961 "service-format": "OC",
962 "clli": "SNJSCAMCJP8",
965 "port-device-name": "ROUTER_SNJSCAMCJP8_000000.00_00",
966 "port-type": "router",
967 "port-name": "Gigabit Ethernet_Tx.ge-5/0/0.0",
968 "port-rack": "000000.00",
972 "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00",
973 "lgx-port-name": "LGX Back.3",
974 "lgx-port-rack": "000000.00",
975 "lgx-port-shelf": "00"
980 "port-device-name": "ROUTER_SNJSCAMCJP8_000000.00_00",
981 "port-type": "router",
982 "port-name": "Gigabit Ethernet_Rx.ge-5/0/0.0",
983 "port-rack": "000000.00",
987 "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00",
988 "lgx-port-name": "LGX Back.4",
989 "lgx-port-rack": "000000.00",
990 "lgx-port-shelf": "00"
996 "service-rate": "100",
997 "node-id": "ROADM-C1",
998 "service-format": "OC",
999 "clli": "SNJSCAMCJT4",
1002 "port-device-name": "ROUTER_SNJSCAMCJT4_000000.00_00",
1003 "port-type": "router",
1004 "port-name": "Gigabit Ethernet_Tx.ge-1/0/0.0",
1005 "port-rack": "000000.00",
1009 "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00",
1010 "lgx-port-name": "LGX Back.29",
1011 "lgx-port-rack": "000000.00",
1012 "lgx-port-shelf": "00"
1017 "port-device-name": "ROUTER_SNJSCAMCJT4_000000.00_00",
1018 "port-type": "router",
1019 "port-name": "Gigabit Ethernet_Rx.ge-1/0/0.0",
1020 "port-rack": "000000.00",
1024 "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00",
1025 "lgx-port-name": "LGX Back.30",
1026 "lgx-port-rack": "000000.00",
1027 "lgx-port-shelf": "00"
1030 "optic-type": "gray"
1032 "due-date": "2016-11-28T00:00:01Z",
1033 "operator-contact": "pw1234"
1036 response = test_utils.post_request(url, data)
1037 self.assertEqual(response.status_code, requests.codes.ok)
1038 res = response.json()
1039 self.assertIn('PCE calculation in progress',
1040 res['output']['configuration-response-common']['response-message'])
1041 time.sleep(self.WAITING)
1043 def test_41_get_oc_service2(self):
1044 url = "{}/operational/org-openroadm-service:service-list/services/service2"
1045 response = test_utils.get_request(url)
1046 self.assertEqual(response.status_code, requests.codes.ok)
1047 res = response.json()
1049 res['services'][0]['administrative-state'],
1052 res['services'][0]['service-name'], 'service2')
1054 res['services'][0]['connection-type'], 'roadm-line')
1056 res['services'][0]['lifecycle-state'], 'planned')
1059 def test_42_check_xc2_ROADMA(self):
1060 response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/SRG1-PP2-TXRX-DEG2-TTP-TXRX-2")
1061 self.assertEqual(response.status_code, requests.codes.ok)
1062 res = response.json()
1063 # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
1064 self.assertDictEqual(
1066 'connection-name': 'SRG1-PP2-TXRX-DEG2-TTP-TXRX-2',
1067 'opticalControlMode': 'gainLoss',
1068 'target-output-power': -3.0
1069 }, **res['roadm-connections'][0]),
1070 res['roadm-connections'][0]
1072 self.assertDictEqual(
1073 {'src-if': 'SRG1-PP2-TXRX-nmc-2'},
1074 res['roadm-connections'][0]['source'])
1075 self.assertDictEqual(
1076 {'dst-if': 'DEG2-TTP-TXRX-nmc-2'},
1077 res['roadm-connections'][0]['destination'])
1080 def test_43_check_topo_ROADMA(self):
1081 self.test_26_check_topo_ROADMA_SRG1()
1082 self.test_27_check_topo_ROADMA_DEG2()
1085 def test_44_delete_oc_service1(self):
1086 url = "{}/operations/org-openroadm-service:service-delete"
1088 "sdnc-request-header": {
1089 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
1090 "rpc-action": "service-delete",
1091 "request-system-id": "appname",
1092 "notification-url": "http://localhost:8585/NotificationServer/notify"
1094 "service-delete-req-info": {
1095 "service-name": "service1",
1096 "tail-retention": "no"
1100 response = test_utils.post_request(url, data)
1101 self.assertEqual(response.status_code, requests.codes.ok)
1102 res = response.json()
1103 self.assertIn('Renderer service delete in progress',
1104 res['output']['configuration-response-common']['response-message'])
1107 def test_45_delete_oc_service2(self):
1108 url = "{}/operations/org-openroadm-service:service-delete"
1110 "sdnc-request-header": {
1111 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
1112 "rpc-action": "service-delete",
1113 "request-system-id": "appname",
1114 "notification-url": "http://localhost:8585/NotificationServer/notify"
1116 "service-delete-req-info": {
1117 "service-name": "service2",
1118 "tail-retention": "no"
1122 response = test_utils.post_request(url, data)
1123 self.assertEqual(response.status_code, requests.codes.ok)
1124 res = response.json()
1125 self.assertIn('Renderer service delete in progress',
1126 res['output']['configuration-response-common']['response-message'])
1129 def test_46_get_no_oc_services(self):
1131 url = "{}/operational/org-openroadm-service:service-list"
1132 response = test_utils.get_request(url)
1133 self.assertEqual(response.status_code, requests.codes.not_found)
1134 res = response.json()
1136 {"error-type": "application", "error-tag": "data-missing",
1137 "error-message": "Request could not be completed because the relevant data model content does not exist"},
1138 res['errors']['error'])
1141 def test_47_get_no_xc_ROADMA(self):
1142 response = test_utils.check_netconf_node_request("ROADM-A1", "")
1143 self.assertEqual(response.status_code, requests.codes.ok)
1144 res = response.json()
1145 self.assertNotIn(['roadm-connections'][0], res['org-openroadm-device'])
1148 def test_48_check_topo_ROADMA(self):
1149 self.test_34_check_topo_ROADMA_SRG1()
1150 self.test_35_check_topo_ROADMA_DEG2()
1152 def test_49_loop_create_eth_service(self):
1153 for i in range(1, 6):
1154 print("iteration number {}".format(i))
1155 print("eth service creation")
1156 self.test_11_create_eth_service1()
1157 print("check xc in ROADM-A1")
1158 self.test_13_check_xc1_ROADMA()
1159 print("check xc in ROADM-C1")
1160 self.test_14_check_xc1_ROADMC()
1161 print("eth service deletion\n")
1162 self.test_30_delete_eth_service1()
1164 def test_50_loop_create_oc_service(self):
1165 url = "{}/operational/org-openroadm-service:service-list/services/service1"
1166 response = test_utils.get_request(url)
1167 if response.status_code != 404:
1168 url = "{}/operations/org-openroadm-service:service-delete"
1170 "sdnc-request-header": {
1171 "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
1172 "rpc-action": "service-delete",
1173 "request-system-id": "appname",
1174 "notification-url": "http://localhost:8585/NotificationServer/notify"
1176 "service-delete-req-info": {
1177 "service-name": "service1",
1178 "tail-retention": "no"
1182 test_utils.post_request(url, data)
1185 for i in range(1, 6):
1186 print("iteration number {}".format(i))
1187 print("oc service creation")
1188 self.test_36_create_oc_service1()
1189 print("check xc in ROADM-A1")
1190 self.test_38_check_xc1_ROADMA()
1191 print("check xc in ROADM-C1")
1192 self.test_39_check_xc1_ROADMC()
1193 print("oc service deletion\n")
1194 self.test_44_delete_oc_service1()
1196 def test_51_disconnect_XPDRA(self):
1197 response = test_utils.unmount_device("XPDR-A1")
1198 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
1200 def test_52_disconnect_XPDRC(self):
1201 response = test_utils.unmount_device("XPDR-C1")
1202 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
1204 def test_53_disconnect_ROADMA(self):
1205 response = test_utils.unmount_device("ROADM-A1")
1206 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
1208 def test_54_disconnect_ROADMC(self):
1209 response = test_utils.unmount_device("ROADM-C1")
1210 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
1213 if __name__ == "__main__":
1214 unittest.main(verbosity=2)