2 ##############################################################################
3 # Copyright (c) 2021 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 ##############################################################################
11 # pylint: disable=invalid-name
12 # pylint: disable=no-member
13 # pylint: disable=too-many-public-methods
17 # pylint: disable=wrong-import-order
21 sys.path.append('transportpce_tests/common/')
22 # pylint: disable=wrong-import-position
23 # pylint: disable=import-error
24 import test_utils # nopep8
27 class TransportPCE400Gtesting(unittest.TestCase):
28 path_computation_input_data = {
29 "service-name": "service-1",
30 "resource-reserve": "true",
31 "service-handler-header": {
32 "request-id": "request1"
35 "service-rate": "400",
37 "service-format": "Ethernet",
41 "service-rate": "400",
43 "service-format": "Ethernet",
46 "pce-routing-metric": "hop-count"
49 simple_topo_bi_dir_data = None
50 port_mapping_data = None
55 # pylint: disable=bare-except
56 sample_files_parsed = False
58 TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
59 "..", "..", "sample_configs",
60 "honeynode-topo400G.json")
61 with open(TOPO_BI_DIR_FILE, 'r', encoding='utf-8') as topo_bi_dir:
62 cls.topo_bi_dir_data = topo_bi_dir.read()
64 OTN_TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
65 "..", "..", "sample_configs",
66 "honeynode-otntopo400G.json")
67 with open(OTN_TOPO_BI_DIR_FILE, 'r', encoding='utf-8') as otn_topo_bi_dir:
68 cls.otn_topo_bi_dir_data = otn_topo_bi_dir.read()
70 OTUC4_OTN_TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
71 "..", "..", "sample_configs",
72 "honeynode-otntopo400GwithOTUC4.json")
73 with open(OTUC4_OTN_TOPO_BI_DIR_FILE, 'r', encoding='utf-8') as otuc4_otn_topo_bi_dir:
74 cls.otuc4_otn_topo_bi_dir_data = otuc4_otn_topo_bi_dir.read()
76 ODUC4_OTN_TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
77 "..", "..", "sample_configs",
78 "honeynode-otntopo400GwithODUC4.json")
79 with open(ODUC4_OTN_TOPO_BI_DIR_FILE, 'r', encoding='utf-8') as oduc4_otn_topo_bi_dir:
80 cls.oduc4_otn_topo_bi_dir_data = oduc4_otn_topo_bi_dir.read()
82 PORT_MAPPING_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
83 "..", "..", "sample_configs",
84 "pce_portmapping_71.json")
85 with open(PORT_MAPPING_FILE, 'r', encoding='utf-8') as port_mapping:
86 cls.port_mapping_data = port_mapping.read()
87 sample_files_parsed = True
88 except PermissionError as err:
89 print("Permission Error when trying to read sample files\n", err)
91 except FileNotFoundError as err:
92 print("File Not found Error when trying to read sample files\n", err)
95 print("Unexpected error when trying to read sample files\n", sys.exc_info()[0])
98 if sample_files_parsed:
99 print("sample files content loaded")
100 cls.processes = test_utils.start_tpce()
103 def tearDownClass(cls):
105 test_utils.del_portmapping()
106 test_utils.del_ietf_network('openroadm-topology')
107 test_utils.del_ietf_network('otn-topology')
108 # pylint: disable=not-an-iterable
109 for process in cls.processes:
110 test_utils.shutdown_process(process)
111 print("all processes killed")
113 def setUp(self): # instruction executed before each test method
114 # pylint: disable=consider-using-f-string
115 print("execution of {}".format(self.id().split(".")[-1]))
119 def test_01_load_port_mapping(self):
120 response = test_utils.post_portmapping(self.port_mapping_data)
121 self.assertIn(response['status_code'], (requests.codes.created, requests.codes.no_content))
124 # Load openroadm topology
125 def test_02_load_openroadm_topology_bi(self):
126 response = test_utils.put_ietf_network('openroadm-topology', self.topo_bi_dir_data)
127 self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content))
130 # Path Computation success
131 def test_03_path_computation_400G_xpdr_bi(self):
132 response = test_utils.transportpce_api_rpc_request('transportpce-pce',
133 'path-computation-request',
134 self.path_computation_input_data)
135 self.assertEqual(response['status_code'], requests.codes.ok)
136 self.assertIn('Path is calculated',
137 response['output']['configuration-response-common']['response-message'])
139 self.assertEqual(1, response['output']['response-parameters']['path-description']
140 ['aToZ-direction']['aToZ-wavelength-number'])
141 self.assertEqual(400, response['output']['response-parameters']['path-description']
142 ['aToZ-direction']['rate'])
143 self.assertEqual(196.0375, float(response['output']['response-parameters']['path-description']
144 ['aToZ-direction']['aToZ-min-frequency']))
145 self.assertEqual(196.12500, float(response['output']['response-parameters']['path-description']
146 ['aToZ-direction']['aToZ-max-frequency']))
147 self.assertEqual('dp-qam16', response['output']['response-parameters']['path-description']
148 ['aToZ-direction']['modulation-format'])
150 self.assertEqual(1, response['output']['response-parameters']['path-description']
151 ['zToA-direction']['zToA-wavelength-number'])
152 self.assertEqual(400, response['output']['response-parameters']['path-description']
153 ['zToA-direction']['rate'])
154 self.assertEqual(196.0375, float(response['output']['response-parameters']['path-description']
155 ['zToA-direction']['zToA-min-frequency']))
156 self.assertEqual(196.12500, float(response['output']['response-parameters']['path-description']
157 ['zToA-direction']['zToA-max-frequency']))
158 self.assertEqual('dp-qam16', response['output']['response-parameters']['path-description']
159 ['zToA-direction']['modulation-format'])
163 def test_04_load_otn_topology_bi(self):
164 response = test_utils.put_ietf_network('otn-topology', self.otn_topo_bi_dir_data)
165 self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content))
168 # Path Computation success
169 def test_05_path_computation_OTUC4_xpdr_bi(self):
170 self.path_computation_input_data["service-name"] = "service-OTUC4"
171 self.path_computation_input_data["service-a-end"]["service-format"] = "OTU"
172 self.path_computation_input_data["service-a-end"]["tx-direction"] =\
173 {"port": {"port-device-name": "XPDR-A2-XPDR2"}}
174 self.path_computation_input_data["service-z-end"]["service-format"] = "OTU"
175 self.path_computation_input_data["service-z-end"]["tx-direction"] =\
176 {"port": {"port-device-name": "XPDR-C2-XPDR2"}}
177 response = test_utils.transportpce_api_rpc_request('transportpce-pce',
178 'path-computation-request',
179 self.path_computation_input_data)
180 self.assertEqual(response['status_code'], requests.codes.ok)
181 self.assertIn('Path is calculated',
182 response['output']['configuration-response-common']['response-message'])
184 self.assertEqual(1, response['output']['response-parameters']['path-description']
185 ['aToZ-direction']['aToZ-wavelength-number'])
186 self.assertEqual(400, response['output']['response-parameters']['path-description']
187 ['aToZ-direction']['rate'])
188 self.assertEqual(196.0375, float(response['output']['response-parameters']['path-description']
189 ['aToZ-direction']['aToZ-min-frequency']))
190 self.assertEqual(196.12500, float(response['output']['response-parameters']['path-description']
191 ['aToZ-direction']['aToZ-max-frequency']))
192 self.assertEqual('dp-qam16', response['output']['response-parameters']['path-description']
193 ['aToZ-direction']['modulation-format'])
195 self.assertEqual(1, response['output']['response-parameters']['path-description']
196 ['zToA-direction']['zToA-wavelength-number'])
197 self.assertEqual(400, response['output']['response-parameters']['path-description']
198 ['zToA-direction']['rate'])
199 self.assertEqual(196.0375, float(response['output']['response-parameters']['path-description']
200 ['zToA-direction']['zToA-min-frequency']))
201 self.assertEqual(196.12500, float(response['output']['response-parameters']['path-description']
202 ['zToA-direction']['zToA-max-frequency']))
203 self.assertEqual('dp-qam16', response['output']['response-parameters']['path-description']
204 ['zToA-direction']['modulation-format'])
207 # Load otn topology with OTUC4 links
208 def test_06_load_otuc4_otn_topology_bi(self):
209 response = test_utils.put_ietf_network('otn-topology', self.otuc4_otn_topo_bi_dir_data)
210 self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content))
213 # Path Computation success
214 def test_07_path_computation_ODUC4_xpdr_bi(self):
215 self.path_computation_input_data["service-name"] = "service-ODUC4"
216 self.path_computation_input_data["service-a-end"]["service-format"] = "ODU"
217 self.path_computation_input_data["service-a-end"]["tx-direction"] = \
218 {"port": {"port-device-name": "XPDR-A2-XPDR2"}}
219 self.path_computation_input_data["service-z-end"]["service-format"] = "ODU"
220 self.path_computation_input_data["service-z-end"]["tx-direction"] = \
221 {"port": {"port-device-name": "XPDR-C2-XPDR2"}}
222 response = test_utils.transportpce_api_rpc_request('transportpce-pce',
223 'path-computation-request',
224 self.path_computation_input_data)
225 self.assertEqual(response['status_code'], requests.codes.ok)
226 self.assertIn('Path is calculated',
227 response['output']['configuration-response-common']['response-message'])
229 self.assertEqual(400, response['output']['response-parameters']['path-description']
230 ['aToZ-direction']['rate'])
231 self.assertEqual('dp-qam16', response['output']['response-parameters']['path-description']
232 ['aToZ-direction']['modulation-format'])
234 self.assertEqual(400, response['output']['response-parameters']['path-description']
235 ['zToA-direction']['rate'])
236 self.assertEqual('dp-qam16', response['output']['response-parameters']['path-description']
237 ['zToA-direction']['modulation-format'])
240 # Load otn topology with OTUC4 links
241 def test_08_load_oduc4_otn_topology_bi(self):
242 response = test_utils.put_ietf_network('otn-topology', self.oduc4_otn_topo_bi_dir_data)
243 self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content))
246 # Path Computation success
247 def test_09_path_computation_100G_xpdr_bi(self):
248 self.path_computation_input_data["service-name"] = "service-100GE"
249 self.path_computation_input_data["service-a-end"]["service-rate"] = "100"
250 self.path_computation_input_data["service-a-end"]["service-format"] = "Ethernet"
251 self.path_computation_input_data["service-a-end"]["tx-direction"] = \
252 {"port": {"port-device-name": "XPDR-A2-XPDR2",
253 "port-name": "XPDR2-CLIENT1"}}
254 self.path_computation_input_data["service-z-end"]["service-rate"] = "100"
255 self.path_computation_input_data["service-z-end"]["service-format"] = "Ethernet"
256 self.path_computation_input_data["service-z-end"]["tx-direction"] = \
257 {"port": {"port-device-name": "XPDR-C2-XPDR2",
258 "port-name": "XPDR2-CLIENT1"}}
259 self.path_computation_input_data["service-z-end"]["service-format"] = "ODU"
260 response = test_utils.transportpce_api_rpc_request('transportpce-pce',
261 'path-computation-request',
262 self.path_computation_input_data)
263 self.assertEqual(response['status_code'], requests.codes.ok)
264 self.assertIn('Path is calculated',
265 response['output']['configuration-response-common']['response-message'])
267 self.assertEqual(100, response['output']['response-parameters']['path-description']
268 ['aToZ-direction']['rate'])
269 self.assertEqual('1.1', response['output']['response-parameters']['path-description']
270 ['aToZ-direction']['min-trib-slot'])
271 self.assertEqual('1.20', response['output']['response-parameters']['path-description']
272 ['aToZ-direction']['max-trib-slot'])
273 self.assertEqual('dp-qpsk', response['output']['response-parameters']['path-description']
274 ['aToZ-direction']['modulation-format'])
276 self.assertEqual(100, response['output']['response-parameters']['path-description']
277 ['zToA-direction']['rate'])
278 self.assertEqual('1.1', response['output']['response-parameters']['path-description']
279 ['zToA-direction']['min-trib-slot'])
280 self.assertEqual('1.20', response['output']['response-parameters']['path-description']
281 ['zToA-direction']['max-trib-slot'])
282 self.assertEqual('dp-qpsk', response['output']['response-parameters']['path-description']
283 ['zToA-direction']['modulation-format'])
287 if __name__ == "__main__":
288 unittest.main(verbosity=2)