3 ##############################################################################
4 # Copyright (c) 2019 Orange, Inc. and others. All rights reserved.
6 # All rights reserved. This program and the accompanying materials
7 # are made available under the terms of the Apache License, Version 2.0
8 # which accompanies this distribution, and is available at
9 # http://www.apache.org/licenses/LICENSE-2.0
10 ##############################################################################
25 class TransportPCEtesting(unittest.TestCase):
29 restconf_baseurl = "http://localhost:8181/restconf"
33 cls.odl_process = test_utils.start_tpce()
34 cls.sim_process1 = test_utils.start_sim('spdrav2')
37 def tearDownClass(cls):
38 for child in psutil.Process(cls.odl_process.pid).children():
39 child.send_signal(signal.SIGINT)
41 cls.odl_process.send_signal(signal.SIGINT)
42 cls.odl_process.wait()
43 for child in psutil.Process(cls.sim_process1.pid).children():
44 child.send_signal(signal.SIGINT)
46 cls.sim_process1.send_signal(signal.SIGINT)
47 cls.sim_process1.wait()
52 def test_01_connect_SPDR_SA1(self):
53 url = ("{}/config/network-topology:"
54 "network-topology/topology/topology-netconf/node/SPDR-SA1"
55 .format(self.restconf_baseurl))
57 "node-id": "SPDR-SA1",
58 "netconf-node-topology:username": "admin",
59 "netconf-node-topology:password": "admin",
60 "netconf-node-topology:host": "127.0.0.1",
61 "netconf-node-topology:port": test_utils.sims['spdrav2']['port'],
62 "netconf-node-topology:tcp-only": "false",
63 "netconf-node-topology:pass-through": {}}]}
64 headers = {'content-type': 'application/json'}
65 response = requests.request(
66 "PUT", url, data=json.dumps(data), headers=headers,
67 auth=('admin', 'admin'))
68 self.assertEqual(response.status_code, requests.codes.created)
70 url = ("{}/operational/network-topology:"
71 "network-topology/topology/topology-netconf/node/SPDR-SA1"
72 .format(self.restconf_baseurl))
73 response = requests.request(
74 "GET", url, headers=headers, auth=('admin', 'admin'))
75 self.assertEqual(response.status_code, requests.codes.ok)
78 res['node'][0]['netconf-node-topology:connection-status'],
81 def test_02_getClliNetwork(self):
82 url = ("{}/config/ietf-network:networks/network/clli-network"
83 .format(self.restconf_baseurl))
84 headers = {'content-type': 'application/json'}
85 response = requests.request(
86 "GET", url, headers=headers, auth=('admin', 'admin'))
87 self.assertEqual(response.status_code, requests.codes.ok)
90 self.assertEqual(res['network'][0]['node'][0]['node-id'], 'NodeSA')
91 self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeSA')
93 def test_03_getOpenRoadmNetwork(self):
94 url = ("{}/config/ietf-network:networks/network/openroadm-network"
95 .format(self.restconf_baseurl))
96 headers = {'content-type': 'application/json'}
97 response = requests.request(
98 "GET", url, headers=headers, auth=('admin', 'admin'))
99 self.assertEqual(response.status_code, requests.codes.ok)
100 res = response.json()
101 self.assertEqual(res['network'][0]['node'][0]['node-id'], 'SPDR-SA1')
102 self.assertEqual(res['network'][0]['node'][0]['supporting-node'][0]['network-ref'], 'clli-network')
103 self.assertEqual(res['network'][0]['node'][0]['supporting-node'][0]['node-ref'], 'NodeSA')
104 self.assertEqual(res['network'][0]['node'][0]['org-openroadm-common-network:node-type'], 'XPONDER')
105 self.assertEqual(res['network'][0]['node'][0]['org-openroadm-network:model'], 'universal-switchponder')
106 self.assertEqual(res['network'][0]['node'][0]['org-openroadm-network:vendor'], 'vendorA')
107 self.assertEqual(res['network'][0]['node'][0]['org-openroadm-network:ip'], '1.2.3.4')
109 def test_04_getLinks_OpenroadmTopology(self):
110 url = ("{}/config/ietf-network:networks/network/openroadm-topology"
111 .format(self.restconf_baseurl))
112 headers = {'content-type': 'application/json'}
113 response = requests.request(
114 "GET", url, headers=headers, auth=('admin', 'admin'))
115 self.assertEqual(response.status_code, requests.codes.ok)
116 res = response.json()
117 # Tests related to links
118 self.assertNotIn('ietf-network-topology:link', res['network'][0])
120 def test_05_getNodes_OpenRoadmTopology(self):
121 url = ("{}/config/ietf-network:networks/network/openroadm-topology"
122 .format(self.restconf_baseurl))
123 headers = {'content-type': 'application/json'}
124 response = requests.request(
125 "GET", url, headers=headers, auth=('admin', 'admin'))
126 res = response.json()
127 # Tests related to nodes
128 self.assertEqual(response.status_code, requests.codes.ok)
129 self.assertIn('node', res['network'][0])
130 if ('node' in res['network'][0]):
131 nbNode = len(res['network'][0]['node'])
132 self.assertEqual(nbNode, 2)
133 listNode = ['SPDR-SA1-XPDR1', 'SPDR-SA1-XPDR2']
134 for i in range(0, nbNode):
135 nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type']
136 nodeId = res['network'][0]['node'][i]['node-id']
137 if (nodeId == 'SPDR-SA1-XPDR1'):
138 self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'},
139 res['network'][0]['node'][i]['supporting-node'])
140 self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeSA'},
141 res['network'][0]['node'][i]['supporting-node'])
142 self.assertEqual(nodeType, 'XPONDER')
143 nbTps = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
146 for j in range(0, nbTps):
147 tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
148 tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
149 if (tpType == 'XPONDER-CLIENT'):
151 elif (tpType == 'XPONDER-NETWORK'):
153 self.assertTrue(client == 0)
154 self.assertTrue(network == 1)
155 listNode.remove(nodeId)
156 elif (nodeId == 'SPDR-SA1-XPDR2'):
157 self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'},
158 res['network'][0]['node'][i]['supporting-node'])
159 self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeSA'},
160 res['network'][0]['node'][i]['supporting-node'])
161 self.assertEqual(nodeType, 'XPONDER')
162 nbTps = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
165 for j in range(0, nbTps):
166 tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
167 tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
168 if (tpType == 'XPONDER-CLIENT'):
170 elif (tpType == 'XPONDER-NETWORK'):
172 self.assertTrue(client == 0)
173 self.assertTrue(network == 4)
174 listNode.remove(nodeId)
176 self.assertFalse(True)
177 self.assertEqual(len(listNode), 0)
179 def test_06_getLinks_OtnTopology(self):
180 url = ("{}/config/ietf-network:networks/network/otn-topology"
181 .format(self.restconf_baseurl))
182 headers = {'content-type': 'application/json'}
183 response = requests.request(
184 "GET", url, headers=headers, auth=('admin', 'admin'))
185 self.assertEqual(response.status_code, requests.codes.ok)
186 res = response.json()
187 self.assertNotIn('ietf-network-topology:link', res['network'][0])
189 def test_07_getNodes_OtnTopology(self):
190 url = ("{}/config/ietf-network:networks/network/otn-topology"
191 .format(self.restconf_baseurl))
192 headers = {'content-type': 'application/json'}
193 response = requests.request(
194 "GET", url, headers=headers, auth=('admin', 'admin'))
195 res = response.json()
196 self.assertEqual(response.status_code, requests.codes.ok)
197 nbNode = len(res['network'][0]['node'])
198 self.assertEqual(nbNode, 2)
199 listNode = ['SPDR-SA1-XPDR1', 'SPDR-SA1-XPDR2']
200 for i in range(0, nbNode):
201 nodeId = res['network'][0]['node'][i]['node-id']
202 if(nodeId == 'SPDR-SA1-XPDR1'):
203 nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type']
204 self.assertEqual(nodeType, 'MUXPDR')
205 self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'},
206 res['network'][0]['node'][i]['supporting-node'])
207 self.assertIn({'network-ref': 'openroadm-topology', 'node-ref': 'SPDR-SA1-XPDR1'},
208 res['network'][0]['node'][i]['supporting-node'])
209 self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeSA'},
210 res['network'][0]['node'][i]['supporting-node'])
211 self.assertEqual(res['network'][0]['node'][i]
212 ['org-openroadm-otn-network-topology:xpdr-attributes']['xpdr-number'], 1)
213 nbTps = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
216 for j in range(0, nbTps):
217 tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
218 tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
219 if (tpType == 'XPONDER-CLIENT'):
221 self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-otn-network-topology:tp-supported-interfaces']
222 ['supported-interface-capability'][0]['if-cap-type'], 'org-openroadm-port-types:if-10GE-ODU2e')
223 self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
224 ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']['rate'], 'org-openroadm-otn-common-types:ODU2e')
225 elif (tpType == 'XPONDER-NETWORK'):
227 self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-otn-network-topology:tp-supported-interfaces']
228 ['supported-interface-capability'][0]['if-cap-type'], 'org-openroadm-port-types:if-OCH-OTU4-ODU4')
229 self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
230 ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']['rate'], 'org-openroadm-otn-common-types:ODU4')
231 self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
232 [j]['supporting-termination-point'][0]['network-ref'], 'openroadm-topology')
233 self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
234 [j]['supporting-termination-point'][0]['node-ref'], 'SPDR-SA1-XPDR1')
235 self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
236 [j]['supporting-termination-point'][0]['tp-ref'], 'XPDR1-NETWORK1')
237 self.assertTrue(client == 4)
238 self.assertTrue(network == 1)
239 listNode.remove(nodeId)
240 nbNbl = len(res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools']
241 ['odu-switching-pools'][0]['non-blocking-list'])
242 self.assertEqual(nbNbl, 4)
243 for k in range(0, nbNbl):
244 nbl = res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools']['odu-switching-pools'][0]['non-blocking-list'][k]
245 if (nbl['nbl-number'] == 1):
246 self.assertEqual(nbl['available-interconnect-bandwidth'], 10)
247 self.assertEqual(nbl['interconnect-bandwidth-unit'], 1000000000)
248 self.assertIn('XPDR1-NETWORK1', nbl['tp-list'])
249 self.assertIn('XPDR1-CLIENT1', nbl['tp-list'])
250 self.assertNotIn('XPDR1-CLIENT2', nbl['tp-list'])
251 elif(nodeId == 'SPDR-SA1-XPDR2'):
252 nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type']
253 self.assertEqual(nodeType, 'SWITCH')
254 self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'},
255 res['network'][0]['node'][i]['supporting-node'])
256 self.assertIn({'network-ref': 'openroadm-topology', 'node-ref': 'SPDR-SA1-XPDR2'},
257 res['network'][0]['node'][i]['supporting-node'])
258 self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeSA'},
259 res['network'][0]['node'][i]['supporting-node'])
260 self.assertEqual(res['network'][0]['node'][i]
261 ['org-openroadm-otn-network-topology:xpdr-attributes']['xpdr-number'], 2)
262 nbTps = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
265 for j in range(0, nbTps):
266 tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
267 tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
268 if (tpType == 'XPONDER-CLIENT'):
270 self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-otn-network-topology:tp-supported-interfaces']
271 ['supported-interface-capability'][0]['if-cap-type'], 'org-openroadm-port-types:if-100GE-ODU4')
272 self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
273 ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']['rate'], 'org-openroadm-otn-common-types:ODU4')
274 elif (tpType == 'XPONDER-NETWORK'):
276 self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-otn-network-topology:tp-supported-interfaces']
277 ['supported-interface-capability'][0]['if-cap-type'], 'org-openroadm-port-types:if-OCH-OTU4-ODU4')
278 self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
279 ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']['rate'], 'org-openroadm-otn-common-types:ODU4')
280 self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
281 [j]['supporting-termination-point'][0]['network-ref'], 'openroadm-topology')
282 self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
283 [j]['supporting-termination-point'][0]['node-ref'], 'SPDR-SA1-XPDR2')
284 self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
285 [j]['supporting-termination-point'][0]['tp-ref'], tpId)
286 self.assertTrue(client == 4)
287 self.assertTrue(network == 4)
288 listNode.remove(nodeId)
289 nbNbl = len(res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools']
290 ['odu-switching-pools'][0]['non-blocking-list'])
291 self.assertEqual(nbNbl, 1)
292 nbl = res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools']['odu-switching-pools'][0]['non-blocking-list'][0]
293 self.assertIn('XPDR2-NETWORK1', nbl['tp-list'])
294 self.assertIn('XPDR2-CLIENT1', nbl['tp-list'])
295 self.assertIn('XPDR2-NETWORK2', nbl['tp-list'])
296 self.assertIn('XPDR2-CLIENT2', nbl['tp-list'])
297 self.assertIn('XPDR2-NETWORK3', nbl['tp-list'])
298 self.assertIn('XPDR2-CLIENT3', nbl['tp-list'])
299 self.assertIn('XPDR2-NETWORK4', nbl['tp-list'])
300 self.assertIn('XPDR2-CLIENT4', nbl['tp-list'])
302 self.assertFalse(True)
303 self.assertEqual(len(listNode), 0)
305 def test_08_disconnect_SPDR_SA1(self):
306 url = ("{}/config/network-topology:"
307 "network-topology/topology/topology-netconf/node/SPDR-SA1"
308 .format(self.restconf_baseurl))
310 headers = {'content-type': 'application/json'}
311 response = requests.request(
312 "DELETE", url, data=json.dumps(data), headers=headers,
313 auth=('admin', 'admin'))
314 self.assertEqual(response.status_code, requests.codes.ok)
316 def test_09_getClliNetwork(self):
317 url = ("{}/config/ietf-network:networks/network/clli-network"
318 .format(self.restconf_baseurl))
319 headers = {'content-type': 'application/json'}
320 response = requests.request(
321 "GET", url, headers=headers, auth=('admin', 'admin'))
322 self.assertEqual(response.status_code, requests.codes.ok)
323 res = response.json()
324 nbNode = len(res['network'][0]['node'])
325 self.assertEqual(nbNode, 1)
326 self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeSA')
328 def test_10_getOpenRoadmNetwork(self):
329 url = ("{}/config/ietf-network:networks/network/openroadm-network"
330 .format(self.restconf_baseurl))
331 headers = {'content-type': 'application/json'}
332 response = requests.request(
333 "GET", url, headers=headers, auth=('admin', 'admin'))
334 self.assertEqual(response.status_code, requests.codes.ok)
335 res = response.json()
336 self.assertNotIn('node', res['network'][0])
338 def test_11_getNodes_OpenRoadmTopology(self):
339 url = ("{}/config/ietf-network:networks/network/openroadm-topology"
340 .format(self.restconf_baseurl))
341 headers = {'content-type': 'application/json'}
342 response = requests.request(
343 "GET", url, headers=headers, auth=('admin', 'admin'))
344 res = response.json()
345 self.assertNotIn('node', res['network'][0])
347 def test_12_getNodes_OtnTopology(self):
348 url = ("{}/config/ietf-network:networks/network/otn-topology"
349 .format(self.restconf_baseurl))
350 headers = {'content-type': 'application/json'}
351 response = requests.request(
352 "GET", url, headers=headers, auth=('admin', 'admin'))
353 res = response.json()
354 self.assertNotIn('node', res['network'][0])
357 if __name__ == "__main__":
358 unittest.main(verbosity=2)