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'},
250 def test_11_xpdr_portmapping_NETWORK2(self):
251 url = ("{}/config/transportpce-portmapping:network/"
252 "nodes/XPDRA01/mapping/XPDR1-NETWORK2"
253 .format(self.restconf_baseurl))
254 headers = {'content-type': 'application/json'}
255 response = requests.request(
256 "GET", url, headers=headers, auth=('admin', 'admin'))
257 self.assertEqual(response.status_code, requests.codes.ok)
258 res = response.json()
260 {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/2-PLUG-NET',
261 'logical-connection-point': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional',
262 'connection-map-lcp': 'XPDR1-CLIENT3', 'port-qual': 'xpdr-network'},
265 def test_12_xpdr_portmapping_CLIENT1(self):
266 url = ("{}/config/transportpce-portmapping:network/"
267 "nodes/XPDRA01/mapping/XPDR1-CLIENT1"
268 .format(self.restconf_baseurl))
269 headers = {'content-type': 'application/json'}
270 response = requests.request(
271 "GET", url, headers=headers, auth=('admin', 'admin'))
272 self.assertEqual(response.status_code, requests.codes.ok)
273 res = response.json()
275 {'supporting-port': 'C1',
276 'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
277 'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
278 'connection-map-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client'},
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'},
297 def test_14_xpdr_portmapping_CLIENT3(self):
298 url = ("{}/config/transportpce-portmapping:network/"
299 "nodes/XPDRA01/mapping/XPDR1-CLIENT3"
300 .format(self.restconf_baseurl))
301 headers = {'content-type': 'application/json'}
302 response = requests.request(
303 "GET", url, headers=headers, auth=('admin', 'admin'))
304 self.assertEqual(response.status_code, requests.codes.ok)
305 res = response.json()
307 {'supporting-port': 'C3',
308 'supporting-circuit-pack-name': '1/0/C3-PLUG-CLIENT',
309 'logical-connection-point': 'XPDR1-CLIENT3',
310 'connection-map-lcp': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional',
311 'port-qual': 'xpdr-client'},
314 def test_15_xpdr_portmapping_CLIENT4(self):
315 url = ("{}/config/transportpce-portmapping:network/"
316 "nodes/XPDRA01/mapping/XPDR1-CLIENT4"
317 .format(self.restconf_baseurl))
318 headers = {'content-type': 'application/json'}
319 response = requests.request(
320 "GET", url, headers=headers, auth=('admin', 'admin'))
321 self.assertEqual(response.status_code, requests.codes.ok)
322 res = response.json()
324 {'supporting-port': 'C4',
325 'supporting-circuit-pack-name': '1/0/C4-PLUG-CLIENT',
326 'logical-connection-point': 'XPDR1-CLIENT4', 'port-direction': 'bidirectional',
327 'port-qual': 'xpdr-client'},
330 def test_16_xpdr_device_disconnected(self):
331 url = ("{}/config/network-topology:"
332 "network-topology/topology/topology-netconf/node/XPDRA01"
333 .format(self.restconf_baseurl))
334 headers = {'content-type': 'application/json'}
335 response = requests.request(
336 "DELETE", url, headers=headers,
337 auth=('admin', 'admin'))
338 self.assertEqual(response.status_code, requests.codes.ok)
341 def test_17_xpdr_device_disconnected(self):
342 url = ("{}/operational/network-topology:network-topology/topology/"
343 "topology-netconf/node/XPDRA01".format(self.restconf_baseurl))
344 headers = {'content-type': 'application/json'}
345 response = requests.request(
346 "GET", url, headers=headers, auth=('admin', 'admin'))
347 self.assertEqual(response.status_code, requests.codes.not_found)
348 res = response.json()
350 {"error-type": "application", "error-tag": "data-missing",
351 "error-message": "Request could not be completed because the relevant data model content does not exist"},
352 res['errors']['error'])
354 def test_18_xpdr_device_disconnected(self):
355 url = ("{}/config/transportpce-portmapping:network/nodes/XPDRA01".format(self.restconf_baseurl))
356 headers = {'content-type': 'application/json'}
357 response = requests.request(
358 "GET", url, headers=headers, auth=('admin', 'admin'))
359 self.assertEqual(response.status_code, requests.codes.not_found)
360 res = response.json()
362 {"error-type": "application", "error-tag": "data-missing",
363 "error-message": "Request could not be completed because the relevant data model content does not exist"},
364 res['errors']['error'])
366 def test_19_rdm_device_disconnected(self):
367 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADMA01"
368 .format(self.restconf_baseurl))
369 headers = {'content-type': 'application/json'}
370 response = requests.request(
371 "DELETE", url, headers=headers,
372 auth=('admin', 'admin'))
373 self.assertEqual(response.status_code, requests.codes.ok)
376 def test_20_rdm_device_disconnected(self):
377 url = ("{}/operational/network-topology:network-topology/topology/topology-netconf/node/ROADMA01"
378 .format(self.restconf_baseurl))
379 headers = {'content-type': 'application/json'}
380 response = requests.request(
381 "GET", url, headers=headers, auth=('admin', 'admin'))
382 self.assertEqual(response.status_code, requests.codes.not_found)
383 res = response.json()
385 {"error-type": "application", "error-tag": "data-missing",
386 "error-message": "Request could not be completed because the relevant data model content does not exist"},
387 res['errors']['error'])
389 def test_21_rdm_device_disconnected(self):
390 url = ("{}/config/transportpce-portmapping:network/nodes/ROADMA01".format(self.restconf_baseurl))
391 headers = {'content-type': 'application/json'}
392 response = requests.request(
393 "GET", url, headers=headers, auth=('admin', 'admin'))
394 self.assertEqual(response.status_code, requests.codes.not_found)
395 res = response.json()
397 {"error-type": "application", "error-tag": "data-missing",
398 "error-message": "Request could not be completed because the relevant data model content does not exist"},
399 res['errors']['error'])
402 if __name__ == "__main__":
403 unittest.main(verbosity=2)