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