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 ##############################################################################
24 class TransportPCEPortMappingTesting(unittest.TestCase):
26 honeynode_process1 = None
27 honeynode_process2 = None
29 restconf_baseurl = "http://localhost:8181/restconf"
33 print("starting honeynode1...")
34 cls.honeynode_process1 = test_utils.start_xpdra_honeynode()
37 print("starting honeynode2...")
38 cls.honeynode_process2 = test_utils.start_roadma_honeynode()
40 print("all honeynodes started")
42 print("starting opendaylight...")
43 cls.odl_process = test_utils.start_tpce()
45 print("opendaylight started")
48 def tearDownClass(cls):
49 for child in psutil.Process(cls.odl_process.pid).children():
50 child.send_signal(signal.SIGINT)
52 cls.odl_process.send_signal(signal.SIGINT)
53 cls.odl_process.wait()
54 for child in psutil.Process(cls.honeynode_process1.pid).children():
55 child.send_signal(signal.SIGINT)
57 cls.honeynode_process1.send_signal(signal.SIGINT)
58 cls.honeynode_process1.wait()
59 for child in psutil.Process(cls.honeynode_process2.pid).children():
60 child.send_signal(signal.SIGINT)
62 cls.honeynode_process2.send_signal(signal.SIGINT)
63 cls.honeynode_process2.wait()
66 print("execution of {}".format(self.id().split(".")[-1]))
69 def test_01_rdm_device_connected(self):
70 url = ("{}/config/network-topology:"
71 "network-topology/topology/topology-netconf/node/ROADM-A1"
72 .format(self.restconf_baseurl))
74 "node-id": "ROADM-A1",
75 "netconf-node-topology:username": "admin",
76 "netconf-node-topology:password": "admin",
77 "netconf-node-topology:host": "127.0.0.1",
78 "netconf-node-topology:port": "17841",
79 "netconf-node-topology:tcp-only": "false",
80 "netconf-node-topology:pass-through": {}}]}
81 headers = {'content-type': 'application/json'}
82 response = requests.request(
83 "PUT", url, data=json.dumps(data), headers=headers,
84 auth=('admin', 'admin'))
85 self.assertEqual(response.status_code, requests.codes.created)
88 def test_02_rdm_device_connected(self):
89 url = ("{}/operational/network-topology:"
90 "network-topology/topology/topology-netconf/node/ROADM-A1"
91 .format(self.restconf_baseurl))
92 headers = {'content-type': 'application/json'}
93 response = requests.request(
94 "GET", url, headers=headers, auth=('admin', 'admin'))
95 self.assertEqual(response.status_code, requests.codes.ok)
98 res['node'][0]['netconf-node-topology:connection-status'],
102 def test_03_rdm_portmapping_info(self):
103 url = ("{}/config/transportpce-portmapping:network/"
104 "nodes/ROADM-A1/node-info"
105 .format(self.restconf_baseurl))
106 headers = {'content-type': 'application/json'}
107 response = requests.request(
108 "GET", url, headers=headers, auth=('admin', 'admin'))
109 self.assertEqual(response.status_code, requests.codes.ok)
110 res = response.json()
112 {u'node-info': {u'node-type': u'rdm',
113 u'node-ip-address': u'127.0.0.11',
114 u'node-clli': u'NodeA',
115 u'openroadm-version': u'2.2.1', u'node-vendor': u'vendorA',
116 u'node-model': u'model2'}},
120 def test_04_rdm_portmapping_DEG1_TTP_TXRX(self):
121 url = ("{}/config/transportpce-portmapping:network/"
122 "nodes/ROADM-A1/mapping/DEG1-TTP-TXRX"
123 .format(self.restconf_baseurl))
124 headers = {'content-type': 'application/json'}
125 response = requests.request(
126 "GET", url, headers=headers, auth=('admin', 'admin'))
127 self.assertEqual(response.status_code, requests.codes.ok)
128 res = response.json()
130 {'supporting-port': 'L1', 'supporting-circuit-pack-name': '1/0',
131 'logical-connection-point': 'DEG1-TTP-TXRX', 'port-direction': 'bidirectional'},
134 def test_05_rdm_portmapping_DEG2_TTP_TXRX_with_ots_oms(self):
135 url = ("{}/config/transportpce-portmapping:network/"
136 "nodes/ROADM-A1/mapping/DEG2-TTP-TXRX"
137 .format(self.restconf_baseurl))
138 headers = {'content-type': 'application/json'}
139 response = requests.request(
140 "GET", url, headers=headers, auth=('admin', 'admin'))
141 self.assertEqual(response.status_code, requests.codes.ok)
142 res = response.json()
144 {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0',
145 'logical-connection-point': 'DEG2-TTP-TXRX',
146 'supporting-oms': 'OMS-DEG2-TTP-TXRX', 'supporting-ots': 'OTS-DEG2-TTP-TXRX',
147 'port-direction': 'bidirectional'},
150 def test_06_rdm_portmapping_SRG1_PP3_TXRX(self):
151 url = ("{}/config/transportpce-portmapping:network/"
152 "nodes/ROADM-A1/mapping/SRG1-PP3-TXRX"
153 .format(self.restconf_baseurl))
154 headers = {'content-type': 'application/json'}
155 response = requests.request(
156 "GET", url, headers=headers, auth=('admin', 'admin'))
157 self.assertEqual(response.status_code, requests.codes.ok)
158 res = response.json()
160 {'supporting-port': 'C3', 'supporting-circuit-pack-name': '3/0',
161 'logical-connection-point': 'SRG1-PP3-TXRX', 'port-direction': 'bidirectional'},
164 def test_07_rdm_portmapping_SRG3_PP1_TXRX(self):
165 url = ("{}/config/transportpce-portmapping:network/"
166 "nodes/ROADM-A1/mapping/SRG3-PP1-TXRX"
167 .format(self.restconf_baseurl))
168 headers = {'content-type': 'application/json'}
169 response = requests.request(
170 "GET", url, headers=headers, auth=('admin', 'admin'))
171 self.assertEqual(response.status_code, requests.codes.ok)
172 res = response.json()
174 {'supporting-port': 'C1', 'supporting-circuit-pack-name': '5/0',
175 'logical-connection-point': 'SRG3-PP1-TXRX', 'port-direction': 'bidirectional'},
178 def test_08_xpdr_device_connected(self):
179 url = ("{}/config/network-topology:"
180 "network-topology/topology/topology-netconf/node/XPDR-A1"
181 .format(self.restconf_baseurl))
183 "node-id": "XPDR-A1",
184 "netconf-node-topology:username": "admin",
185 "netconf-node-topology:password": "admin",
186 "netconf-node-topology:host": "127.0.0.1",
187 "netconf-node-topology:port": "17840",
188 "netconf-node-topology:tcp-only": "false",
189 "netconf-node-topology:pass-through": {}}]}
190 headers = {'content-type': 'application/json'}
191 response = requests.request(
192 "PUT", url, data=json.dumps(data), headers=headers,
193 auth=('admin', 'admin'))
194 self.assertEqual(response.status_code, requests.codes.created)
197 def test_09_xpdr_device_connected(self):
198 url = ("{}/operational/network-topology:"
199 "network-topology/topology/topology-netconf/node/XPDR-A1"
200 .format(self.restconf_baseurl))
201 headers = {'content-type': 'application/json'}
202 response = requests.request(
203 "GET", url, headers=headers, auth=('admin', 'admin'))
204 self.assertEqual(response.status_code, requests.codes.ok)
205 res = response.json()
207 res['node'][0]['netconf-node-topology:connection-status'],
211 def test_10_xpdr_portmapping_info(self):
212 url = ("{}/config/transportpce-portmapping:network/"
213 "nodes/XPDR-A1/node-info"
214 .format(self.restconf_baseurl))
215 headers = {'content-type': 'application/json'}
216 response = requests.request(
217 "GET", url, headers=headers, auth=('admin', 'admin'))
218 self.assertEqual(response.status_code, requests.codes.ok)
219 res = response.json()
221 {u'node-info': {u'node-type': u'xpdr',
222 u'node-ip-address': u'1.2.3.4',
223 u'node-clli': u'NodeA',
224 u'openroadm-version': u'2.2.1', u'node-vendor': u'vendorA',
225 u'node-model': u'model2'}},
229 def test_11_xpdr_portmapping_NETWORK1(self):
230 url = ("{}/config/transportpce-portmapping:network/"
231 "nodes/XPDR-A1/mapping/XPDR1-NETWORK1"
232 .format(self.restconf_baseurl))
233 headers = {'content-type': 'application/json'}
234 response = requests.request(
235 "GET", url, headers=headers, auth=('admin', 'admin'))
236 self.assertEqual(response.status_code, requests.codes.ok)
237 res = response.json()
239 {'supported-interface-capability': ['org-openroadm-port-types:if-OCH'],
240 'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
241 'logical-connection-point': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-network',
242 'port-direction': 'bidirectional', 'connection-map-lcp': 'XPDR1-CLIENT1',
243 'lcp-hash-val': '8e128ba57560403cfd4ffafae38cd941'},
246 def test_12_xpdr_portmapping_NETWORK2(self):
247 url = ("{}/config/transportpce-portmapping:network/"
248 "nodes/XPDR-A1/mapping/XPDR1-NETWORK2"
249 .format(self.restconf_baseurl))
250 headers = {'content-type': 'application/json'}
251 response = requests.request(
252 "GET", url, headers=headers, auth=('admin', 'admin'))
253 self.assertEqual(response.status_code, requests.codes.ok)
254 res = response.json()
256 {'supported-interface-capability': ['org-openroadm-port-types:if-OCH'],
257 'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/2-PLUG-NET',
258 'logical-connection-point': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional',
259 'connection-map-lcp': 'XPDR1-CLIENT2', 'port-qual': 'xpdr-network',
260 'lcp-hash-val': '8e128ba57560403cfd4ffafae38cd942'},
263 def test_13_xpdr_portmapping_CLIENT1(self):
264 url = ("{}/config/transportpce-portmapping:network/"
265 "nodes/XPDR-A1/mapping/XPDR1-CLIENT1"
266 .format(self.restconf_baseurl))
267 headers = {'content-type': 'application/json'}
268 response = requests.request(
269 "GET", url, headers=headers, auth=('admin', 'admin'))
270 self.assertEqual(response.status_code, requests.codes.ok)
271 res = response.json()
273 {'supported-interface-capability': ['org-openroadm-port-types:if-100GE'],
274 'supporting-port': 'C1',
275 'supporting-circuit-pack-name': '1/0/1-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': '3ed8ed1336784ac7c2f66c22f2f03d8'},
281 def test_14_xpdr_portmapping_CLIENT2(self):
282 url = ("{}/config/transportpce-portmapping:network/"
283 "nodes/XPDR-A1/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 {'supported-interface-capability': ['org-openroadm-port-types:if-100GE'],
292 'supporting-port': 'C1',
293 'supporting-circuit-pack-name': '1/0/2-PLUG-CLIENT',
294 'logical-connection-point': 'XPDR1-CLIENT2', 'port-direction': 'bidirectional',
295 'connection-map-lcp': 'XPDR1-NETWORK2', 'port-qual': 'xpdr-client',
296 'lcp-hash-val': '3ed8ed1336784ac7c2f66c22f2f03db'},
299 def test_15_xpdr_device_disconnected(self):
300 url = ("{}/config/network-topology:"
301 "network-topology/topology/topology-netconf/node/XPDR-A1"
302 .format(self.restconf_baseurl))
303 headers = {'content-type': 'application/json'}
304 response = requests.request(
305 "DELETE", url, headers=headers,
306 auth=('admin', 'admin'))
307 self.assertEqual(response.status_code, requests.codes.ok)
310 def test_16_xpdr_device_disconnected(self):
311 url = ("{}/operational/network-topology:network-topology/topology/"
312 "topology-netconf/node/XPDR-A1".format(self.restconf_baseurl))
313 headers = {'content-type': 'application/json'}
314 response = requests.request(
315 "GET", url, headers=headers, auth=('admin', 'admin'))
316 self.assertEqual(response.status_code, requests.codes.not_found)
317 res = response.json()
319 {"error-type": "application", "error-tag": "data-missing",
320 "error-message": "Request could not be completed because the relevant data model content does not exist"},
321 res['errors']['error'])
323 def test_17_xpdr_device_disconnected(self):
324 url = ("{}/config/transportpce-portmapping:network/nodes/XPDR-A1".format(self.restconf_baseurl))
325 headers = {'content-type': 'application/json'}
326 response = requests.request(
327 "GET", url, headers=headers, auth=('admin', 'admin'))
328 self.assertEqual(response.status_code, requests.codes.not_found)
329 res = response.json()
331 {"error-type": "application", "error-tag": "data-missing",
332 "error-message": "Request could not be completed because the relevant data model content does not exist"},
333 res['errors']['error'])
335 def test_18_rdm_device_disconnected(self):
336 url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADM-A1"
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_19_rdm_device_disconnected(self):
346 url = ("{}/operational/network-topology:network-topology/topology/topology-netconf/node/ROADM-A1"
347 .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_20_rdm_device_disconnected(self):
359 url = ("{}/config/transportpce-portmapping:network/nodes/ROADM-A1".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'])
371 if __name__ == "__main__":
372 unittest.main(verbosity=2)