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 ##############################################################################
17 from common import test_utils
20 CREATED_SUCCESSFULLY = 'Result message should contain Xponder Roadm Link created successfully'
23 class TransportTapitesting(unittest.TestCase):
29 cls.init_failed = False
31 cls.processes = test_utils.start_tpce()
32 # TAPI feature is not installed by default in Karaf
33 if "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
34 print("installing tapi feature...")
35 result = test_utils.install_karaf_feature("odl-transportpce-tapi")
36 if result.returncode != 0:
37 cls.init_failed = True
38 print("Restarting OpenDaylight...")
39 test_utils.shutdown_process(cls.processes[0])
40 cls.processes = test_utils.start_tpce()
41 cls.init_failed = not test_utils.wait_until_log_contains(
42 test_utils.KARAF_LOG, test_utils.KARAF_OK_START_MSG, time_to_wait=60)
44 print("tapi installaiton feature failed...")
45 test_utils.shutdown_process(cls.odl_processes[0])
47 cls.processes = test_utils.start_sims(['xpdra', 'roadma', 'roadmc', 'xpdrc', 'spdrav2'])
50 def tearDownClass(cls):
51 for process in cls.processes:
52 test_utils.shutdown_process(process)
53 print("all processes killed")
55 def setUp(self): # instruction executed before each test method
57 self.fail('Feature installation failed')
58 print("execution of {}".format(self.id().split(".")[-1]))
60 def test_00_connect_spdr_sa1(self):
61 response = test_utils.mount_device("SPDR-SA1", 'spdrav2')
62 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
64 # TODO replace connect and disconnect timers with test_utils.wait_until_log_contains
66 def test_01_connect_xpdra(self):
67 response = test_utils.mount_device("XPDR-A1", 'xpdra')
68 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
71 def test_02_connect_xpdrc(self):
72 response = test_utils.mount_device("XPDR-C1", 'xpdrc')
73 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
76 def test_03_connect_rdma(self):
77 response = test_utils.mount_device("ROADM-A1", 'roadma')
78 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
81 def test_04_connect_rdmc(self):
82 response = test_utils.mount_device("ROADM-C1", 'roadmc')
83 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
86 def test_05_connect_xprda_n1_to_roadma_pp1(self):
87 url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL)
88 data = test_utils.generate_link_data("XPDR-A1", "1", "1", "ROADM-A1", "1", "SRG1-PP1-TXRX")
89 response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
90 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
92 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
96 def test_06_connect_roadma_pp1_to_xpdra_n1(self):
97 url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL)
98 data = test_utils.generate_link_data("XPDR-A1", "1", "1", "ROADM-A1", "1", "SRG1-PP1-TXRX")
99 response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
100 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
101 res = response.json()
102 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
103 CREATED_SUCCESSFULLY)
106 def test_07_connect_xprdc_n1_to_roadmc_pp1(self):
107 url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL)
108 data = test_utils.generate_link_data("XPDR-C1", "1", "1", "ROADM-C1", "1", "SRG1-PP1-TXRX")
109 response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
110 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
111 res = response.json()
112 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
113 CREATED_SUCCESSFULLY)
116 def test_08_connect_roadmc_pp1_to_xpdrc_n1(self):
117 url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL)
118 data = test_utils.generate_link_data("XPDR-C1", "1", "1", "ROADM-C1", "1", "SRG1-PP1-TXRX")
119 response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
120 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
121 res = response.json()
122 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
123 CREATED_SUCCESSFULLY)
126 def test_09_connect_xprda_n2_to_roadma_pp2(self):
127 url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL)
128 data = test_utils.generate_link_data("XPDR-A1", "1", "2", "ROADM-A1", "1", "SRG1-PP2-TXRX")
129 response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
130 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
131 res = response.json()
132 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
133 CREATED_SUCCESSFULLY)
136 def test_10_connect_roadma_pp2_to_xpdra_n2(self):
137 url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL)
138 data = test_utils.generate_link_data("XPDR-A1", "1", "2", "ROADM-A1", "1", "SRG1-PP2-TXRX")
139 response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
140 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
141 res = response.json()
142 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
143 CREATED_SUCCESSFULLY)
146 def test_11_connect_xprdc_n2_to_roadmc_pp2(self):
147 url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL)
148 data = test_utils.generate_link_data("XPDR-C1", "1", "2", "ROADM-C1", "1", "SRG1-PP2-TXRX")
149 response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
150 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
151 res = response.json()
152 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
153 CREATED_SUCCESSFULLY)
156 def test_12_connect_roadmc_pp2_to_xpdrc_n2(self):
157 url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL)
158 data = test_utils.generate_link_data("XPDR-C1", "1", "2", "ROADM-C1", "1", "SRG1-PP2-TXRX")
159 response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
160 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
161 res = response.json()
162 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
163 CREATED_SUCCESSFULLY)
166 def test_13_get_tapi_openroadm_topology(self):
167 url = "{}/operations/tapi-topology:get-topology-details".format(test_utils.RESTCONF_BASE_URL)
169 "tapi-topology:input": {
170 "tapi-topology:topology-id-or-name": "openroadm-topology"
174 response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
175 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
176 res = response.json()
177 self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'There should be 1 node')
178 self.assertEqual(len(res["output"]["topology"]["node"][0]["owned-node-edge-point"]), 4,
179 'There should be 4 owned-node-edge-points')
181 def test_14_get_tapi_otn_topology(self):
182 url = "{}/operations/tapi-topology:get-topology-details".format(test_utils.RESTCONF_BASE_URL)
184 "tapi-topology:input": {
185 "tapi-topology:topology-id-or-name": "otn-topology"
189 response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
190 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
191 res = response.json()
192 self.assertEqual(len(res["output"]["topology"]["node"]), 4, 'There should be 4 nodes')
193 self.assertEqual(len(res["output"]["topology"]["link"]), 5, 'There should be 5 links')
194 link_to_check = res["output"]["topology"]["link"][0]
195 # get info from first link to do deeper check
196 node1_uid = link_to_check["node-edge-point"][0]["node-uuid"]
197 node2_uid = link_to_check["node-edge-point"][1]["node-uuid"]
198 node_edge_point1_uid = link_to_check["node-edge-point"][0]["node-edge-point-uuid"]
199 node_edge_point2_uid = link_to_check["node-edge-point"][1]["node-edge-point-uuid"]
200 # get node associated to link info
201 nodes = res["output"]["topology"]["node"]
202 node1 = find_object_with_key(nodes, "uuid", node1_uid)
203 self.assertIsNotNone(node1, 'Node with uuid ' + node1_uid + ' should not be null')
204 node2 = find_object_with_key(nodes, "uuid", node2_uid)
205 self.assertIsNotNone(node2, 'Node with uuid ' + node2_uid + ' should not be null')
206 # get edge-point associated to nodes
207 node1_edge_point = node1["owned-node-edge-point"]
208 node2_edge_point = node2["owned-node-edge-point"]
209 node_edge_point1 = find_object_with_key(node1_edge_point, "uuid", node_edge_point1_uid)
210 self.assertIsNotNone(node_edge_point1, 'Node edge point with uuid ' + node_edge_point1_uid + 'should not be '
212 node_edge_point2 = find_object_with_key(node2_edge_point, "uuid", node_edge_point2_uid)
213 self.assertIsNotNone(node_edge_point2, 'Node edge point with uuid ' + node_edge_point2_uid + 'should not be '
215 self.assertEqual(len(node_edge_point1["name"]), 1, 'There should be 1 name')
216 self.assertEqual(len(node_edge_point2["name"]), 1, 'There should be 1 name')
217 if node_edge_point1["layer-protocol-name"] == 'ODU':
218 self.assertIn('NodeEdgePoint_N', node_edge_point1["name"][0]["value-name"], 'Value name should be '
220 elif node_edge_point1["layer-protocol-name"] == 'PHOTONIC_MEDIA':
221 self.assertIn('iNodeEdgePoint_', node_edge_point1["name"][0]["value-name"], 'Value name should be '
224 self.fail('Wrong layer protocol name')
226 if node_edge_point2["layer-protocol-name"] == 'ODU':
227 self.assertIn('NodeEdgePoint_N', node_edge_point2["name"][0]["value-name"], 'Value name should be '
229 elif node_edge_point2["layer-protocol-name"] == 'PHOTONIC_MEDIA':
230 self.assertIn('iNodeEdgePoint_', node_edge_point2["name"][0]["value-name"], 'Value name should be '
233 self.fail('Wrong layer protocol name')
235 def test_15_disconnect_xpdra(self):
236 response = test_utils.unmount_device("XPDR-A1")
237 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
240 def test_16_disconnect_xpdrc(self):
241 response = test_utils.unmount_device("XPDR-C1")
242 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
245 def test_17_disconnect_roadma(self):
246 response = test_utils.unmount_device("ROADM-A1")
247 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
250 def test_18_disconnect_roadmc(self):
251 response = test_utils.unmount_device("ROADM-C1")
252 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
255 def test_19_disconnect_spdr_sa1(self):
256 response = test_utils.unmount_device("SPDR-SA1")
257 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
260 def find_object_with_key(list_dicts, key, value):
261 for dict_ in list_dicts:
262 if dict_[key] == value:
267 if __name__ == "__main__":
268 unittest.main(verbosity=2)