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 ##############################################################################
24 class TransportPCEPortMappingTesting(unittest.TestCase):
26 honeynode_process1 = None
27 honeynode_process2 = None
29 restconf_baseurl = "http://localhost:8181/restconf"
31 #START_IGNORE_XTESTING
35 print ("starting honeynode1...")
36 cls.honeynode_process1 = test_utils.start_xpdra_honeynode()
39 print ("starting honeynode2...")
40 cls.honeynode_process2 = test_utils.start_roadma_honeynode()
43 print ("starting opendaylight...")
44 cls.odl_process = test_utils.start_tpce()
46 print ("opendaylight started")
49 def tearDownClass(cls):
50 for child in psutil.Process(cls.odl_process.pid).children():
51 child.send_signal(signal.SIGINT)
53 cls.odl_process.send_signal(signal.SIGINT)
54 cls.odl_process.wait()
55 for child in psutil.Process(cls.honeynode_process1.pid).children():
56 child.send_signal(signal.SIGINT)
58 cls.honeynode_process1.send_signal(signal.SIGINT)
59 cls.honeynode_process1.wait()
60 for child in psutil.Process(cls.honeynode_process2.pid).children():
61 child.send_signal(signal.SIGINT)
63 cls.honeynode_process2.send_signal(signal.SIGINT)
64 cls.honeynode_process2.wait()
67 print ("execution of {}".format(self.id().split(".")[-1]))
72 # def test_01_restconfAPI(self):
73 # url = ("{}/operational/network-topology:network-topology/topology/"
74 # "topology-netconf/node/controller-config".format(self.restconf_baseurl))
75 # headers = {'content-type': 'application/json'}
76 # response = requests.request("GET", url, headers=headers, auth=('admin', 'admin'))
77 # self.assertEqual(response.status_code, requests.codes.ok)
78 # res = response.json()
79 # self.assertEqual(res['node'] [0] ['netconf-node-topology:connection-status'],
82 # def test_02_restconfAPI(self):
83 # url = ("{}/config/transportpce-portmapping:network/nodes/controller-config".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.not_found)
88 # res = response.json()
90 # {"error-type":"application", "error-tag":"data-missing",
91 # "error-message":"Request could not be completed because the relevant data model content does not exist "},
92 # res['errors']['error'])
94 def test_01_rdm_device_connected(self):
95 url = ("{}/config/network-topology:"
96 "network-topology/topology/topology-netconf/node/ROADMA01"
97 .format(self.restconf_baseurl))
99 "node-id": "ROADMA01",
100 "netconf-node-topology:username": "admin",
101 "netconf-node-topology:password": "admin",
102 "netconf-node-topology:host": "127.0.0.1",
103 "netconf-node-topology:port": "17831",
104 "netconf-node-topology:tcp-only": "false",
105 "netconf-node-topology:pass-through": {}}]}
106 headers = {'content-type': 'application/json'}
107 response = requests.request(
108 "PUT", url, data=json.dumps(data), headers=headers,
109 auth=('admin', 'admin'))
110 self.assertEqual(response.status_code, requests.codes.created)
113 def test_02_rdm_device_connected(self):
114 url = ("{}/operational/network-topology:"
115 "network-topology/topology/topology-netconf/node/ROADMA01"
116 .format(self.restconf_baseurl))
117 headers = {'content-type': 'application/json'}
118 response = requests.request(
119 "GET", url, headers=headers, auth=('admin', 'admin'))
120 self.assertEqual(response.status_code, requests.codes.ok)
121 res = response.json()
123 res['node'][0]['netconf-node-topology:connection-status'],
127 def test_03_rdm_portmapping_info(self):
128 url = ("{}/config/transportpce-portmapping:network/"
129 "nodes/ROADMA01/node-info"
130 .format(self.restconf_baseurl))
131 headers = {'content-type': 'application/json'}
132 response = requests.request(
133 "GET", url, headers=headers, auth=('admin', 'admin'))
134 self.assertEqual(response.status_code, requests.codes.ok)
135 res = response.json()
137 {u'node-info': {u'node-type': u'rdm',
138 u'node-ip-address': u'127.0.0.12',
139 u'node-clli': u'NodeA',
140 u'openroadm-version': u'1.2.1', u'node-vendor': u'vendorA',
141 u'node-model': u'2'}},
145 def test_04_rdm_portmapping_DEG1_TTP_TXRX(self):
146 url = ("{}/config/transportpce-portmapping:network/"
147 "nodes/ROADMA01/mapping/DEG1-TTP-TXRX"
148 .format(self.restconf_baseurl))
149 headers = {'content-type': 'application/json'}
150 response = requests.request(
151 "GET", url, headers=headers, auth=('admin', 'admin'))
152 self.assertEqual(response.status_code, requests.codes.ok)
153 res = response.json()
155 {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0',
156 'logical-connection-point': 'DEG1-TTP-TXRX', 'port-direction': 'bidirectional'},
159 def test_05_rdm_portmapping_SRG1_PP7_TXRX(self):
160 url = ("{}/config/transportpce-portmapping:network/"
161 "nodes/ROADMA01/mapping/SRG1-PP7-TXRX"
162 .format(self.restconf_baseurl))
163 headers = {'content-type': 'application/json'}
164 response = requests.request(
165 "GET", url, headers=headers, auth=('admin', 'admin'))
166 self.assertEqual(response.status_code, requests.codes.ok)
167 res = response.json()
169 {'supporting-port': 'C7', 'supporting-circuit-pack-name': '4/0',
170 'logical-connection-point': 'SRG1-PP7-TXRX', 'port-direction': 'bidirectional'},
173 def test_06_rdm_portmapping_SRG3_PP1_TXRX(self):
174 url = ("{}/config/transportpce-portmapping:network/"
175 "nodes/ROADMA01/mapping/SRG3-PP1-TXRX"
176 .format(self.restconf_baseurl))
177 headers = {'content-type': 'application/json'}
178 response = requests.request(
179 "GET", url, headers=headers, auth=('admin', 'admin'))
180 self.assertEqual(response.status_code, requests.codes.ok)
181 res = response.json()
183 {'supporting-port': 'C1', 'supporting-circuit-pack-name': '5/0',
184 'logical-connection-point': 'SRG3-PP1-TXRX', 'port-direction': 'bidirectional'},
187 def test_07_xpdr_device_connected(self):
188 url = ("{}/config/network-topology:"
189 "network-topology/topology/topology-netconf/node/XPDRA01"
190 .format(self.restconf_baseurl))
192 "node-id": "XPDRA01",
193 "netconf-node-topology:username": "admin",
194 "netconf-node-topology:password": "admin",
195 "netconf-node-topology:host": "127.0.0.1",
196 "netconf-node-topology:port": "17830",
197 "netconf-node-topology:tcp-only": "false",
198 "netconf-node-topology:pass-through": {}}]}
199 headers = {'content-type': 'application/json'}
200 response = requests.request(
201 "PUT", url, data=json.dumps(data), headers=headers,
202 auth=('admin', 'admin'))
203 self.assertEqual(response.status_code, requests.codes.created)
206 def test_08_xpdr_device_connected(self):
207 url = ("{}/operational/network-topology:"
208 "network-topology/topology/topology-netconf/node/XPDRA01"
209 .format(self.restconf_baseurl))
210 headers = {'content-type': 'application/json'}
211 response = requests.request(
212 "GET", url, headers=headers, auth=('admin', 'admin'))
213 self.assertEqual(response.status_code, requests.codes.ok)
214 res = response.json()
216 res['node'][0]['netconf-node-topology:connection-status'],
220 def test_09_xpdr_portmapping_info(self):
221 url = ("{}/config/transportpce-portmapping:network/"
222 "nodes/XPDRA01/node-info"
223 .format(self.restconf_baseurl))
224 headers = {'content-type': 'application/json'}
225 response = requests.request(
226 "GET", url, headers=headers, auth=('admin', 'admin'))
227 self.assertEqual(response.status_code, requests.codes.ok)
228 res = response.json()
230 {u'node-info': {u'node-type': u'xpdr',
231 u'node-ip-address': u'127.0.0.10',
232 u'node-clli': u'NodeA',
233 u'openroadm-version': u'1.2.1', u'node-vendor': u'vendorA',
234 u'node-model': u'1'}},
238 def test_10_xpdr_portmapping_NETWORK1(self):
239 url = ("{}/config/transportpce-portmapping:network/"
240 "nodes/XPDRA01/mapping/XPDR1-NETWORK1"
241 .format(self.restconf_baseurl))
242 headers = {'content-type': 'application/json'}
243 response = requests.request(
244 "GET", url, headers=headers, auth=('admin', 'admin'))
245 self.assertEqual(response.status_code, requests.codes.ok)
246 res = response.json()
248 {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
249 'logical-connection-point': 'XPDR1-NETWORK1', 'port-direction': 'bidirectional',
250 'connection-map-lcp': 'XPDR1-CLIENT1', 'port-qual': 'xpdr-network'},
253 def test_11_xpdr_portmapping_NETWORK2(self):
254 url = ("{}/config/transportpce-portmapping:network/"
255 "nodes/XPDRA01/mapping/XPDR1-NETWORK2"
256 .format(self.restconf_baseurl))
257 headers = {'content-type': 'application/json'}
258 response = requests.request(
259 "GET", url, headers=headers, auth=('admin', 'admin'))
260 self.assertEqual(response.status_code, requests.codes.ok)
261 res = response.json()
263 {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/2-PLUG-NET',
264 'logical-connection-point': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional',
265 'connection-map-lcp': 'XPDR1-CLIENT3', 'port-qual': 'xpdr-network'},
268 def test_12_xpdr_portmapping_CLIENT1(self):
269 url = ("{}/config/transportpce-portmapping:network/"
270 "nodes/XPDRA01/mapping/XPDR1-CLIENT1"
271 .format(self.restconf_baseurl))
272 headers = {'content-type': 'application/json'}
273 response = requests.request(
274 "GET", url, headers=headers, auth=('admin', 'admin'))
275 self.assertEqual(response.status_code, requests.codes.ok)
276 res = response.json()
278 {'supporting-port': 'C1',
279 'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
280 'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
281 'connection-map-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client'},
284 def test_13_xpdr_portmapping_CLIENT2(self):
285 url = ("{}/config/transportpce-portmapping:network/"
286 "nodes/XPDRA01/mapping/XPDR1-CLIENT2"
287 .format(self.restconf_baseurl))
288 headers = {'content-type': 'application/json'}
289 response = requests.request(
290 "GET", url, headers=headers, auth=('admin', 'admin'))
291 self.assertEqual(response.status_code, requests.codes.ok)
292 res = response.json()
294 {'supporting-port': 'C2',
295 'supporting-circuit-pack-name': '1/0/C2-PLUG-CLIENT',
296 'logical-connection-point': 'XPDR1-CLIENT2', 'port-direction': 'bidirectional',
297 'port-qual': 'xpdr-client'},
300 def test_14_xpdr_portmapping_CLIENT3(self):
301 url = ("{}/config/transportpce-portmapping:network/"
302 "nodes/XPDRA01/mapping/XPDR1-CLIENT3"
303 .format(self.restconf_baseurl))
304 headers = {'content-type': 'application/json'}
305 response = requests.request(
306 "GET", url, headers=headers, auth=('admin', 'admin'))
307 self.assertEqual(response.status_code, requests.codes.ok)
308 res = response.json()
310 {'supporting-port': 'C3',
311 'supporting-circuit-pack-name': '1/0/C3-PLUG-CLIENT',
312 'logical-connection-point': 'XPDR1-CLIENT3',
313 'connection-map-lcp': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional',
314 'port-qual': 'xpdr-client'},
317 def test_15_xpdr_portmapping_CLIENT4(self):
318 url = ("{}/config/transportpce-portmapping:network/"
319 "nodes/XPDRA01/mapping/XPDR1-CLIENT4"
320 .format(self.restconf_baseurl))
321 headers = {'content-type': 'application/json'}
322 response = requests.request(
323 "GET", url, headers=headers, auth=('admin', 'admin'))
324 self.assertEqual(response.status_code, requests.codes.ok)
325 res = response.json()
327 {'supporting-port': 'C4',
328 'supporting-circuit-pack-name': '1/0/C4-PLUG-CLIENT',
329 'logical-connection-point': 'XPDR1-CLIENT4', 'port-direction': 'bidirectional',
330 'port-qual': 'xpdr-client'},
333 def test_16_xpdr_device_disconnected(self):
334 url = ("{}/config/network-topology:"
335 "network-topology/topology/topology-netconf/node/XPDRA01"
336 .format(self.restconf_baseurl))
337 headers = {'content-type': 'application/json'}
338 response = requests.request(
339 "DELETE", url, headers=headers,
340 auth=('admin', 'admin'))
341 self.assertEqual(response.status_code, requests.codes.ok)
344 def test_17_xpdr_device_disconnected(self):
345 url = ("{}/operational/network-topology:network-topology/topology/"
346 "topology-netconf/node/XPDRA01".format(self.restconf_baseurl))
347 headers = {'content-type': 'application/json'}
348 response = requests.request(
349 "GET", url, headers=headers, auth=('admin', 'admin'))
350 self.assertEqual(response.status_code, requests.codes.not_found)
351 res = response.json()
353 {"error-type":"application", "error-tag":"data-missing",
354 "error-message":"Request could not be completed because the relevant data model content does not exist"},
355 res['errors']['error'])
357 def test_18_xpdr_device_disconnected(self):
358 url = ("{}/config/transportpce-portmapping:network/nodes/XPDRA01".format(self.restconf_baseurl))
359 headers = {'content-type': 'application/json'}
360 response = requests.request(
361 "GET", url, headers=headers, auth=('admin', 'admin'))
362 self.assertEqual(response.status_code, requests.codes.not_found)
363 res = response.json()
365 {"error-type":"application", "error-tag":"data-missing",
366 "error-message":"Request could not be completed because the relevant data model content does not exist"},
367 res['errors']['error'])
369 def test_19_rdm_device_disconnected(self):
370 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADMA01"
371 .format(self.restconf_baseurl))
372 headers = {'content-type': 'application/json'}
373 response = requests.request(
374 "DELETE", url, headers=headers,
375 auth=('admin', 'admin'))
376 self.assertEqual(response.status_code, requests.codes.ok)
379 def test_20_rdm_device_disconnected(self):
380 url = ("{}/operational/network-topology:network-topology/topology/topology-netconf/node/ROADMA01"
381 .format(self.restconf_baseurl))
382 headers = {'content-type': 'application/json'}
383 response = requests.request(
384 "GET", url, headers=headers, auth=('admin', 'admin'))
385 self.assertEqual(response.status_code, requests.codes.not_found)
386 res = response.json()
388 {"error-type":"application", "error-tag":"data-missing",
389 "error-message":"Request could not be completed because the relevant data model content does not exist"},
390 res['errors']['error'])
392 def test_21_rdm_device_disconnected(self):
393 url = ("{}/config/transportpce-portmapping:network/nodes/ROADMA01".format(self.restconf_baseurl))
394 headers = {'content-type': 'application/json'}
395 response = requests.request(
396 "GET", url, headers=headers, auth=('admin', 'admin'))
397 self.assertEqual(response.status_code, requests.codes.not_found)
398 res = response.json()
400 {"error-type":"application", "error-tag":"data-missing",
401 "error-message":"Request could not be completed because the relevant data model content does not exist"},
402 res['errors']['error'])
405 if __name__ == "__main__":
406 unittest.main(verbosity=2)