2 ##############################################################################
3 # Copyright (c) 2020 Orange, Inc. and others. All rights reserved.
5 # All rights reserved. This program and the accompanying materials
6 # are made available under the terms of the Apache License, Version 2.0
7 # which accompanies this distribution, and is available at
8 # http://www.apache.org/licenses/LICENSE-2.0
9 ##############################################################################
19 RESTCONF_BASE_URL = "http://localhost:8181/restconf"
21 CODE_SHOULD_BE_200 = 'Http status code should be 200'
23 CODE_SHOULD_BE_201 = 'Http status code should be 201'
25 CREATED_SUCCESSFULLY = 'Result message should contain Xponder Roadm Link created successfully'
28 class TransportTapitesting(unittest.TestCase):
30 honeynode_process1 = None
31 honeynode_process2 = None
32 honeynode_process3 = None
33 honeynode_process4 = None
34 honeynode_process5 = None
36 # START_IGNORE_XTESTING
40 cls.init_failed = False
42 print("starting opendaylight...")
43 cls.odl_process = test_utils.start_tpce()
44 if "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
45 karaf_log = os.path.join(
46 os.path.dirname(os.path.realpath(__file__)),
47 "..", "..", "..", "karaf", "target", "assembly", "data", "log", "karaf.log")
48 searched_expr = re.escape("Blueprint container for bundle "
49 "org.opendaylight.netconf.restconf") + ".* was successfully created"
50 found = test_utils.wait_until_log_contains(karaf_log, searched_expr, time_to_wait=60)
51 cls.init_failed = not found
52 if not cls.init_failed:
53 print("opendaylight started")
54 print("installing tapi feature...")
55 result = test_utils.install_karaf_feature("odl-transportpce-tapi")
56 if result.returncode != 0:
57 cls.init_failed = True
58 print("Restarting opendaylight...")
59 test_utils.shutdown_process(cls.odl_process)
60 cls.odl_process = test_utils.start_tpce()
61 found = test_utils.wait_until_log_contains(karaf_log, searched_expr, time_to_wait=60)
62 cls.init_failed = not found
63 if not cls.init_failed:
64 print("starting XPDRA...")
65 cls.honeynode_process1 = test_utils.start_xpdra_honeynode()
67 print("starting ROADMA...")
68 cls.honeynode_process2 = test_utils.start_roadma_honeynode()
70 print("starting ROADMC...")
71 cls.honeynode_process3 = test_utils.start_roadmc_honeynode()
73 print("starting XPDRC...")
74 cls.honeynode_process4 = test_utils.start_xpdrc_honeynode()
76 print("starting SPDRA...")
77 cls.honeynode_process5 = test_utils.start_spdra_honeynode()
78 print("all honeynodes started")
81 def tearDownClass(cls):
82 test_utils.shutdown_process(cls.odl_process)
83 test_utils.shutdown_process(cls.honeynode_process1)
84 test_utils.shutdown_process(cls.honeynode_process2)
85 test_utils.shutdown_process(cls.honeynode_process3)
86 test_utils.shutdown_process(cls.honeynode_process4)
87 test_utils.shutdown_process(cls.honeynode_process5)
88 print("all processes killed")
90 def setUp(self): # instruction executed before each test method
92 self.fail('Feature installation failed')
93 print("execution of {}".format(self.id().split(".")[-1]))
97 # connect netconf devices
98 def test_00_connect_spdr_sa1(self):
99 url = ("{}/config/network-topology:"
100 "network-topology/topology/topology-netconf/node/SPDR-SA1"
101 .format(RESTCONF_BASE_URL))
102 data = test_utils.generate_connect_data("SPDR-SA1", "17845")
103 response = test_utils.put_request(url, data, 'admin', 'admin')
104 self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201) # pylint: disable=no-member
107 def test_01_connect_xpdra(self):
108 url = ("{}/config/network-topology:"
109 "network-topology/topology/topology-netconf/node/XPDR-A1"
110 .format(RESTCONF_BASE_URL))
111 data = test_utils.generate_connect_data("XPDR-A1", "17840")
112 response = test_utils.put_request(url, data, 'admin', 'admin')
113 self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201) # pylint: disable=no-member
116 def test_02_connect_xpdrc(self):
117 url = ("{}/config/network-topology:"
118 "network-topology/topology/topology-netconf/node/XPDR-C1"
119 .format(RESTCONF_BASE_URL))
120 data = test_utils.generate_connect_data("XPDR-C1", "17844")
121 response = test_utils.put_request(url, data, 'admin', 'admin')
122 self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201) # pylint: disable=no-member
125 def test_03_connect_rdma(self):
126 url = ("{}/config/network-topology:"
127 "network-topology/topology/topology-netconf/node/ROADM-A1"
128 .format(RESTCONF_BASE_URL))
129 data = test_utils.generate_connect_data("ROADM-A1", "17841")
130 response = test_utils.put_request(url, data, 'admin', 'admin')
131 self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201) # pylint: disable=no-member
134 def test_04_connect_rdmc(self):
135 url = ("{}/config/network-topology:"
136 "network-topology/topology/topology-netconf/node/ROADM-C1"
137 .format(RESTCONF_BASE_URL))
138 data = test_utils.generate_connect_data("ROADM-C1", "17843")
139 response = test_utils.put_request(url, data, 'admin', 'admin')
140 self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201) # pylint: disable=no-member
143 def test_05_connect_xprda_n1_to_roadma_pp1(self):
144 url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(RESTCONF_BASE_URL)
145 data = test_utils.generate_link_data("XPDR-A1", "1", "1", "ROADM-A1", "1", "SRG1-PP1-TXRX")
146 response = test_utils.post_request(url, data, 'admin', 'admin')
147 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
148 res = response.json()
149 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
150 CREATED_SUCCESSFULLY)
153 def test_06_connect_roadma_pp1_to_xpdra_n1(self):
154 url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(RESTCONF_BASE_URL)
155 data = test_utils.generate_link_data("XPDR-A1", "1", "1", "ROADM-A1", "1", "SRG1-PP1-TXRX")
156 response = test_utils.post_request(url, data, 'admin', 'admin')
157 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
158 res = response.json()
159 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
160 CREATED_SUCCESSFULLY)
163 def test_07_connect_xprdc_n1_to_roadmc_pp1(self):
164 url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(RESTCONF_BASE_URL)
165 data = test_utils.generate_link_data("XPDR-C1", "1", "1", "ROADM-C1", "1", "SRG1-PP1-TXRX")
166 response = test_utils.post_request(url, data, 'admin', 'admin')
167 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
168 res = response.json()
169 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
170 CREATED_SUCCESSFULLY)
173 def test_08_connect_roadmc_pp1_to_xpdrc_n1(self):
174 url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(RESTCONF_BASE_URL)
175 data = test_utils.generate_link_data("XPDR-C1", "1", "1", "ROADM-C1", "1", "SRG1-PP1-TXRX")
176 response = test_utils.post_request(url, data, 'admin', 'admin')
177 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
178 res = response.json()
179 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
180 CREATED_SUCCESSFULLY)
183 def test_09_connect_xprda_n2_to_roadma_pp2(self):
184 url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(RESTCONF_BASE_URL)
185 data = test_utils.generate_link_data("XPDR-A1", "1", "2", "ROADM-A1", "1", "SRG1-PP2-TXRX")
186 response = test_utils.post_request(url, data, 'admin', 'admin')
187 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
188 res = response.json()
189 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
190 CREATED_SUCCESSFULLY)
193 def test_10_connect_roadma_pp2_to_xpdra_n2(self):
194 url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(RESTCONF_BASE_URL)
195 data = test_utils.generate_link_data("XPDR-A1", "1", "2", "ROADM-A1", "1", "SRG1-PP2-TXRX")
196 response = test_utils.post_request(url, data, 'admin', 'admin')
197 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
198 res = response.json()
199 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
200 CREATED_SUCCESSFULLY)
203 def test_11_connect_xprdc_n2_to_roadmc_pp2(self):
204 url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(RESTCONF_BASE_URL)
205 data = test_utils.generate_link_data("XPDR-C1", "1", "2", "ROADM-C1", "1", "SRG1-PP2-TXRX")
206 response = test_utils.post_request(url, data, 'admin', 'admin')
207 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
208 res = response.json()
209 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
210 CREATED_SUCCESSFULLY)
213 def test_12_connect_roadmc_pp2_to_xpdrc_n2(self):
214 url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(RESTCONF_BASE_URL)
215 data = test_utils.generate_link_data("XPDR-C1", "1", "2", "ROADM-C1", "1", "SRG1-PP2-TXRX")
216 response = test_utils.post_request(url, data, 'admin', 'admin')
217 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
218 res = response.json()
219 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
220 CREATED_SUCCESSFULLY)
223 def test_13_get_tapi_openroadm_topology(self):
224 url = "{}/operations/tapi-topology:get-topology-details".format(RESTCONF_BASE_URL)
226 "tapi-topology:input": {
227 "tapi-topology:topology-id-or-name": "openroadm-topology"
231 response = test_utils.post_request(url, data, 'admin', 'admin')
232 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
233 res = response.json()
234 self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'There should be 1 node')
235 self.assertEqual(len(res["output"]["topology"]["node"][0]["owned-node-edge-point"]), 4,
236 'There should be 4 owned-node-edge-points')
238 def test_14_get_tapi_otn_topology(self):
239 url = "{}/operations/tapi-topology:get-topology-details".format(RESTCONF_BASE_URL)
241 "tapi-topology:input": {
242 "tapi-topology:topology-id-or-name": "otn-topology"
246 response = test_utils.post_request(url, data, 'admin', 'admin')
247 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
248 res = response.json()
249 self.assertEqual(len(res["output"]["topology"]["node"]), 4, 'There should be 4 nodes')
250 self.assertEqual(len(res["output"]["topology"]["link"]), 5, 'There should be 5 links')
251 link_to_check = res["output"]["topology"]["link"][0]
252 # get info from first link to do deeper check
253 node1_uid = link_to_check["node-edge-point"][0]["node-uuid"]
254 node2_uid = link_to_check["node-edge-point"][1]["node-uuid"]
255 node_edge_point1_uid = link_to_check["node-edge-point"][0]["node-edge-point-uuid"]
256 node_edge_point2_uid = link_to_check["node-edge-point"][1]["node-edge-point-uuid"]
257 # get node associated to link info
258 nodes = res["output"]["topology"]["node"]
259 node1 = find_object_with_key(nodes, "uuid", node1_uid)
260 self.assertIsNotNone(node1, 'Node with uuid ' + node1_uid + ' should not be null')
261 node2 = find_object_with_key(nodes, "uuid", node2_uid)
262 self.assertIsNotNone(node2, 'Node with uuid ' + node2_uid + ' should not be null')
263 # get edge-point associated to nodes
264 node1_edge_point = node1["owned-node-edge-point"]
265 node2_edge_point = node2["owned-node-edge-point"]
266 node_edge_point1 = find_object_with_key(node1_edge_point, "uuid", node_edge_point1_uid)
267 self.assertIsNotNone(node_edge_point1, 'Node edge point with uuid ' + node_edge_point1_uid + 'should not be '
269 node_edge_point2 = find_object_with_key(node2_edge_point, "uuid", node_edge_point2_uid)
270 self.assertIsNotNone(node_edge_point2, 'Node edge point with uuid ' + node_edge_point2_uid + 'should not be '
272 self.assertEqual(len(node_edge_point1["name"]), 1, 'There should be 1 name')
273 self.assertEqual(len(node_edge_point2["name"]), 1, 'There should be 1 name')
274 if node_edge_point1["layer-protocol-name"] == 'ODU':
275 self.assertIn('NodeEdgePoint_N', node_edge_point1["name"][0]["value-name"], 'Value name should be '
277 elif node_edge_point1["layer-protocol-name"] == 'PHOTONIC_MEDIA':
278 self.assertIn('iNodeEdgePoint_', node_edge_point1["name"][0]["value-name"], 'Value name should be '
281 self.fail('Wrong layer protocol name')
283 if node_edge_point2["layer-protocol-name"] == 'ODU':
284 self.assertIn('NodeEdgePoint_N', node_edge_point2["name"][0]["value-name"], 'Value name should be '
286 elif node_edge_point2["layer-protocol-name"] == 'PHOTONIC_MEDIA':
287 self.assertIn('iNodeEdgePoint_', node_edge_point2["name"][0]["value-name"], 'Value name should be '
290 self.fail('Wrong layer protocol name')
292 def test_15_disconnect_xpdra(self):
293 url = ("{}/config/network-topology:"
294 "network-topology/topology/topology-netconf/node/XPDR-A1"
295 .format(RESTCONF_BASE_URL))
297 response = test_utils.delete_request(url, 'admin', 'admin')
298 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
301 def test_16_disconnect_xpdrc(self):
302 url = ("{}/config/network-topology:"
303 "network-topology/topology/topology-netconf/node/XPDR-C1"
304 .format(RESTCONF_BASE_URL))
306 response = test_utils.delete_request(url, 'admin', 'admin')
307 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
310 def test_17_disconnect_roadma(self):
311 url = ("{}/config/network-topology:"
312 "network-topology/topology/topology-netconf/node/ROADM-A1"
313 .format(RESTCONF_BASE_URL))
315 response = test_utils.delete_request(url, 'admin', 'admin')
316 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
319 def test_18_disconnect_roadmc(self):
320 url = ("{}/config/network-topology:"
321 "network-topology/topology/topology-netconf/node/ROADM-C1"
322 .format(RESTCONF_BASE_URL))
324 response = test_utils.delete_request(url, 'admin', 'admin')
325 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
328 def test_19_disconnect_spdr_sa1(self):
329 url = ("{}/config/network-topology:"
330 "network-topology/topology/topology-netconf/node/SPDR-SA1"
331 .format(RESTCONF_BASE_URL))
332 response = test_utils.delete_request(url, 'admin', 'admin')
333 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
336 def find_object_with_key(list_dicts, key, value):
337 for dict_ in list_dicts:
338 if dict_[key] == value:
343 if __name__ == "__main__":
344 unittest.main(verbosity=2)