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
39 @unittest.skipIf("USE_LIGHTY" in os.environ and os.environ['USE_LIGHTY'] == 'True',
40 "not supported for lighty")
42 cls.init_failed = False
43 karaf_log = os.path.join(
44 os.path.dirname(os.path.realpath(__file__)),
45 "..", "..", "..", "karaf", "target", "assembly", "data", "log", "karaf.log")
46 searched_expr = re.escape("Blueprint container for bundle "
47 "org.opendaylight.netconf.restconf") + ".* was successfully created"
49 print("starting opendaylight...")
50 cls.odl_process = test_utils.start_tpce()
51 found = test_utils.wait_until_log_contains(karaf_log, searched_expr, time_to_wait=60)
52 cls.init_failed = not found
53 if not cls.init_failed:
54 print("opendaylight started")
56 print("installing tapi feature...")
57 result = test_utils.install_karaf_feature("odl-transportpce-tapi")
58 if result.returncode != 0:
59 cls.init_failed = True
60 print("Restarting opendaylight...")
61 test_utils.shutdown_process(cls.odl_process)
62 cls.odl_process = test_utils.start_tpce()
63 found = test_utils.wait_until_log_contains(karaf_log, searched_expr, time_to_wait=60)
64 cls.init_failed = not found
65 if not cls.init_failed:
66 print("starting XPDRA...")
67 cls.honeynode_process1 = test_utils.start_xpdra_honeynode()
69 print("starting ROADMA...")
70 cls.honeynode_process2 = test_utils.start_roadma_honeynode()
72 print("starting ROADMC...")
73 cls.honeynode_process3 = test_utils.start_roadmc_honeynode()
75 print("starting XPDRC...")
76 cls.honeynode_process4 = test_utils.start_xpdrc_honeynode()
78 print("starting SPDRA...")
79 cls.honeynode_process5 = test_utils.start_spdra_honeynode()
80 print("all honeynodes started")
83 def tearDownClass(cls):
84 test_utils.shutdown_process(cls.odl_process)
85 test_utils.shutdown_process(cls.honeynode_process1)
86 test_utils.shutdown_process(cls.honeynode_process2)
87 test_utils.shutdown_process(cls.honeynode_process3)
88 test_utils.shutdown_process(cls.honeynode_process4)
89 test_utils.shutdown_process(cls.honeynode_process5)
90 print("all processes killed")
92 def setUp(self): # instruction executed before each test method
94 self.fail('Feature installation failed')
95 print("execution of {}".format(self.id().split(".")[-1]))
99 # connect netconf devices
100 def test_00_connect_spdr_sa1(self):
101 url = ("{}/config/network-topology:"
102 "network-topology/topology/topology-netconf/node/SPDR-SA1"
103 .format(RESTCONF_BASE_URL))
104 data = test_utils.generate_connect_data("SPDR-SA1", "17845")
105 response = test_utils.put_request(url, data, 'admin', 'admin')
106 self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201) # pylint: disable=no-member
109 def test_01_connect_xpdra(self):
110 url = ("{}/config/network-topology:"
111 "network-topology/topology/topology-netconf/node/XPDR-A1"
112 .format(RESTCONF_BASE_URL))
113 data = test_utils.generate_connect_data("XPDR-A1", "17840")
114 response = test_utils.put_request(url, data, 'admin', 'admin')
115 self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201) # pylint: disable=no-member
118 def test_02_connect_xpdrc(self):
119 url = ("{}/config/network-topology:"
120 "network-topology/topology/topology-netconf/node/XPDR-C1"
121 .format(RESTCONF_BASE_URL))
122 data = test_utils.generate_connect_data("XPDR-C1", "17844")
123 response = test_utils.put_request(url, data, 'admin', 'admin')
124 self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201) # pylint: disable=no-member
127 def test_03_connect_rdma(self):
128 url = ("{}/config/network-topology:"
129 "network-topology/topology/topology-netconf/node/ROADM-A1"
130 .format(RESTCONF_BASE_URL))
131 data = test_utils.generate_connect_data("ROADM-A1", "17841")
132 response = test_utils.put_request(url, data, 'admin', 'admin')
133 self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201) # pylint: disable=no-member
136 def test_04_connect_rdmc(self):
137 url = ("{}/config/network-topology:"
138 "network-topology/topology/topology-netconf/node/ROADM-C1"
139 .format(RESTCONF_BASE_URL))
140 data = test_utils.generate_connect_data("ROADM-C1", "17843")
141 response = test_utils.put_request(url, data, 'admin', 'admin')
142 self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201) # pylint: disable=no-member
145 def test_05_connect_xprda_n1_to_roadma_pp1(self):
146 url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(RESTCONF_BASE_URL)
147 data = test_utils.generate_link_data("XPDR-A1", "1", "1", "ROADM-A1", "1", "SRG1-PP1-TXRX")
148 response = test_utils.post_request(url, data, 'admin', 'admin')
149 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
150 res = response.json()
151 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
152 CREATED_SUCCESSFULLY)
155 def test_06_connect_roadma_pp1_to_xpdra_n1(self):
156 url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(RESTCONF_BASE_URL)
157 data = test_utils.generate_link_data("XPDR-A1", "1", "1", "ROADM-A1", "1", "SRG1-PP1-TXRX")
158 response = test_utils.post_request(url, data, 'admin', 'admin')
159 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
160 res = response.json()
161 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
162 CREATED_SUCCESSFULLY)
165 def test_07_connect_xprdc_n1_to_roadmc_pp1(self):
166 url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(RESTCONF_BASE_URL)
167 data = test_utils.generate_link_data("XPDR-C1", "1", "1", "ROADM-C1", "1", "SRG1-PP1-TXRX")
168 response = test_utils.post_request(url, data, 'admin', 'admin')
169 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
170 res = response.json()
171 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
172 CREATED_SUCCESSFULLY)
175 def test_08_connect_roadmc_pp1_to_xpdrc_n1(self):
176 url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(RESTCONF_BASE_URL)
177 data = test_utils.generate_link_data("XPDR-C1", "1", "1", "ROADM-C1", "1", "SRG1-PP1-TXRX")
178 response = test_utils.post_request(url, data, 'admin', 'admin')
179 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
180 res = response.json()
181 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
182 CREATED_SUCCESSFULLY)
185 def test_09_connect_xprda_n2_to_roadma_pp2(self):
186 url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(RESTCONF_BASE_URL)
187 data = test_utils.generate_link_data("XPDR-A1", "1", "2", "ROADM-A1", "1", "SRG1-PP2-TXRX")
188 response = test_utils.post_request(url, data, 'admin', 'admin')
189 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
190 res = response.json()
191 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
192 CREATED_SUCCESSFULLY)
195 def test_10_connect_roadma_pp2_to_xpdra_n2(self):
196 url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(RESTCONF_BASE_URL)
197 data = test_utils.generate_link_data("XPDR-A1", "1", "2", "ROADM-A1", "1", "SRG1-PP2-TXRX")
198 response = test_utils.post_request(url, data, 'admin', 'admin')
199 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
200 res = response.json()
201 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
202 CREATED_SUCCESSFULLY)
205 def test_11_connect_xprdc_n2_to_roadmc_pp2(self):
206 url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(RESTCONF_BASE_URL)
207 data = test_utils.generate_link_data("XPDR-C1", "1", "2", "ROADM-C1", "1", "SRG1-PP2-TXRX")
208 response = test_utils.post_request(url, data, 'admin', 'admin')
209 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
210 res = response.json()
211 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
212 CREATED_SUCCESSFULLY)
215 def test_12_connect_roadmc_pp2_to_xpdrc_n2(self):
216 url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(RESTCONF_BASE_URL)
217 data = test_utils.generate_link_data("XPDR-C1", "1", "2", "ROADM-C1", "1", "SRG1-PP2-TXRX")
218 response = test_utils.post_request(url, data, 'admin', 'admin')
219 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
220 res = response.json()
221 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
222 CREATED_SUCCESSFULLY)
225 def test_13_get_tapi_openroadm_topology(self):
226 url = "{}/operations/tapi-topology:get-topology-details".format(RESTCONF_BASE_URL)
228 "tapi-topology:input": {
229 "tapi-topology:topology-id-or-name": "openroadm-topology"
233 response = test_utils.post_request(url, data, 'admin', 'admin')
234 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
235 res = response.json()
236 self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'There should be 1 node')
237 self.assertEqual(len(res["output"]["topology"]["node"][0]["owned-node-edge-point"]), 4,
238 'There should be 4 owned-node-edge-points')
240 def test_14_get_tapi_otn_topology(self):
241 url = "{}/operations/tapi-topology:get-topology-details".format(RESTCONF_BASE_URL)
243 "tapi-topology:input": {
244 "tapi-topology:topology-id-or-name": "otn-topology"
248 response = test_utils.post_request(url, data, 'admin', 'admin')
249 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
250 res = response.json()
251 self.assertEqual(len(res["output"]["topology"]["node"]), 4, 'There should be 4 nodes')
252 self.assertEqual(len(res["output"]["topology"]["link"]), 5, 'There should be 5 links')
253 link_to_check = res["output"]["topology"]["link"][0]
254 # get info from first link to do deeper check
255 node1_uid = link_to_check["node-edge-point"][0]["node-uuid"]
256 node2_uid = link_to_check["node-edge-point"][1]["node-uuid"]
257 node_edge_point1_uid = link_to_check["node-edge-point"][0]["node-edge-point-uuid"]
258 node_edge_point2_uid = link_to_check["node-edge-point"][1]["node-edge-point-uuid"]
259 # get node associated to link info
260 nodes = res["output"]["topology"]["node"]
261 node1 = find_object_with_key(nodes, "uuid", node1_uid)
262 self.assertIsNotNone(node1, 'Node with uuid ' + node1_uid + ' should not be null')
263 node2 = find_object_with_key(nodes, "uuid", node2_uid)
264 self.assertIsNotNone(node2, 'Node with uuid ' + node2_uid + ' should not be null')
265 # get edge-point associated to nodes
266 node1_edge_point = node1["owned-node-edge-point"]
267 node2_edge_point = node2["owned-node-edge-point"]
268 node_edge_point1 = find_object_with_key(node1_edge_point, "uuid", node_edge_point1_uid)
269 self.assertIsNotNone(node_edge_point1, 'Node edge point with uuid ' + node_edge_point1_uid + 'should not be '
271 node_edge_point2 = find_object_with_key(node2_edge_point, "uuid", node_edge_point2_uid)
272 self.assertIsNotNone(node_edge_point2, 'Node edge point with uuid ' + node_edge_point2_uid + 'should not be '
274 self.assertEqual(len(node_edge_point1["name"]), 1, 'There should be 1 name')
275 self.assertEqual(len(node_edge_point2["name"]), 1, 'There should be 1 name')
276 if node_edge_point1["layer-protocol-name"] == 'ODU':
277 self.assertIn('NodeEdgePoint_N', node_edge_point1["name"][0]["value-name"], 'Value name should be '
279 elif node_edge_point1["layer-protocol-name"] == 'PHOTONIC_MEDIA':
280 self.assertIn('iNodeEdgePoint_', node_edge_point1["name"][0]["value-name"], 'Value name should be '
283 self.fail('Wrong layer protocol name')
285 if node_edge_point2["layer-protocol-name"] == 'ODU':
286 self.assertIn('NodeEdgePoint_N', node_edge_point2["name"][0]["value-name"], 'Value name should be '
288 elif node_edge_point2["layer-protocol-name"] == 'PHOTONIC_MEDIA':
289 self.assertIn('iNodeEdgePoint_', node_edge_point2["name"][0]["value-name"], 'Value name should be '
292 self.fail('Wrong layer protocol name')
294 def test_15_disconnect_xpdra(self):
295 url = ("{}/config/network-topology:"
296 "network-topology/topology/topology-netconf/node/XPDR-A1"
297 .format(RESTCONF_BASE_URL))
299 response = test_utils.delete_request(url, 'admin', 'admin')
300 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
303 def test_16_disconnect_xpdrc(self):
304 url = ("{}/config/network-topology:"
305 "network-topology/topology/topology-netconf/node/XPDR-C1"
306 .format(RESTCONF_BASE_URL))
308 response = test_utils.delete_request(url, 'admin', 'admin')
309 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
312 def test_17_disconnect_roadma(self):
313 url = ("{}/config/network-topology:"
314 "network-topology/topology/topology-netconf/node/ROADM-A1"
315 .format(RESTCONF_BASE_URL))
317 response = test_utils.delete_request(url, 'admin', 'admin')
318 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
321 def test_18_disconnect_roadmc(self):
322 url = ("{}/config/network-topology:"
323 "network-topology/topology/topology-netconf/node/ROADM-C1"
324 .format(RESTCONF_BASE_URL))
326 response = test_utils.delete_request(url, 'admin', 'admin')
327 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
330 def test_19_disconnect_spdr_sa1(self):
331 url = ("{}/config/network-topology:"
332 "network-topology/topology/topology-netconf/node/SPDR-SA1"
333 .format(RESTCONF_BASE_URL))
334 response = test_utils.delete_request(url, 'admin', 'admin')
335 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
338 def find_object_with_key(list_dicts, key, value):
339 for dict_ in list_dicts:
340 if dict_[key] == value:
345 if __name__ == "__main__":
346 unittest.main(verbosity=2)