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):
36 # START_IGNORE_XTESTING
40 cls.init_failed = False
42 cls.odl_process = test_utils.start_tpce()
43 if "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
44 karaf_log = os.path.join(
45 os.path.dirname(os.path.realpath(__file__)),
46 "..", "..", "..", "karaf", "target", "assembly", "data", "log", "karaf.log")
47 searched_expr = re.escape("Blueprint container for bundle "
48 "org.opendaylight.netconf.restconf") + ".* was successfully created"
49 found = test_utils.wait_until_log_contains(karaf_log, searched_expr, time_to_wait=60)
50 cls.init_failed = not found
51 if not cls.init_failed:
52 print("opendaylight started")
53 print("installing tapi feature...")
54 result = test_utils.install_karaf_feature("odl-transportpce-tapi")
55 if result.returncode != 0:
56 cls.init_failed = True
57 print("Restarting opendaylight...")
58 test_utils.shutdown_process(cls.odl_process)
59 cls.odl_process = test_utils.start_tpce()
60 found = test_utils.wait_until_log_contains(karaf_log, searched_expr, time_to_wait=60)
61 cls.init_failed = not found
62 if not cls.init_failed:
63 cls.sim_process1 = test_utils.start_sim('xpdra')
65 cls.sim_process2 = test_utils.start_sim('roadma')
67 cls.sim_process3 = test_utils.start_sim('roadmc')
69 cls.sim_process4 = test_utils.start_sim('xpdrc')
71 cls.sim_process5 = test_utils.start_sim('spdrav2')
72 print("all sims started")
75 def tearDownClass(cls):
76 test_utils.shutdown_process(cls.odl_process)
77 test_utils.shutdown_process(cls.sim_process1)
78 test_utils.shutdown_process(cls.sim_process2)
79 test_utils.shutdown_process(cls.sim_process3)
80 test_utils.shutdown_process(cls.sim_process4)
81 test_utils.shutdown_process(cls.sim_process5)
82 print("all processes killed")
84 def setUp(self): # instruction executed before each test method
86 self.fail('Feature installation failed')
87 print("execution of {}".format(self.id().split(".")[-1]))
91 # connect netconf devices
92 def test_00_connect_spdr_sa1(self):
93 url = ("{}/config/network-topology:"
94 "network-topology/topology/topology-netconf/node/SPDR-SA1"
95 .format(RESTCONF_BASE_URL))
96 data = test_utils.generate_connect_data("SPDR-SA1", test_utils.sims['spdrav2']['port'])
97 response = test_utils.put_request(url, data, 'admin', 'admin')
98 self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201) # pylint: disable=no-member
101 def test_01_connect_xpdra(self):
102 url = ("{}/config/network-topology:"
103 "network-topology/topology/topology-netconf/node/XPDR-A1"
104 .format(RESTCONF_BASE_URL))
105 data = test_utils.generate_connect_data("XPDR-A1", test_utils.sims['xpdra']['port'])
106 response = test_utils.put_request(url, data, 'admin', 'admin')
107 self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201) # pylint: disable=no-member
110 def test_02_connect_xpdrc(self):
111 url = ("{}/config/network-topology:"
112 "network-topology/topology/topology-netconf/node/XPDR-C1"
113 .format(RESTCONF_BASE_URL))
114 data = test_utils.generate_connect_data("XPDR-C1", test_utils.sims['xpdrc']['port'])
115 response = test_utils.put_request(url, data, 'admin', 'admin')
116 self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201) # pylint: disable=no-member
119 def test_03_connect_rdma(self):
120 url = ("{}/config/network-topology:"
121 "network-topology/topology/topology-netconf/node/ROADM-A1"
122 .format(RESTCONF_BASE_URL))
123 data = test_utils.generate_connect_data("ROADM-A1", test_utils.sims['roadma']['port'])
124 response = test_utils.put_request(url, data, 'admin', 'admin')
125 self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201) # pylint: disable=no-member
128 def test_04_connect_rdmc(self):
129 url = ("{}/config/network-topology:"
130 "network-topology/topology/topology-netconf/node/ROADM-C1"
131 .format(RESTCONF_BASE_URL))
132 data = test_utils.generate_connect_data("ROADM-C1", test_utils.sims['roadmc']['port'])
133 response = test_utils.put_request(url, data, 'admin', 'admin')
134 self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201) # pylint: disable=no-member
137 def test_05_connect_xprda_n1_to_roadma_pp1(self):
138 url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(RESTCONF_BASE_URL)
139 data = test_utils.generate_link_data("XPDR-A1", "1", "1", "ROADM-A1", "1", "SRG1-PP1-TXRX")
140 response = test_utils.post_request(url, data, 'admin', 'admin')
141 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
142 res = response.json()
143 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
144 CREATED_SUCCESSFULLY)
147 def test_06_connect_roadma_pp1_to_xpdra_n1(self):
148 url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(RESTCONF_BASE_URL)
149 data = test_utils.generate_link_data("XPDR-A1", "1", "1", "ROADM-A1", "1", "SRG1-PP1-TXRX")
150 response = test_utils.post_request(url, data, 'admin', 'admin')
151 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
152 res = response.json()
153 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
154 CREATED_SUCCESSFULLY)
157 def test_07_connect_xprdc_n1_to_roadmc_pp1(self):
158 url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(RESTCONF_BASE_URL)
159 data = test_utils.generate_link_data("XPDR-C1", "1", "1", "ROADM-C1", "1", "SRG1-PP1-TXRX")
160 response = test_utils.post_request(url, data, 'admin', 'admin')
161 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
162 res = response.json()
163 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
164 CREATED_SUCCESSFULLY)
167 def test_08_connect_roadmc_pp1_to_xpdrc_n1(self):
168 url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(RESTCONF_BASE_URL)
169 data = test_utils.generate_link_data("XPDR-C1", "1", "1", "ROADM-C1", "1", "SRG1-PP1-TXRX")
170 response = test_utils.post_request(url, data, 'admin', 'admin')
171 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
172 res = response.json()
173 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
174 CREATED_SUCCESSFULLY)
177 def test_09_connect_xprda_n2_to_roadma_pp2(self):
178 url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(RESTCONF_BASE_URL)
179 data = test_utils.generate_link_data("XPDR-A1", "1", "2", "ROADM-A1", "1", "SRG1-PP2-TXRX")
180 response = test_utils.post_request(url, data, 'admin', 'admin')
181 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
182 res = response.json()
183 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
184 CREATED_SUCCESSFULLY)
187 def test_10_connect_roadma_pp2_to_xpdra_n2(self):
188 url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(RESTCONF_BASE_URL)
189 data = test_utils.generate_link_data("XPDR-A1", "1", "2", "ROADM-A1", "1", "SRG1-PP2-TXRX")
190 response = test_utils.post_request(url, data, 'admin', 'admin')
191 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
192 res = response.json()
193 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
194 CREATED_SUCCESSFULLY)
197 def test_11_connect_xprdc_n2_to_roadmc_pp2(self):
198 url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(RESTCONF_BASE_URL)
199 data = test_utils.generate_link_data("XPDR-C1", "1", "2", "ROADM-C1", "1", "SRG1-PP2-TXRX")
200 response = test_utils.post_request(url, data, 'admin', 'admin')
201 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
202 res = response.json()
203 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
204 CREATED_SUCCESSFULLY)
207 def test_12_connect_roadmc_pp2_to_xpdrc_n2(self):
208 url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(RESTCONF_BASE_URL)
209 data = test_utils.generate_link_data("XPDR-C1", "1", "2", "ROADM-C1", "1", "SRG1-PP2-TXRX")
210 response = test_utils.post_request(url, data, 'admin', 'admin')
211 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
212 res = response.json()
213 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
214 CREATED_SUCCESSFULLY)
217 def test_13_get_tapi_openroadm_topology(self):
218 url = "{}/operations/tapi-topology:get-topology-details".format(RESTCONF_BASE_URL)
220 "tapi-topology:input": {
221 "tapi-topology:topology-id-or-name": "openroadm-topology"
225 response = test_utils.post_request(url, data, 'admin', 'admin')
226 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
227 res = response.json()
228 self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'There should be 1 node')
229 self.assertEqual(len(res["output"]["topology"]["node"][0]["owned-node-edge-point"]), 4,
230 'There should be 4 owned-node-edge-points')
232 def test_14_get_tapi_otn_topology(self):
233 url = "{}/operations/tapi-topology:get-topology-details".format(RESTCONF_BASE_URL)
235 "tapi-topology:input": {
236 "tapi-topology:topology-id-or-name": "otn-topology"
240 response = test_utils.post_request(url, data, 'admin', 'admin')
241 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
242 res = response.json()
243 self.assertEqual(len(res["output"]["topology"]["node"]), 4, 'There should be 4 nodes')
244 self.assertEqual(len(res["output"]["topology"]["link"]), 5, 'There should be 5 links')
245 link_to_check = res["output"]["topology"]["link"][0]
246 # get info from first link to do deeper check
247 node1_uid = link_to_check["node-edge-point"][0]["node-uuid"]
248 node2_uid = link_to_check["node-edge-point"][1]["node-uuid"]
249 node_edge_point1_uid = link_to_check["node-edge-point"][0]["node-edge-point-uuid"]
250 node_edge_point2_uid = link_to_check["node-edge-point"][1]["node-edge-point-uuid"]
251 # get node associated to link info
252 nodes = res["output"]["topology"]["node"]
253 node1 = find_object_with_key(nodes, "uuid", node1_uid)
254 self.assertIsNotNone(node1, 'Node with uuid ' + node1_uid + ' should not be null')
255 node2 = find_object_with_key(nodes, "uuid", node2_uid)
256 self.assertIsNotNone(node2, 'Node with uuid ' + node2_uid + ' should not be null')
257 # get edge-point associated to nodes
258 node1_edge_point = node1["owned-node-edge-point"]
259 node2_edge_point = node2["owned-node-edge-point"]
260 node_edge_point1 = find_object_with_key(node1_edge_point, "uuid", node_edge_point1_uid)
261 self.assertIsNotNone(node_edge_point1, 'Node edge point with uuid ' + node_edge_point1_uid + 'should not be '
263 node_edge_point2 = find_object_with_key(node2_edge_point, "uuid", node_edge_point2_uid)
264 self.assertIsNotNone(node_edge_point2, 'Node edge point with uuid ' + node_edge_point2_uid + 'should not be '
266 self.assertEqual(len(node_edge_point1["name"]), 1, 'There should be 1 name')
267 self.assertEqual(len(node_edge_point2["name"]), 1, 'There should be 1 name')
268 if node_edge_point1["layer-protocol-name"] == 'ODU':
269 self.assertIn('NodeEdgePoint_N', node_edge_point1["name"][0]["value-name"], 'Value name should be '
271 elif node_edge_point1["layer-protocol-name"] == 'PHOTONIC_MEDIA':
272 self.assertIn('iNodeEdgePoint_', node_edge_point1["name"][0]["value-name"], 'Value name should be '
275 self.fail('Wrong layer protocol name')
277 if node_edge_point2["layer-protocol-name"] == 'ODU':
278 self.assertIn('NodeEdgePoint_N', node_edge_point2["name"][0]["value-name"], 'Value name should be '
280 elif node_edge_point2["layer-protocol-name"] == 'PHOTONIC_MEDIA':
281 self.assertIn('iNodeEdgePoint_', node_edge_point2["name"][0]["value-name"], 'Value name should be '
284 self.fail('Wrong layer protocol name')
286 def test_15_disconnect_xpdra(self):
287 url = ("{}/config/network-topology:"
288 "network-topology/topology/topology-netconf/node/XPDR-A1"
289 .format(RESTCONF_BASE_URL))
291 response = test_utils.delete_request(url, 'admin', 'admin')
292 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
295 def test_16_disconnect_xpdrc(self):
296 url = ("{}/config/network-topology:"
297 "network-topology/topology/topology-netconf/node/XPDR-C1"
298 .format(RESTCONF_BASE_URL))
300 response = test_utils.delete_request(url, 'admin', 'admin')
301 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
304 def test_17_disconnect_roadma(self):
305 url = ("{}/config/network-topology:"
306 "network-topology/topology/topology-netconf/node/ROADM-A1"
307 .format(RESTCONF_BASE_URL))
309 response = test_utils.delete_request(url, 'admin', 'admin')
310 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
313 def test_18_disconnect_roadmc(self):
314 url = ("{}/config/network-topology:"
315 "network-topology/topology/topology-netconf/node/ROADM-C1"
316 .format(RESTCONF_BASE_URL))
318 response = test_utils.delete_request(url, 'admin', 'admin')
319 self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member
322 def test_19_disconnect_spdr_sa1(self):
323 url = ("{}/config/network-topology:"
324 "network-topology/topology/topology-netconf/node/SPDR-SA1"
325 .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 find_object_with_key(list_dicts, key, value):
331 for dict_ in list_dicts:
332 if dict_[key] == value:
337 if __name__ == "__main__":
338 unittest.main(verbosity=2)