3 ##############################################################################
4 # Copyright (c) 2017 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 ##############################################################################
21 class TransportPCEPortMappingTesting(unittest.TestCase):
26 restconf_baseurl = "http://localhost:8181/restconf"
28 # START_IGNORE_XTESTING
32 cls.sim_process1 = test_utils.start_sim('xpdra')
35 cls.sim_process2 = test_utils.start_sim('roadma')
38 cls.odl_process = test_utils.start_tpce()
40 print("opendaylight started")
43 def tearDownClass(cls):
44 for child in psutil.Process(cls.odl_process.pid).children():
45 child.send_signal(signal.SIGINT)
47 cls.odl_process.send_signal(signal.SIGINT)
48 cls.odl_process.wait()
49 for child in psutil.Process(cls.sim_process1.pid).children():
50 child.send_signal(signal.SIGINT)
52 cls.sim_process1.send_signal(signal.SIGINT)
53 cls.sim_process1.wait()
54 for child in psutil.Process(cls.sim_process2.pid).children():
55 child.send_signal(signal.SIGINT)
57 cls.sim_process2.send_signal(signal.SIGINT)
58 cls.sim_process2.wait()
61 print("execution of {}".format(self.id().split(".")[-1]))
66 # def test_01_restconfAPI(self):
67 # url = ("{}/operational/network-topology:network-topology/topology/"
68 # "topology-netconf/node/controller-config".format(self.restconf_baseurl))
69 # headers = {'content-type': 'application/json'}
70 # response = requests.request("GET", url, headers=headers, auth=('admin', 'admin'))
71 # self.assertEqual(response.status_code, requests.codes.ok)
72 # res = response.json()
73 # self.assertEqual(res['node'] [0] ['netconf-node-topology:connection-status'],
76 # def test_02_restconfAPI(self):
77 # url = ("{}/config/transportpce-portmapping:network/nodes/controller-config".format(self.restconf_baseurl))
78 # headers = {'content-type': 'application/json'}
79 # response = requests.request(
80 # "GET", url, headers=headers, auth=('admin', 'admin'))
81 # self.assertEqual(response.status_code, requests.codes.not_found)
82 # res = response.json()
84 # {"error-type":"application", "error-tag":"data-missing",
85 # "error-message":"Request could not be completed because the relevant data model content does not exist "},
86 # res['errors']['error'])
88 def test_01_rdm_device_connected(self):
89 url = ("{}/config/network-topology:"
90 "network-topology/topology/topology-netconf/node/ROADMA01"
91 .format(self.restconf_baseurl))
93 "node-id": "ROADMA01",
94 "netconf-node-topology:username": "admin",
95 "netconf-node-topology:password": "admin",
96 "netconf-node-topology:host": "127.0.0.1",
97 "netconf-node-topology:port": test_utils.sims['roadma']['port'],
98 "netconf-node-topology:tcp-only": "false",
99 "netconf-node-topology:pass-through": {}}]}
100 headers = {'content-type': 'application/json'}
101 response = requests.request(
102 "PUT", url, data=json.dumps(data), headers=headers,
103 auth=('admin', 'admin'))
104 self.assertEqual(response.status_code, requests.codes.created)
107 def test_02_rdm_device_connected(self):
108 url = ("{}/operational/network-topology:"
109 "network-topology/topology/topology-netconf/node/ROADMA01"
110 .format(self.restconf_baseurl))
111 headers = {'content-type': 'application/json'}
112 response = requests.request(
113 "GET", url, headers=headers, auth=('admin', 'admin'))
114 self.assertEqual(response.status_code, requests.codes.ok)
115 res = response.json()
117 res['node'][0]['netconf-node-topology:connection-status'],
121 def test_03_rdm_portmapping_info(self):
122 url = ("{}/config/transportpce-portmapping:network/"
123 "nodes/ROADMA01/node-info"
124 .format(self.restconf_baseurl))
125 headers = {'content-type': 'application/json'}
126 response = requests.request(
127 "GET", url, headers=headers, auth=('admin', 'admin'))
128 self.assertEqual(response.status_code, requests.codes.ok)
129 res = response.json()
131 {u'node-info': {u'node-type': u'rdm',
132 u'node-ip-address': u'127.0.0.12',
133 u'node-clli': u'NodeA',
134 u'openroadm-version': u'1.2.1', u'node-vendor': u'vendorA',
135 u'node-model': u'2'}},
139 def test_04_rdm_portmapping_DEG1_TTP_TXRX(self):
140 url = ("{}/config/transportpce-portmapping:network/"
141 "nodes/ROADMA01/mapping/DEG1-TTP-TXRX"
142 .format(self.restconf_baseurl))
143 headers = {'content-type': 'application/json'}
144 response = requests.request(
145 "GET", url, headers=headers, auth=('admin', 'admin'))
146 self.assertEqual(response.status_code, requests.codes.ok)
147 res = response.json()
149 {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0',
150 'logical-connection-point': 'DEG1-TTP-TXRX', 'port-direction': 'bidirectional'},
153 def test_05_rdm_portmapping_SRG1_PP7_TXRX(self):
154 url = ("{}/config/transportpce-portmapping:network/"
155 "nodes/ROADMA01/mapping/SRG1-PP7-TXRX"
156 .format(self.restconf_baseurl))
157 headers = {'content-type': 'application/json'}
158 response = requests.request(
159 "GET", url, headers=headers, auth=('admin', 'admin'))
160 self.assertEqual(response.status_code, requests.codes.ok)
161 res = response.json()
163 {'supporting-port': 'C7', 'supporting-circuit-pack-name': '4/0',
164 'logical-connection-point': 'SRG1-PP7-TXRX', 'port-direction': 'bidirectional'},
167 def test_06_rdm_portmapping_SRG3_PP1_TXRX(self):
168 url = ("{}/config/transportpce-portmapping:network/"
169 "nodes/ROADMA01/mapping/SRG3-PP1-TXRX"
170 .format(self.restconf_baseurl))
171 headers = {'content-type': 'application/json'}
172 response = requests.request(
173 "GET", url, headers=headers, auth=('admin', 'admin'))
174 self.assertEqual(response.status_code, requests.codes.ok)
175 res = response.json()
177 {'supporting-port': 'C1', 'supporting-circuit-pack-name': '5/0',
178 'logical-connection-point': 'SRG3-PP1-TXRX', 'port-direction': 'bidirectional'},
181 def test_07_xpdr_device_connected(self):
182 url = ("{}/config/network-topology:"
183 "network-topology/topology/topology-netconf/node/XPDRA01"
184 .format(self.restconf_baseurl))
186 "node-id": "XPDRA01",
187 "netconf-node-topology:username": "admin",
188 "netconf-node-topology:password": "admin",
189 "netconf-node-topology:host": "127.0.0.1",
190 "netconf-node-topology:port": test_utils.sims['xpdra']['port'],
191 "netconf-node-topology:tcp-only": "false",
192 "netconf-node-topology:pass-through": {}}]}
193 headers = {'content-type': 'application/json'}
194 response = requests.request(
195 "PUT", url, data=json.dumps(data), headers=headers,
196 auth=('admin', 'admin'))
197 self.assertEqual(response.status_code, requests.codes.created)
200 def test_08_xpdr_device_connected(self):
201 url = ("{}/operational/network-topology:"
202 "network-topology/topology/topology-netconf/node/XPDRA01"
203 .format(self.restconf_baseurl))
204 headers = {'content-type': 'application/json'}
205 response = requests.request(
206 "GET", url, headers=headers, auth=('admin', 'admin'))
207 self.assertEqual(response.status_code, requests.codes.ok)
208 res = response.json()
210 res['node'][0]['netconf-node-topology:connection-status'],
214 def test_09_xpdr_portmapping_info(self):
215 url = ("{}/config/transportpce-portmapping:network/"
216 "nodes/XPDRA01/node-info"
217 .format(self.restconf_baseurl))
218 headers = {'content-type': 'application/json'}
219 response = requests.request(
220 "GET", url, headers=headers, auth=('admin', 'admin'))
221 self.assertEqual(response.status_code, requests.codes.ok)
222 res = response.json()
224 {u'node-info': {u'node-type': u'xpdr',
225 u'node-ip-address': u'127.0.0.10',
226 u'node-clli': u'NodeA',
227 u'openroadm-version': u'1.2.1', u'node-vendor': u'vendorA',
228 u'node-model': u'1'}},
232 def test_10_xpdr_portmapping_NETWORK1(self):
233 url = ("{}/config/transportpce-portmapping:network/"
234 "nodes/XPDRA01/mapping/XPDR1-NETWORK1"
235 .format(self.restconf_baseurl))
236 headers = {'content-type': 'application/json'}
237 response = requests.request(
238 "GET", url, headers=headers, auth=('admin', 'admin'))
239 self.assertEqual(response.status_code, requests.codes.ok)
240 res = response.json()
242 {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
243 'logical-connection-point': 'XPDR1-NETWORK1', 'port-direction': 'bidirectional',
244 'connection-map-lcp': 'XPDR1-CLIENT1', 'port-qual': 'xpdr-network',
245 'lcp-hash-val': '3b3ab304d2a6eb3c3623e52746dbb7aa'},
248 def test_11_xpdr_portmapping_NETWORK2(self):
249 url = ("{}/config/transportpce-portmapping:network/"
250 "nodes/XPDRA01/mapping/XPDR1-NETWORK2"
251 .format(self.restconf_baseurl))
252 headers = {'content-type': 'application/json'}
253 response = requests.request(
254 "GET", url, headers=headers, auth=('admin', 'admin'))
255 self.assertEqual(response.status_code, requests.codes.ok)
256 res = response.json()
258 {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/2-PLUG-NET',
259 'logical-connection-point': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional',
260 'connection-map-lcp': 'XPDR1-CLIENT3', 'port-qual': 'xpdr-network',
261 'lcp-hash-val': '3b3ab304d2a6eb3c3623e52746dbb7a9'},
264 def test_12_xpdr_portmapping_CLIENT1(self):
265 url = ("{}/config/transportpce-portmapping:network/"
266 "nodes/XPDRA01/mapping/XPDR1-CLIENT1"
267 .format(self.restconf_baseurl))
268 headers = {'content-type': 'application/json'}
269 response = requests.request(
270 "GET", url, headers=headers, auth=('admin', 'admin'))
271 self.assertEqual(response.status_code, requests.codes.ok)
272 res = response.json()
274 {'supporting-port': 'C1',
275 'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
276 'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
277 'connection-map-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client',
278 'lcp-hash-val': '64b8effe7ba72211420bf267d0ca1ae5'},
281 def test_13_xpdr_portmapping_CLIENT2(self):
282 url = ("{}/config/transportpce-portmapping:network/"
283 "nodes/XPDRA01/mapping/XPDR1-CLIENT2"
284 .format(self.restconf_baseurl))
285 headers = {'content-type': 'application/json'}
286 response = requests.request(
287 "GET", url, headers=headers, auth=('admin', 'admin'))
288 self.assertEqual(response.status_code, requests.codes.ok)
289 res = response.json()
291 {'supporting-port': 'C2',
292 'supporting-circuit-pack-name': '1/0/C2-PLUG-CLIENT',
293 'logical-connection-point': 'XPDR1-CLIENT2', 'port-direction': 'bidirectional',
294 'port-qual': 'xpdr-client',
295 'lcp-hash-val': '64b8effe7ba72211420bf267d0ca1ae6'},
298 def test_14_xpdr_portmapping_CLIENT3(self):
299 url = ("{}/config/transportpce-portmapping:network/"
300 "nodes/XPDRA01/mapping/XPDR1-CLIENT3"
301 .format(self.restconf_baseurl))
302 headers = {'content-type': 'application/json'}
303 response = requests.request(
304 "GET", url, headers=headers, auth=('admin', 'admin'))
305 self.assertEqual(response.status_code, requests.codes.ok)
306 res = response.json()
308 {'supporting-port': 'C3',
309 'supporting-circuit-pack-name': '1/0/C3-PLUG-CLIENT',
310 'logical-connection-point': 'XPDR1-CLIENT3',
311 'connection-map-lcp': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional',
312 'port-qual': 'xpdr-client', 'lcp-hash-val': '64b8effe7ba72211420bf267d0ca1ae7'},
315 def test_15_xpdr_portmapping_CLIENT4(self):
316 url = ("{}/config/transportpce-portmapping:network/"
317 "nodes/XPDRA01/mapping/XPDR1-CLIENT4"
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()
325 {'supporting-port': 'C4',
326 'supporting-circuit-pack-name': '1/0/C4-PLUG-CLIENT',
327 'logical-connection-point': 'XPDR1-CLIENT4', 'port-direction': 'bidirectional',
328 'port-qual': 'xpdr-client', 'lcp-hash-val': '64b8effe7ba72211420bf267d0ca1ae0'},
331 def test_16_xpdr_device_disconnected(self):
332 url = ("{}/config/network-topology:"
333 "network-topology/topology/topology-netconf/node/XPDRA01"
334 .format(self.restconf_baseurl))
335 headers = {'content-type': 'application/json'}
336 response = requests.request(
337 "DELETE", url, headers=headers,
338 auth=('admin', 'admin'))
339 self.assertEqual(response.status_code, requests.codes.ok)
342 def test_17_xpdr_device_disconnected(self):
343 url = ("{}/operational/network-topology:network-topology/topology/"
344 "topology-netconf/node/XPDRA01".format(self.restconf_baseurl))
345 headers = {'content-type': 'application/json'}
346 response = requests.request(
347 "GET", url, headers=headers, auth=('admin', 'admin'))
348 self.assertEqual(response.status_code, requests.codes.not_found)
349 res = response.json()
351 {"error-type": "application", "error-tag": "data-missing",
352 "error-message": "Request could not be completed because the relevant data model content does not exist"},
353 res['errors']['error'])
355 def test_18_xpdr_device_disconnected(self):
356 url = ("{}/config/transportpce-portmapping:network/nodes/XPDRA01".format(self.restconf_baseurl))
357 headers = {'content-type': 'application/json'}
358 response = requests.request(
359 "GET", url, headers=headers, auth=('admin', 'admin'))
360 self.assertEqual(response.status_code, requests.codes.not_found)
361 res = response.json()
363 {"error-type": "application", "error-tag": "data-missing",
364 "error-message": "Request could not be completed because the relevant data model content does not exist"},
365 res['errors']['error'])
367 def test_19_rdm_device_disconnected(self):
368 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADMA01"
369 .format(self.restconf_baseurl))
370 headers = {'content-type': 'application/json'}
371 response = requests.request(
372 "DELETE", url, headers=headers,
373 auth=('admin', 'admin'))
374 self.assertEqual(response.status_code, requests.codes.ok)
377 def test_20_rdm_device_disconnected(self):
378 url = ("{}/operational/network-topology:network-topology/topology/topology-netconf/node/ROADMA01"
379 .format(self.restconf_baseurl))
380 headers = {'content-type': 'application/json'}
381 response = requests.request(
382 "GET", url, headers=headers, auth=('admin', 'admin'))
383 self.assertEqual(response.status_code, requests.codes.not_found)
384 res = response.json()
386 {"error-type": "application", "error-tag": "data-missing",
387 "error-message": "Request could not be completed because the relevant data model content does not exist"},
388 res['errors']['error'])
390 def test_21_rdm_device_disconnected(self):
391 url = ("{}/config/transportpce-portmapping:network/nodes/ROADMA01".format(self.restconf_baseurl))
392 headers = {'content-type': 'application/json'}
393 response = requests.request(
394 "GET", url, headers=headers, auth=('admin', 'admin'))
395 self.assertEqual(response.status_code, requests.codes.not_found)
396 res = response.json()
398 {"error-type": "application", "error-tag": "data-missing",
399 "error-message": "Request could not be completed because the relevant data model content does not exist"},
400 res['errors']['error'])
403 if __name__ == "__main__":
404 unittest.main(verbosity=2)