run autopep8 on functional tests
[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         response = test_utils.put_xmlrequest(test_utils.URL_CONFIG_ORDM_TOPO, self.simple_topo_bi_dir_data)
74         self.assertEqual(response.status_code, requests.codes.ok)
75         time.sleep(2)
76
77     # Get existing nodeId
78     def test_02_get_nodeId(self):
79         response = test_utils.get_ordm_topo_request("node/ROADMA01-SRG1")
80         self.assertEqual(response.status_code, requests.codes.ok)
81         res = response.json()
82         self.assertEqual(
83             res['node'][0]['node-id'], 'ROADMA01-SRG1')
84         time.sleep(1)
85
86     # Get existing linkId
87     def test_03_get_linkId(self):
88         response = test_utils.get_ordm_topo_request("link/XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX")
89         self.assertEqual(response.status_code, requests.codes.ok)
90         res = response.json()
91         self.assertEqual(
92             res['ietf-network-topology:link'][0]['link-id'],
93             'XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX')
94         time.sleep(1)
95
96     # Path Computation success
97     def test_04_path_computation_xpdr_bi(self):
98         response = test_utils.path_computation_request("request-1", "service-1",
99                                                        {"node-id": "XPDRA01", "service-rate": "100",
100                                                            "service-format": "Ethernet", "clli": "nodeA"},
101                                                        {"node-id": "XPDRC01", "service-rate": "100", "service-format": "Ethernet", "clli": "nodeC"})
102         self.assertEqual(response.status_code, requests.codes.ok)
103         res = response.json()
104         self.assertIn('Path is calculated',
105                       res['output']['configuration-response-common']['response-message'])
106         time.sleep(5)
107
108     # Path Computation success
109     def test_05_path_computation_rdm_bi(self):
110         response = test_utils.path_computation_request("request-1", "service-1",
111                                                        {"node-id": "ROADMA01", "service-rate": "100",
112                                                            "service-format": "Ethernet", "clli": "NodeA"},
113                                                        {"node-id": "ROADMC01", "service-rate": "100", "service-format": "Ethernet", "clli": "NodeC"})
114         self.assertEqual(response.status_code, requests.codes.ok)
115         res = response.json()
116         self.assertIn('Path is calculated',
117                       res['output']['configuration-response-common']['response-message'])
118         time.sleep(5)
119
120     # Delete topology
121     def test_06_delete_simple_topology_bi(self):
122         response = test_utils.delete_request(test_utils.URL_CONFIG_ORDM_TOPO)
123         self.assertEqual(response.status_code, requests.codes.ok)
124         time.sleep(2)
125
126     # Test deleted topology
127     def test_07_test_topology_simple_bi_deleted(self):
128         response = test_utils.get_ordm_topo_request("node/ROADMA01-SRG1")
129         self.assertEqual(response.status_code, requests.codes.conflict)
130         time.sleep(1)
131
132     # Load simple bidirectional topology
133     def test_08_load_simple_topology_uni(self):
134         response = test_utils.put_xmlrequest(test_utils.URL_CONFIG_ORDM_TOPO, self.simple_topo_uni_dir_data)
135         self.assertEqual(response.status_code, 201)
136         time.sleep(2)
137
138     # Get existing nodeId
139     def test_09_get_nodeId(self):
140         response = test_utils.get_ordm_topo_request("node/XPONDER-1-2")
141         self.assertEqual(response.status_code, requests.codes.ok)
142         res = response.json()
143         self.assertEqual(
144             res['node'][0]['node-id'],
145             'XPONDER-1-2')
146         time.sleep(1)
147
148     # Get existing linkId
149     def test_10_get_linkId(self):
150         response = test_utils.get_ordm_topo_request("link/XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX")
151         self.assertEqual(response.status_code, requests.codes.ok)
152         res = response.json()
153         self.assertEqual(
154             res['ietf-network-topology:link'][0]['link-id'],
155             'XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX')
156         time.sleep(1)
157
158     # Path Computation success
159     def test_11_path_computation_xpdr_uni(self):
160         response = test_utils.path_computation_request("request-1", "service-1",
161                                                        {"node-id": "XPONDER-1-2", "service-rate": "100",
162                                                            "service-format": "Ethernet", "clli": "ORANGE1"},
163                                                        {"node-id": "XPONDER-3-2", "service-rate": "100", "service-format": "Ethernet", "clli": "ORANGE3"})
164         self.assertEqual(response.status_code, requests.codes.ok)
165         res = response.json()
166         self.assertIn('Path is calculated',
167                       res['output']['configuration-response-common']['response-message'])
168         time.sleep(5)
169
170     # Path Computation success
171     def test_12_path_computation_rdm_uni(self):
172         response = test_utils.path_computation_request("request1", "service1",
173                                                        {"service-rate": "100", "service-format": "Ethernet",
174                                                            "clli": "cll21", "node-id": "OpenROADM-2-1"},
175                                                        {"service-rate": "100", "service-format": "Ethernet", "clli": "ncli22", "node-id": "OpenROADM-2-2"})
176         self.assertEqual(response.status_code, requests.codes.ok)
177         res = response.json()
178         self.assertIn('Path is calculated',
179                       res['output']['configuration-response-common']['response-message'])
180         # ZtoA path test
181         atozList = len(res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'])
182         ztoaList = len(res['output']['response-parameters']['path-description']['zToA-direction']['zToA'])
183         self.assertEqual(atozList, 15)
184         self.assertEqual(ztoaList, 15)
185         for i in range(0, 15):
186             atoz = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]
187             ztoa = res['output']['response-parameters']['path-description']['zToA-direction']['zToA'][i]
188             if (atoz['id'] == '14'):
189                 self.assertEqual(atoz['resource']['tp-id'], 'SRG1-PP1-TX')
190             if (ztoa['id'] == '0'):
191                 self.assertEqual(ztoa['resource']['tp-id'], 'SRG1-PP1-RX')
192         time.sleep(5)
193
194     # Delete topology
195     def test_13_delete_simple_topology(self):
196         response = test_utils.delete_request(test_utils.URL_CONFIG_ORDM_TOPO)
197         self.assertEqual(response.status_code, requests.codes.ok)
198         time.sleep(2)
199
200     # Test deleted topology
201     def test_14_test_topology_simple_deleted(self):
202         response = test_utils.get_ordm_topo_request("node/XPONDER-1-2")
203         self.assertEqual(response.status_code, requests.codes.conflict)
204         time.sleep(1)
205
206     # Load complex topology
207     def test_15_load_complex_topology(self):
208         response = test_utils.put_xmlrequest(test_utils.URL_CONFIG_ORDM_TOPO, self.complex_topo_uni_dir_data)
209         self.assertEqual(response.status_code, 201)
210         time.sleep(2)
211
212     # Get existing nodeId
213     def test_16_get_nodeId(self):
214         response = test_utils.get_ordm_topo_request("node/XPONDER-3-2")
215         self.assertEqual(response.status_code, requests.codes.ok)
216         res = response.json()
217         self.assertEqual(
218             res['node'][0]['node-id'],
219             'XPONDER-3-2')
220         time.sleep(1)
221
222     # Test failed path computation
223     def test_17_fail_path_computation(self):
224         response = test_utils.post_request(test_utils.URL_PATH_COMPUTATION_REQUEST,
225                                            {"input": {"service-handler-header": {"request-id": "request-1"}}})
226         self.assertEqual(response.status_code, requests.codes.ok)
227         res = response.json()
228         self.assertIn('Service Name is not set',
229                       res['output']['configuration-response-common']['response-message'])
230         time.sleep(2)
231
232     # Test1 success path computation
233     def test_18_success1_path_computation(self):
234         response = test_utils.path_computation_request("request1", "service1",
235                                                        {"service-format": "Ethernet", "service-rate": "100", "clli": "ORANGE2", "node-id": "XPONDER-2-2",
236                                                         "tx-direction": {"port": {
237                                                             "port-device-name": "Some port-device-name",
238                                                             "port-type": "Some port-type",
239                                                             "port-name": "Some port-name",
240                                                             "port-rack": "Some port-rack",
241                                                             "port-shelf": "Some port-shelf",
242                                                             "port-slot": "Some port-slot",
243                                                             "port-sub-slot": "Some port-sub-slot"
244                                                         }},
245                                                            "rx-direction": {"port": {
246                                                                "port-device-name": "Some port-device-name",
247                                                                "port-type": "Some port-type",
248                                                                "port-name": "Some port-name",
249                                                                "port-rack": "Some port-rack",
250                                                                "port-shelf": "Some port-shelf",
251                                                                "port-slot": "Some port-slot",
252                                                                "port-sub-slot": "Some port-sub-slot"
253                                                            }}},
254                                                        {"service-format": "Ethernet", "service-rate": "100", "clli": "ORANGE1", "node-id": "XPONDER-1-2",
255                                                            "tx-direction": {"port": {
256                                                                "port-device-name": "Some port-device-name",
257                                                                "port-type": "Some port-type",
258                                                                "port-name": "Some port-name",
259                                                                "port-rack": "Some port-rack",
260                                                                "port-shelf": "Some port-shelf",
261                                                                "port-slot": "Some port-slot",
262                                                                "port-sub-slot": "Some port-sub-slot"
263                                                            }},
264                                                            "rx-direction": {"port": {
265                                                                "port-device-name": "Some port-device-name",
266                                                                "port-type": "Some port-type",
267                                                                "port-name": "Some port-name",
268                                                                "port-rack": "Some port-rack",
269                                                                "port-shelf": "Some port-shelf",
270                                                                "port-slot": "Some port-slot",
271                                                                "port-sub-slot": "Some port-sub-slot"
272                                                            }}},
273                                                        {"customer-code": ["Some customer-code"],
274                                                            "co-routing": {"existing-service": ["Some existing-service"]}},
275                                                        {"customer-code": ["Some customer-code"],
276                                                            "co-routing": {"existing-service": ["Some existing-service"]}},
277                                                        "hop-count", {"locally-protected-links": "true"})
278         self.assertEqual(response.status_code, requests.codes.ok)
279         res = response.json()
280         self.assertIn('Path is calculated',
281                       res['output']['configuration-response-common']['response-message'])
282         time.sleep(5)
283
284     # Test2 success path computation with path description
285     def test_19_success2_path_computation(self):
286         response = test_utils.path_computation_request("request 1", "service 1",
287                                                        {"service-rate": "100", "service-format": "Ethernet",
288                                                            "node-id": "XPONDER-1-2", "clli": "ORANGE1"},
289                                                        {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-3-2", "clli": "ORANGE3"})
290         self.assertEqual(response.status_code, requests.codes.ok)
291         res = response.json()
292         self.assertIn('Path is calculated',
293                       res['output']['configuration-response-common']['response-message'])
294         self.assertEqual(5, res['output']['response-parameters']['path-description']
295                          ['aToZ-direction']['aToZ-wavelength-number'])
296         self.assertEqual(5, res['output']['response-parameters']['path-description']
297                          ['zToA-direction']['zToA-wavelength-number'])
298         time.sleep(5)
299
300     # Test3 success path computation with hard-constraints exclude
301     def test_20_success3_path_computation(self):
302         response = test_utils.path_computation_request("request 1", "service 1",
303                                                        {"service-rate": "100", "service-format": "Ethernet",
304                                                            "node-id": "XPONDER-1-2", "clli": "ORANGE1"},
305                                                        {"service-rate": "100", "service-format": "Ethernet",
306                                                            "node-id": "XPONDER-3-2", "clli": "ORANGE3"},
307                                                        {"exclude_": {"node-id": ["OpenROADM-2-1", "OpenROADM-2-2"]}})
308         self.assertEqual(response.status_code, requests.codes.ok)
309         res = response.json()
310         self.assertIn('Path is calculated',
311                       res['output']['configuration-response-common']['response-message'])
312         self.assertEqual(9, res['output']['response-parameters']['path-description']
313                          ['aToZ-direction']['aToZ-wavelength-number'])
314         self.assertEqual(9, res['output']['response-parameters']['path-description']
315                          ['zToA-direction']['zToA-wavelength-number'])
316         time.sleep(5)
317
318     # Path computation before deleting oms-attribute of the link :openroadm1-3 to openroadm1-2
319     def test_21_path_computation_before_oms_attribute_deletion(self):
320         response = test_utils.path_computation_request("request 1", "service 1",
321                                                        {"service-rate": "100", "service-format": "Ethernet",
322                                                            "node-id": "XPONDER-2-2", "clli": "ORANGE2"},
323                                                        {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-1-2", "clli": "ORANGE1"})
324         self.assertEqual(response.status_code, requests.codes.ok)
325         res = response.json()
326         self.assertIn('Path is calculated',
327                       res['output']['configuration-response-common']['response-message'])
328         nbElmPath = len(res['output']['response-parameters']['path-description']
329                         ['aToZ-direction']['aToZ'])
330         self.assertEqual(31, nbElmPath)
331         link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2"}
332         find = False
333         for i in range(0, nbElmPath):
334             resource_i = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]['resource']
335             if(resource_i == link):
336                 find = True
337         self.assertEqual(find, True)
338         time.sleep(5)
339
340     # Delete oms-attribute in the link :openroadm1-3 to openroadm1-2
341     def test_22_delete_oms_attribute_in_openroadm13toopenroadm12_link(self):
342         response = test_utils.del_oms_attr_request("OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2")
343         self.assertEqual(response.status_code, requests.codes.ok)
344         time.sleep(2)
345
346     # Path computation after deleting oms-attribute of the link :openroadm1-3 to openroadm1-2
347     def test_23_path_computation_after_oms_attribute_deletion(self):
348         response = test_utils.path_computation_request("request 1", "service 1",
349                                                        {"service-rate": "100", "service-format": "Ethernet",
350                                                            "node-id": "XPONDER-2-2", "clli": "ORANGE2"},
351                                                        {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-1-2", "clli": "ORANGE1"})
352         self.assertEqual(response.status_code, requests.codes.ok)
353         res = response.json()
354         self.assertIn('Path is calculated',
355                       res['output']['configuration-response-common']['response-message'])
356         nbElmPath = len(res['output']['response-parameters']['path-description']
357                         ['aToZ-direction']['aToZ'])
358         self.assertEqual(47, nbElmPath)
359         link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2"}
360         find = False
361         for i in range(0, nbElmPath):
362             resource_i = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]['resource']
363             if (resource_i == link):
364                 find = True
365         self.assertNotEqual(find, True)
366         time.sleep(5)
367
368     # Delete complex topology
369     def test_24_delete_complex_topology(self):
370         response = test_utils.delete_request(test_utils.URL_CONFIG_ORDM_TOPO)
371         self.assertEqual(response.status_code, requests.codes.ok)
372         time.sleep(2)
373
374     # Test deleted complex topology
375     def test_25_test_topology_complex_deleted(self):
376         response = test_utils.get_ordm_topo_request("node/XPONDER-3-2")
377         self.assertEqual(response.status_code, requests.codes.conflict)
378         time.sleep(1)
379
380
381 if __name__ == "__main__":
382     unittest.main(verbosity=2)