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):
23 honeynode_process1 = None
24 honeynode_process2 = None
26 restconf_baseurl = "http://localhost:8181/restconf"
28 # START_IGNORE_XTESTING
32 print("starting honeynode1...")
33 cls.honeynode_process1 = test_utils.start_xpdra_honeynode()
36 print("starting honeynode2...")
37 cls.honeynode_process2 = test_utils.start_roadma_honeynode()
40 print("starting opendaylight...")
41 cls.odl_process = test_utils.start_tpce()
43 print("opendaylight started")
46 def tearDownClass(cls):
47 for child in psutil.Process(cls.odl_process.pid).children():
48 child.send_signal(signal.SIGINT)
50 cls.odl_process.send_signal(signal.SIGINT)
51 cls.odl_process.wait()
52 for child in psutil.Process(cls.honeynode_process1.pid).children():
53 child.send_signal(signal.SIGINT)
55 cls.honeynode_process1.send_signal(signal.SIGINT)
56 cls.honeynode_process1.wait()
57 for child in psutil.Process(cls.honeynode_process2.pid).children():
58 child.send_signal(signal.SIGINT)
60 cls.honeynode_process2.send_signal(signal.SIGINT)
61 cls.honeynode_process2.wait()
64 print("execution of {}".format(self.id().split(".")[-1]))
69 # def test_01_restconfAPI(self):
70 # url = ("{}/operational/network-topology:network-topology/topology/"
71 # "topology-netconf/node/controller-config".format(self.restconf_baseurl))
72 # headers = {'content-type': 'application/json'}
73 # response = requests.request("GET", url, headers=headers, auth=('admin', 'admin'))
74 # self.assertEqual(response.status_code, requests.codes.ok)
75 # res = response.json()
76 # self.assertEqual(res['node'] [0] ['netconf-node-topology:connection-status'],
79 # def test_02_restconfAPI(self):
80 # url = ("{}/config/transportpce-portmapping:network/nodes/controller-config".format(self.restconf_baseurl))
81 # headers = {'content-type': 'application/json'}
82 # response = requests.request(
83 # "GET", url, headers=headers, auth=('admin', 'admin'))
84 # self.assertEqual(response.status_code, requests.codes.not_found)
85 # res = response.json()
87 # {"error-type":"application", "error-tag":"data-missing",
88 # "error-message":"Request could not be completed because the relevant data model content does not exist "},
89 # res['errors']['error'])
91 def test_01_rdm_device_connected(self):
92 url = ("{}/config/network-topology:"
93 "network-topology/topology/topology-netconf/node/ROADMA01"
94 .format(self.restconf_baseurl))
96 "node-id": "ROADMA01",
97 "netconf-node-topology:username": "admin",
98 "netconf-node-topology:password": "admin",
99 "netconf-node-topology:host": "127.0.0.1",
100 "netconf-node-topology:port": "17831",
101 "netconf-node-topology:tcp-only": "false",
102 "netconf-node-topology:pass-through": {}}]}
103 headers = {'content-type': 'application/json'}
104 response = requests.request(
105 "PUT", url, data=json.dumps(data), headers=headers,
106 auth=('admin', 'admin'))
107 self.assertEqual(response.status_code, requests.codes.created)
110 def test_02_rdm_device_connected(self):
111 url = ("{}/operational/network-topology:"
112 "network-topology/topology/topology-netconf/node/ROADMA01"
113 .format(self.restconf_baseurl))
114 headers = {'content-type': 'application/json'}
115 response = requests.request(
116 "GET", url, headers=headers, auth=('admin', 'admin'))
117 self.assertEqual(response.status_code, requests.codes.ok)
118 res = response.json()
120 res['node'][0]['netconf-node-topology:connection-status'],
124 def test_03_rdm_portmapping_info(self):
125 url = ("{}/config/transportpce-portmapping:network/"
126 "nodes/ROADMA01/node-info"
127 .format(self.restconf_baseurl))
128 headers = {'content-type': 'application/json'}
129 response = requests.request(
130 "GET", url, headers=headers, auth=('admin', 'admin'))
131 self.assertEqual(response.status_code, requests.codes.ok)
132 res = response.json()
134 {u'node-info': {u'node-type': u'rdm',
135 u'node-ip-address': u'127.0.0.12',
136 u'node-clli': u'NodeA',
137 u'openroadm-version': u'1.2.1', u'node-vendor': u'vendorA',
138 u'node-model': u'2'}},
142 def test_04_rdm_portmapping_DEG1_TTP_TXRX(self):
143 url = ("{}/config/transportpce-portmapping:network/"
144 "nodes/ROADMA01/mapping/DEG1-TTP-TXRX"
145 .format(self.restconf_baseurl))
146 headers = {'content-type': 'application/json'}
147 response = requests.request(
148 "GET", url, headers=headers, auth=('admin', 'admin'))
149 self.assertEqual(response.status_code, requests.codes.ok)
150 res = response.json()
152 {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0',
153 'logical-connection-point': 'DEG1-TTP-TXRX', 'port-direction': 'bidirectional'},
156 def test_05_rdm_portmapping_SRG1_PP7_TXRX(self):
157 url = ("{}/config/transportpce-portmapping:network/"
158 "nodes/ROADMA01/mapping/SRG1-PP7-TXRX"
159 .format(self.restconf_baseurl))
160 headers = {'content-type': 'application/json'}
161 response = requests.request(
162 "GET", url, headers=headers, auth=('admin', 'admin'))
163 self.assertEqual(response.status_code, requests.codes.ok)
164 res = response.json()
166 {'supporting-port': 'C7', 'supporting-circuit-pack-name': '4/0',
167 'logical-connection-point': 'SRG1-PP7-TXRX', 'port-direction': 'bidirectional'},
170 def test_06_rdm_portmapping_SRG3_PP1_TXRX(self):
171 url = ("{}/config/transportpce-portmapping:network/"
172 "nodes/ROADMA01/mapping/SRG3-PP1-TXRX"
173 .format(self.restconf_baseurl))
174 headers = {'content-type': 'application/json'}
175 response = requests.request(
176 "GET", url, headers=headers, auth=('admin', 'admin'))
177 self.assertEqual(response.status_code, requests.codes.ok)
178 res = response.json()
180 {'supporting-port': 'C1', 'supporting-circuit-pack-name': '5/0',
181 'logical-connection-point': 'SRG3-PP1-TXRX', 'port-direction': 'bidirectional'},
184 def test_07_xpdr_device_connected(self):
185 url = ("{}/config/network-topology:"
186 "network-topology/topology/topology-netconf/node/XPDRA01"
187 .format(self.restconf_baseurl))
189 "node-id": "XPDRA01",
190 "netconf-node-topology:username": "admin",
191 "netconf-node-topology:password": "admin",
192 "netconf-node-topology:host": "127.0.0.1",
193 "netconf-node-topology:port": "17830",
194 "netconf-node-topology:tcp-only": "false",
195 "netconf-node-topology:pass-through": {}}]}
196 headers = {'content-type': 'application/json'}
197 response = requests.request(
198 "PUT", url, data=json.dumps(data), headers=headers,
199 auth=('admin', 'admin'))
200 self.assertEqual(response.status_code, requests.codes.created)
203 def test_08_xpdr_device_connected(self):
204 url = ("{}/operational/network-topology:"
205 "network-topology/topology/topology-netconf/node/XPDRA01"
206 .format(self.restconf_baseurl))
207 headers = {'content-type': 'application/json'}
208 response = requests.request(
209 "GET", url, headers=headers, auth=('admin', 'admin'))
210 self.assertEqual(response.status_code, requests.codes.ok)
211 res = response.json()
213 res['node'][0]['netconf-node-topology:connection-status'],
217 def test_09_xpdr_portmapping_info(self):
218 url = ("{}/config/transportpce-portmapping:network/"
219 "nodes/XPDRA01/node-info"
220 .format(self.restconf_baseurl))
221 headers = {'content-type': 'application/json'}
222 response = requests.request(
223 "GET", url, headers=headers, auth=('admin', 'admin'))
224 self.assertEqual(response.status_code, requests.codes.ok)
225 res = response.json()
227 {u'node-info': {u'node-type': u'xpdr',
228 u'node-ip-address': u'127.0.0.10',
229 u'node-clli': u'NodeA',
230 u'openroadm-version': u'1.2.1', u'node-vendor': u'vendorA',
231 u'node-model': u'1'}},
235 def test_10_xpdr_portmapping_NETWORK1(self):
236 url = ("{}/config/transportpce-portmapping:network/"
237 "nodes/XPDRA01/mapping/XPDR1-NETWORK1"
238 .format(self.restconf_baseurl))
239 headers = {'content-type': 'application/json'}
240 response = requests.request(
241 "GET", url, headers=headers, auth=('admin', 'admin'))
242 self.assertEqual(response.status_code, requests.codes.ok)
243 res = response.json()
245 {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
246 'logical-connection-point': 'XPDR1-NETWORK1', 'port-direction': 'bidirectional',
247 'connection-map-lcp': 'XPDR1-CLIENT1', 'port-qual': 'xpdr-network',
248 'lcp-hash-val': '3b3ab304d2a6eb3c3623e52746dbb7aa'},
251 def test_11_xpdr_portmapping_NETWORK2(self):
252 url = ("{}/config/transportpce-portmapping:network/"
253 "nodes/XPDRA01/mapping/XPDR1-NETWORK2"
254 .format(self.restconf_baseurl))
255 headers = {'content-type': 'application/json'}
256 response = requests.request(
257 "GET", url, headers=headers, auth=('admin', 'admin'))
258 self.assertEqual(response.status_code, requests.codes.ok)
259 res = response.json()
261 {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/2-PLUG-NET',
262 'logical-connection-point': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional',
263 'connection-map-lcp': 'XPDR1-CLIENT3', 'port-qual': 'xpdr-network',
264 'lcp-hash-val': '3b3ab304d2a6eb3c3623e52746dbb7a9'},
267 def test_12_xpdr_portmapping_CLIENT1(self):
268 url = ("{}/config/transportpce-portmapping:network/"
269 "nodes/XPDRA01/mapping/XPDR1-CLIENT1"
270 .format(self.restconf_baseurl))
271 headers = {'content-type': 'application/json'}
272 response = requests.request(
273 "GET", url, headers=headers, auth=('admin', 'admin'))
274 self.assertEqual(response.status_code, requests.codes.ok)
275 res = response.json()
277 {'supporting-port': 'C1',
278 'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
279 'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
280 'connection-map-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client',
281 'lcp-hash-val': '64b8effe7ba72211420bf267d0ca1ae5'},
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',
298 'lcp-hash-val': '64b8effe7ba72211420bf267d0ca1ae6'},
301 def test_14_xpdr_portmapping_CLIENT3(self):
302 url = ("{}/config/transportpce-portmapping:network/"
303 "nodes/XPDRA01/mapping/XPDR1-CLIENT3"
304 .format(self.restconf_baseurl))
305 headers = {'content-type': 'application/json'}
306 response = requests.request(
307 "GET", url, headers=headers, auth=('admin', 'admin'))
308 self.assertEqual(response.status_code, requests.codes.ok)
309 res = response.json()
311 {'supporting-port': 'C3',
312 'supporting-circuit-pack-name': '1/0/C3-PLUG-CLIENT',
313 'logical-connection-point': 'XPDR1-CLIENT3',
314 'connection-map-lcp': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional',
315 'port-qual': 'xpdr-client' , 'lcp-hash-val': '64b8effe7ba72211420bf267d0ca1ae7'},
318 def test_15_xpdr_portmapping_CLIENT4(self):
319 url = ("{}/config/transportpce-portmapping:network/"
320 "nodes/XPDRA01/mapping/XPDR1-CLIENT4"
321 .format(self.restconf_baseurl))
322 headers = {'content-type': 'application/json'}
323 response = requests.request(
324 "GET", url, headers=headers, auth=('admin', 'admin'))
325 self.assertEqual(response.status_code, requests.codes.ok)
326 res = response.json()
328 {'supporting-port': 'C4',
329 'supporting-circuit-pack-name': '1/0/C4-PLUG-CLIENT',
330 'logical-connection-point': 'XPDR1-CLIENT4', 'port-direction': 'bidirectional',
331 'port-qual': 'xpdr-client' , 'lcp-hash-val': '64b8effe7ba72211420bf267d0ca1ae0'},
334 def test_16_xpdr_device_disconnected(self):
335 url = ("{}/config/network-topology:"
336 "network-topology/topology/topology-netconf/node/XPDRA01"
337 .format(self.restconf_baseurl))
338 headers = {'content-type': 'application/json'}
339 response = requests.request(
340 "DELETE", url, headers=headers,
341 auth=('admin', 'admin'))
342 self.assertEqual(response.status_code, requests.codes.ok)
345 def test_17_xpdr_device_disconnected(self):
346 url = ("{}/operational/network-topology:network-topology/topology/"
347 "topology-netconf/node/XPDRA01".format(self.restconf_baseurl))
348 headers = {'content-type': 'application/json'}
349 response = requests.request(
350 "GET", url, headers=headers, auth=('admin', 'admin'))
351 self.assertEqual(response.status_code, requests.codes.not_found)
352 res = response.json()
354 {"error-type": "application", "error-tag": "data-missing",
355 "error-message": "Request could not be completed because the relevant data model content does not exist"},
356 res['errors']['error'])
358 def test_18_xpdr_device_disconnected(self):
359 url = ("{}/config/transportpce-portmapping:network/nodes/XPDRA01".format(self.restconf_baseurl))
360 headers = {'content-type': 'application/json'}
361 response = requests.request(
362 "GET", url, headers=headers, auth=('admin', 'admin'))
363 self.assertEqual(response.status_code, requests.codes.not_found)
364 res = response.json()
366 {"error-type": "application", "error-tag": "data-missing",
367 "error-message": "Request could not be completed because the relevant data model content does not exist"},
368 res['errors']['error'])
370 def test_19_rdm_device_disconnected(self):
371 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADMA01"
372 .format(self.restconf_baseurl))
373 headers = {'content-type': 'application/json'}
374 response = requests.request(
375 "DELETE", url, headers=headers,
376 auth=('admin', 'admin'))
377 self.assertEqual(response.status_code, requests.codes.ok)
380 def test_20_rdm_device_disconnected(self):
381 url = ("{}/operational/network-topology:network-topology/topology/topology-netconf/node/ROADMA01"
382 .format(self.restconf_baseurl))
383 headers = {'content-type': 'application/json'}
384 response = requests.request(
385 "GET", url, headers=headers, auth=('admin', 'admin'))
386 self.assertEqual(response.status_code, requests.codes.not_found)
387 res = response.json()
389 {"error-type": "application", "error-tag": "data-missing",
390 "error-message": "Request could not be completed because the relevant data model content does not exist"},
391 res['errors']['error'])
393 def test_21_rdm_device_disconnected(self):
394 url = ("{}/config/transportpce-portmapping:network/nodes/ROADMA01".format(self.restconf_baseurl))
395 headers = {'content-type': 'application/json'}
396 response = requests.request(
397 "GET", url, headers=headers, auth=('admin', 'admin'))
398 self.assertEqual(response.status_code, requests.codes.not_found)
399 res = response.json()
401 {"error-type": "application", "error-tag": "data-missing",
402 "error-message": "Request could not be completed because the relevant data model content does not exist"},
403 res['errors']['error'])
406 if __name__ == "__main__":
407 unittest.main(verbosity=2)