a1b35323b03b679e47eb5b6d7321ebd9f8bc2609
[transportpce.git] / tests / transportpce_tests / 1.2.1 / test_pce.py
1 #!/usr/bin/env python
2 ##############################################################################
3 # Copyright (c) 2017 Orange, Inc. and others.  All rights reserved.
4 #
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 ##############################################################################
10
11 import unittest
12 import json
13 import os
14 import time
15 import requests
16 from common import test_utils
17
18
19 class TransportPCEtesting(unittest.TestCase):
20
21     simple_topo_bi_dir_data = None
22     simple_topo_uni_dir_data = None
23     complex_topo_uni_dir_data = None
24
25     @classmethod
26     def _get_file(cls):
27         topo_bi_dir_file = os.path.join(os.path.dirname(os.path.realpath(__file__)),
28                                         "..", "..", "sample_configs", "honeynode-topo.xml")
29         if os.path.isfile(topo_bi_dir_file):
30             with open(topo_bi_dir_file, 'r') as topo_bi_dir:
31                 cls.simple_topo_bi_dir_data = topo_bi_dir.read()
32         topo_uni_dir_file = os.path.join(os.path.dirname(os.path.realpath(__file__)),
33                                          "..", "..", "sample_configs", "NW-simple-topology.xml")
34         if os.path.isfile(topo_uni_dir_file):
35             with open(topo_uni_dir_file, 'r') as topo_uni_dir:
36                 cls.simple_topo_uni_dir_data = topo_uni_dir.read()
37         topo_uni_dir_complex_file = os.path.join(os.path.dirname(os.path.realpath(__file__)),
38                                                  "..", "..", "sample_configs", "NW-for-test-5-4.xml")
39         if os.path.isfile(topo_uni_dir_complex_file):
40             with open(topo_uni_dir_complex_file, 'r') as topo_uni_dir_complex:
41                 cls.complex_topo_uni_dir_data = topo_uni_dir_complex.read()
42
43     processes = None
44
45     @classmethod
46     def setUpClass(cls):
47         cls._get_file()
48         cls.processes = test_utils.start_tpce()
49
50     @classmethod
51     def tearDownClass(cls):
52         for process in cls.processes:
53             test_utils.shutdown_process(process)
54         print("all processes killed")
55
56     def setUp(self):  # instruction executed before each test method
57         time.sleep(1)
58
59      # Load simple bidirectional topology
60     def test_01_load_simple_topology_bi(self):
61         url = "{}/config/ietf-network:networks/network/openroadm-topology"
62         data = self.simple_topo_bi_dir_data
63         response = test_utils.put_xmlrequest(url, data)
64         self.assertEqual(response.status_code, requests.codes.ok)
65         time.sleep(2)
66
67     # Get existing nodeId
68     def test_02_get_nodeId(self):
69         url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-SRG1"
70         response = test_utils.get_request(url)
71         self.assertEqual(response.status_code, requests.codes.ok)
72         res = response.json()
73         self.assertEqual(
74             res['node'][0]['node-id'], 'ROADMA01-SRG1')
75         time.sleep(1)
76
77     # Get existing linkId
78     def test_03_get_linkId(self):
79         url = "{}/config/ietf-network:networks/network/openroadm-topology/link/XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX"
80         response = test_utils.get_request(url)
81         self.assertEqual(response.status_code, requests.codes.ok)
82         res = response.json()
83         self.assertEqual(
84             res['ietf-network-topology:link'][0]['link-id'],
85             'XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX')
86         time.sleep(1)
87
88     # Path Computation success
89     def test_04_path_computation_xpdr_bi(self):
90         url = "{}/operations/transportpce-pce:path-computation-request"
91         data = {"input": {
92                 "service-name": "service-1",
93                 "resource-reserve": "true",
94                 "pce-metric": "hop-count",
95                 "service-handler-header": {
96                     "request-id": "request-1"
97                 },
98                 "service-a-end": {
99                     "node-id": "XPDRA01",
100                     "service-rate": "100",
101                     "service-format": "Ethernet",
102                     "clli": "nodeA"
103                 },
104                 "service-z-end": {
105                     "node-id": "XPDRC01",
106                     "service-rate": "100",
107                     "service-format": "Ethernet",
108                     "clli": "nodeC"
109                 }
110                 }
111                 }
112         response = test_utils.post_request(url, data)
113         self.assertEqual(response.status_code, requests.codes.ok)
114         res = response.json()
115         self.assertIn('Path is calculated',
116                       res['output']['configuration-response-common']['response-message'])
117         time.sleep(5)
118
119     # Path Computation success
120     def test_05_path_computation_rdm_bi(self):
121         url = "{}/operations/transportpce-pce:path-computation-request"
122         data = {"input": {
123                 "service-name": "service-1",
124                 "resource-reserve": "true",
125                 "pce-metric": "hop-count",
126                 "service-handler-header": {
127                     "request-id": "request-1"
128                 },
129                 "service-a-end": {
130                     "node-id": "ROADMA01",
131                     "service-rate": "100",
132                     "service-format": "Ethernet",
133                     "clli": "NodeA"
134                 },
135                 "service-z-end": {
136                     "node-id": "ROADMC01",
137                     "service-rate": "100",
138                     "service-format": "Ethernet",
139                     "clli": "NodeC"
140                 }
141                 }
142                 }
143         response = test_utils.post_request(url, data)
144         self.assertEqual(response.status_code, requests.codes.ok)
145         res = response.json()
146         self.assertIn('Path is calculated',
147                       res['output']['configuration-response-common']['response-message'])
148         time.sleep(5)
149
150     # Delete topology
151     def test_06_delete_simple_topology_bi(self):
152         url = "{}/config/ietf-network:networks/network/openroadm-topology"
153         response = test_utils.delete_request(url)
154         self.assertEqual(response.status_code, requests.codes.ok)
155         time.sleep(2)
156
157     # Test deleted topology
158     def test_07_test_topology_simple_bi_deleted(self):
159         url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-SRG1"
160         response = test_utils.get_request(url)
161         self.assertEqual(response.status_code, 404)
162         time.sleep(1)
163
164     # Load simple bidirectional topology
165     def test_08_load_simple_topology_uni(self):
166         url = "{}/config/ietf-network:networks/network/openroadm-topology"
167         data = self.simple_topo_uni_dir_data
168         response = test_utils.put_xmlrequest(url, data)
169         self.assertEqual(response.status_code, 201)
170         time.sleep(2)
171
172     # Get existing nodeId
173     def test_09_get_nodeId(self):
174         url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-1-2"
175         response = test_utils.get_request(url)
176         self.assertEqual(response.status_code, requests.codes.ok)
177         res = response.json()
178         self.assertEqual(
179             res['node'][0]['node-id'],
180             'XPONDER-1-2')
181         time.sleep(1)
182
183     # Get existing linkId
184     def test_10_get_linkId(self):
185         url = "{}/config/ietf-network:networks/network/openroadm-topology/link/XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX"
186         response = test_utils.get_request(url)
187         self.assertEqual(response.status_code, requests.codes.ok)
188         res = response.json()
189         self.assertEqual(
190             res['ietf-network-topology:link'][0]['link-id'],
191             'XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX')
192         time.sleep(1)
193
194     # Path Computation success
195     def test_11_path_computation_xpdr_uni(self):
196         url = "{}/operations/transportpce-pce:path-computation-request"
197         data = {"input": {
198                 "service-name": "service-1",
199                 "resource-reserve": "true",
200                 "pce-metric": "hop-count",
201                 "service-handler-header": {
202                     "request-id": "request-1"
203                 },
204                 "service-a-end": {
205                     "node-id": "XPONDER-1-2",
206                     "service-rate": "100",
207                     "service-format": "Ethernet",
208                     "clli": "ORANGE1"
209                 },
210                 "service-z-end": {
211                     "node-id": "XPONDER-3-2",
212                     "service-rate": "100",
213                     "service-format": "Ethernet",
214                     "clli": "ORANGE3"
215                 }
216                 }
217                 }
218         response = test_utils.post_request(url, data)
219         self.assertEqual(response.status_code, requests.codes.ok)
220         res = response.json()
221         self.assertIn('Path is calculated',
222                       res['output']['configuration-response-common']['response-message'])
223         time.sleep(5)
224
225     # Path Computation success
226     def test_12_path_computation_rdm_uni(self):
227         url = "{}/operations/transportpce-pce:path-computation-request"
228         data = {"input": {
229                 "service-name": "service1",
230                 "resource-reserve": "true",
231                 "service-handler-header": {
232                     "request-id": "request1"
233                 },
234                 "service-a-end": {
235                     "service-rate": "100",
236                     "service-format": "Ethernet",
237                     "clli": "cll21",
238                     "node-id": "OpenROADM-2-1"
239                 },
240                 "service-z-end": {
241                     "service-rate": "100",
242                     "service-format": "Ethernet",
243                     "clli": "ncli22",
244                     "node-id": "OpenROADM-2-2"
245                 },
246                 "pce-metric": "hop-count"
247                 }
248                 }
249         response = test_utils.post_request(url, data)
250         self.assertEqual(response.status_code, requests.codes.ok)
251         res = response.json()
252         self.assertIn('Path is calculated',
253                       res['output']['configuration-response-common']['response-message'])
254         # ZtoA path test
255         atozList = len(res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'])
256         ztoaList = len(res['output']['response-parameters']['path-description']['zToA-direction']['zToA'])
257         self.assertEqual(atozList, 15)
258         self.assertEqual(ztoaList, 15)
259         for i in range(0, 15):
260             atoz = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]
261             ztoa = res['output']['response-parameters']['path-description']['zToA-direction']['zToA'][i]
262             if (atoz['id'] == '14'):
263                 self.assertEqual(atoz['resource']['tp-id'], 'SRG1-PP1-TX')
264             if (ztoa['id'] == '0'):
265                 self.assertEqual(ztoa['resource']['tp-id'], 'SRG1-PP1-RX')
266         time.sleep(5)
267
268     # Delete topology
269     def test_13_delete_simple_topology(self):
270         url = "{}/config/ietf-network:networks/network/openroadm-topology"
271         response = test_utils.delete_request(url)
272         self.assertEqual(response.status_code, requests.codes.ok)
273         time.sleep(2)
274
275     # Test deleted topology
276     def test_14_test_topology_simple_deleted(self):
277         url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-1-2"
278         response = test_utils.get_request(url)
279         self.assertEqual(response.status_code, 404)
280         time.sleep(1)
281
282     # Load complex topology
283     def test_15_load_complex_topology(self):
284         url = "{}/config/ietf-network:networks/network/openroadm-topology"
285         data = self.complex_topo_uni_dir_data
286         response = test_utils.put_xmlrequest(url, data)
287         self.assertEqual(response.status_code, 201)
288         time.sleep(2)
289
290     # Get existing nodeId
291     def test_16_get_nodeId(self):
292         url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-3-2"
293         response = test_utils.get_request(url)
294         self.assertEqual(response.status_code, requests.codes.ok)
295         res = response.json()
296         self.assertEqual(
297             res['node'][0]['node-id'],
298             'XPONDER-3-2')
299         time.sleep(1)
300
301     # Test failed path computation
302     def test_17_fail_path_computation(self):
303         url = "{}/operations/transportpce-pce:path-computation-request"
304         data = {"input": {
305                 "service-handler-header": {
306                     "request-id": "request-1"
307                 }
308                 }
309                 }
310         response = test_utils.post_request(url, data)
311         self.assertEqual(response.status_code, requests.codes.ok)
312         res = response.json()
313         self.assertIn('Service Name is not set',
314                       res['output']['configuration-response-common']['response-message'])
315         time.sleep(2)
316
317     # Test1 success path computation
318     def test_18_success1_path_computation(self):
319         url = "{}/operations/transportpce-pce:path-computation-request"
320         data = {"input": {
321                 "service-name": "service1",
322                 "resource-reserve": "true",
323                 "service-handler-header": {
324                     "request-id": "request1"
325                 },
326                 "service-a-end": {
327                     "service-format": "Ethernet",
328                     "service-rate": "100",
329                     "clli": "ORANGE2",
330                     "node-id": "XPONDER-2-2",
331                     "tx-direction": {
332                         "port": {
333                             "port-device-name": "Some port-device-name",
334                             "port-type": "Some port-type",
335                             "port-name": "Some port-name",
336                             "port-rack": "Some port-rack",
337                             "port-shelf": "Some port-shelf",
338                             "port-slot": "Some port-slot",
339                             "port-sub-slot": "Some port-sub-slot"
340                         }
341                     },
342                     "rx-direction": {
343                         "port": {
344                             "port-device-name": "Some port-device-name",
345                             "port-type": "Some port-type",
346                             "port-name": "Some port-name",
347                             "port-rack": "Some port-rack",
348                             "port-shelf": "Some port-shelf",
349                             "port-slot": "Some port-slot",
350                             "port-sub-slot": "Some port-sub-slot"
351                         }
352                     }
353                 },
354                 "service-z-end": {
355                     "service-format": "Ethernet",
356                     "service-rate": "100",
357                     "clli": "ORANGE1",
358                     "node-id": "XPONDER-1-2",
359                     "tx-direction": {
360                         "port": {
361                             "port-device-name": "Some port-device-name",
362                             "port-type": "Some port-type",
363                             "port-name": "Some port-name",
364                             "port-rack": "Some port-rack",
365                             "port-shelf": "Some port-shelf",
366                             "port-slot": "Some port-slot",
367                             "port-sub-slot": "Some port-sub-slot"
368                         }
369                     },
370                     "rx-direction": {
371                         "port": {
372                             "port-device-name": "Some port-device-name",
373                             "port-type": "Some port-type",
374                             "port-name": "Some port-name",
375                             "port-rack": "Some port-rack",
376                             "port-shelf": "Some port-shelf",
377                             "port-slot": "Some port-slot",
378                             "port-sub-slot": "Some port-sub-slot"
379                         }
380                     }
381                 },
382                 "hard-constraints": {
383                     "customer-code": [
384                         "Some customer-code"
385                     ],
386                     "co-routing": {
387                         "existing-service": [
388                             "Some existing-service"
389                         ]
390                     }
391                 },
392                 "soft-constraints": {
393                     "customer-code": [
394                         "Some customer-code"
395                     ],
396                     "co-routing": {
397                         "existing-service": [
398                             "Some existing-service"
399                         ]
400                     }
401                 },
402                 "pce-metric": "hop-count",
403                 "locally-protected-links": "true"
404                 }
405                 }
406         response = test_utils.post_request(url, data)
407         self.assertEqual(response.status_code, requests.codes.ok)
408         res = response.json()
409         self.assertIn('Path is calculated',
410                       res['output']['configuration-response-common']['response-message'])
411         time.sleep(5)
412
413     # Test2 success path computation with path description
414     def test_19_success2_path_computation(self):
415         url = "{}/operations/transportpce-pce:path-computation-request"
416         data = {"input": {
417                 "service-name": "service 1",
418                 "resource-reserve": "true",
419                 "service-handler-header": {
420                     "request-id": "request 1"
421                 },
422                 "service-a-end": {
423                     "service-rate": "100",
424                     "service-format": "Ethernet",
425                     "node-id": "XPONDER-1-2",
426                     "clli": "ORANGE1"
427                 },
428                 "service-z-end": {
429                     "service-rate": "100",
430                     "service-format": "Ethernet",
431                     "node-id": "XPONDER-3-2",
432                     "clli": "ORANGE3"
433                 },
434                 "pce-metric": "hop-count"
435                 }
436                 }
437         response = test_utils.post_request(url, data)
438         self.assertEqual(response.status_code, requests.codes.ok)
439         res = response.json()
440         self.assertIn('Path is calculated',
441                       res['output']['configuration-response-common']['response-message'])
442         self.assertEqual(5, res['output']['response-parameters']['path-description']
443                          ['aToZ-direction']['aToZ-wavelength-number'])
444         self.assertEqual(5, res['output']['response-parameters']['path-description']
445                          ['zToA-direction']['zToA-wavelength-number'])
446         time.sleep(5)
447
448     # Test3 success path computation with hard-constraints exclude
449     def test_20_success3_path_computation(self):
450         url = "{}/operations/transportpce-pce:path-computation-request"
451         data = {"input": {
452                 "service-name": "service 1",
453                 "resource-reserve": "true",
454                 "service-handler-header": {
455                     "request-id": "request 1"
456                 },
457                 "service-a-end": {
458                     "service-rate": "100",
459                     "service-format": "Ethernet",
460                     "node-id": "XPONDER-1-2",
461                     "clli": "ORANGE1"
462                 },
463                 "service-z-end": {
464                     "service-rate": "100",
465                     "service-format": "Ethernet",
466                     "node-id": "XPONDER-3-2",
467                     "clli": "ORANGE3"
468                 },
469                 "hard-constraints": {
470                     "exclude_": {
471                         "node-id": ["OpenROADM-2-1", "OpenROADM-2-2"]
472                     }
473                 },
474                 "pce-metric": "hop-count"
475                 }
476                 }
477         response = test_utils.post_request(url, data)
478         self.assertEqual(response.status_code, requests.codes.ok)
479         res = response.json()
480         self.assertIn('Path is calculated',
481                       res['output']['configuration-response-common']['response-message'])
482         self.assertEqual(9, res['output']['response-parameters']['path-description']
483                          ['aToZ-direction']['aToZ-wavelength-number'])
484         self.assertEqual(9, res['output']['response-parameters']['path-description']
485                          ['zToA-direction']['zToA-wavelength-number'])
486         time.sleep(5)
487
488     # Path computation before deleting oms-attribute of the link :openroadm1-3 to openroadm1-2
489     def test_21_path_computation_before_oms_attribute_deletion(self):
490         url = "{}/operations/transportpce-pce:path-computation-request"
491         data = {"input": {
492                 "service-name": "service 1",
493                 "resource-reserve": "true",
494                 "service-handler-header": {
495                     "request-id": "request 1"
496                 },
497                 "service-a-end": {
498                     "service-rate": "100",
499                     "service-format": "Ethernet",
500                     "node-id": "XPONDER-2-2",
501                     "clli": "ORANGE2"
502                 },
503                 "service-z-end": {
504                     "service-rate": "100",
505                     "service-format": "Ethernet",
506                     "node-id": "XPONDER-1-2",
507                     "clli": "ORANGE1"
508                 },
509                 "pce-metric": "hop-count"
510                 }
511                 }
512         response = test_utils.post_request(url, data)
513         self.assertEqual(response.status_code, requests.codes.ok)
514         res = response.json()
515         self.assertIn('Path is calculated',
516                       res['output']['configuration-response-common']['response-message'])
517         nbElmPath = len(res['output']['response-parameters']['path-description']
518                         ['aToZ-direction']['aToZ'])
519         self.assertEqual(31, nbElmPath)
520         link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2"}
521         find = False
522         for i in range(0, nbElmPath):
523             resource_i = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]['resource']
524             if(resource_i == link):
525                 find = True
526         self.assertEqual(find, True)
527         time.sleep(5)
528
529     # Delete oms-attribute in the link :openroadm1-3 to openroadm1-2
530     def test_22_delete_oms_attribute_in_openroadm13toopenroadm12_link(self):
531         url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:link/"
532                "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2/org-openroadm-network-topology:OMS-attributes/span"
533                )
534         response = test_utils.delete_request(url)
535         self.assertEqual(response.status_code, requests.codes.ok)
536         time.sleep(2)
537
538     # Path computation after deleting oms-attribute of the link :openroadm1-3 to openroadm1-2
539     def test_23_path_computation_after_oms_attribute_deletion(self):
540         url = "{}/operations/transportpce-pce:path-computation-request"
541         data = {"input": {
542                 "service-name": "service 1",
543                 "resource-reserve": "true",
544                 "service-handler-header": {
545                     "request-id": "request 1"
546                 },
547                 "service-a-end": {
548                     "service-rate": "100",
549                     "service-format": "Ethernet",
550                     "node-id": "XPONDER-2-2",
551                     "clli": "ORANGE2"
552                 },
553                 "service-z-end": {
554                     "service-rate": "100",
555                     "service-format": "Ethernet",
556                     "node-id": "XPONDER-1-2",
557                     "clli": "ORANGE1"
558                 },
559                 "pce-metric": "hop-count"
560                 }
561                 }
562         response = test_utils.post_request(url, data)
563         self.assertEqual(response.status_code, requests.codes.ok)
564         res = response.json()
565         self.assertIn('Path is calculated',
566                       res['output']['configuration-response-common']['response-message'])
567         nbElmPath = len(res['output']['response-parameters']['path-description']
568                         ['aToZ-direction']['aToZ'])
569         self.assertEqual(47, nbElmPath)
570         link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2"}
571         find = False
572         for i in range(0, nbElmPath):
573             resource_i = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]['resource']
574             if (resource_i == link):
575                 find = True
576         self.assertNotEqual(find, True)
577         time.sleep(5)
578
579     # Delete complex topology
580     def test_24_delete_complex_topology(self):
581         url = "{}/config/ietf-network:networks/network/openroadm-topology"
582         response = test_utils.delete_request(url)
583         self.assertEqual(response.status_code, requests.codes.ok)
584         time.sleep(2)
585
586     # Test deleted complex topology
587     def test_25_test_topology_complex_deleted(self):
588         url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-3-2"
589         response = test_utils.get_request(url)
590         self.assertEqual(response.status_code, 404)
591         time.sleep(1)
592
593
594 if __name__ == "__main__":
595     unittest.main(verbosity=2)