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 ##############################################################################
11 # pylint: disable=no-member
12 # pylint: disable=too-many-public-methods
13 # some pylint false positives specific to tapi test
14 # pylint: disable=unsubscriptable-object
15 # pylint: disable=unsupported-assignment-operation
24 from common import test_utils
27 CREATED_SUCCESSFULLY = 'Result message should contain Xponder Roadm Link created successfully'
30 class TransportTapitesting(unittest.TestCase):
36 cls.init_failed = False
37 os.environ['JAVA_MIN_MEM'] = '1024M'
38 os.environ['JAVA_MAX_MEM'] = '4069M'
39 cls.processes = test_utils.start_tpce()
40 # TAPI feature is not installed by default in Karaf
41 if "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
42 print("installing tapi feature...")
43 result = test_utils.install_karaf_feature("odl-transportpce-tapi")
44 if result.returncode != 0:
45 cls.init_failed = True
46 print("Restarting OpenDaylight...")
47 test_utils.shutdown_process(cls.processes[0])
48 cls.processes[0] = test_utils.start_karaf()
49 test_utils.process_list[0] = cls.processes[0]
50 cls.init_failed = not test_utils.wait_until_log_contains(
51 test_utils.KARAF_LOG, test_utils.KARAF_OK_START_MSG, time_to_wait=60)
53 print("tapi installation feature failed...")
54 test_utils.shutdown_process(cls.processes[0])
56 cls.processes = test_utils.start_sims(['xpdra', 'roadma', 'roadmc', 'xpdrc', 'spdra'])
59 def tearDownClass(cls):
60 # pylint: disable=not-an-iterable
61 for process in cls.processes:
62 test_utils.shutdown_process(process)
63 print("all processes killed")
65 def setUp(self): # instruction executed before each test method
67 self.fail('Feature installation failed')
68 print("execution of {}".format(self.id().split(".")[-1]))
70 def test_00_connect_spdr_sa1(self):
71 response = test_utils.mount_device("SPDR-SA1", 'spdra')
72 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
74 # TODO replace connect and disconnect timers with test_utils.wait_until_log_contains
76 def test_01_connect_xpdra(self):
77 response = test_utils.mount_device("XPDR-A1", 'xpdra')
78 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
81 def test_02_connect_xpdrc(self):
82 response = test_utils.mount_device("XPDR-C1", 'xpdrc')
83 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
86 def test_03_connect_rdma(self):
87 response = test_utils.mount_device("ROADM-A1", 'roadma')
88 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
91 def test_04_connect_rdmc(self):
92 response = test_utils.mount_device("ROADM-C1", 'roadmc')
93 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
96 def test_05_connect_xprda_n1_to_roadma_pp1(self):
97 response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "1",
98 "ROADM-A1", "1", "SRG1-PP1-TXRX")
99 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
100 res = response.json()
101 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
102 CREATED_SUCCESSFULLY)
105 def test_06_connect_roadma_pp1_to_xpdra_n1(self):
106 response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "1",
107 "ROADM-A1", "1", "SRG1-PP1-TXRX")
108 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
109 res = response.json()
110 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
111 CREATED_SUCCESSFULLY)
114 def test_07_connect_xprdc_n1_to_roadmc_pp1(self):
115 response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "1",
116 "ROADM-C1", "1", "SRG1-PP1-TXRX")
117 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
118 res = response.json()
119 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
120 CREATED_SUCCESSFULLY)
123 def test_08_connect_roadmc_pp1_to_xpdrc_n1(self):
124 response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "1",
125 "ROADM-C1", "1", "SRG1-PP1-TXRX")
126 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
127 res = response.json()
128 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
129 CREATED_SUCCESSFULLY)
132 def test_09_connect_xprda_n2_to_roadma_pp2(self):
133 response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "2",
134 "ROADM-A1", "1", "SRG1-PP2-TXRX")
135 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
136 res = response.json()
137 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
138 CREATED_SUCCESSFULLY)
141 def test_10_connect_roadma_pp2_to_xpdra_n2(self):
142 response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "2",
143 "ROADM-A1", "1", "SRG1-PP2-TXRX")
144 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
145 res = response.json()
146 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
147 CREATED_SUCCESSFULLY)
150 def test_11_connect_xprdc_n2_to_roadmc_pp2(self):
151 response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "2",
152 "ROADM-C1", "1", "SRG1-PP2-TXRX")
153 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
154 res = response.json()
155 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
156 CREATED_SUCCESSFULLY)
159 def test_12_connect_roadmc_pp2_to_xpdrc_n2(self):
160 response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "2",
161 "ROADM-C1", "1", "SRG1-PP2-TXRX")
162 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
163 res = response.json()
164 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
165 CREATED_SUCCESSFULLY)
168 def test_13_get_tapi_openroadm_topology(self):
169 url = "{}/operations/tapi-topology:get-topology-details"
171 "tapi-topology:input": {
172 "tapi-topology:topology-id-or-name": "openroadm-topology"
176 response = test_utils.post_request(url, data)
177 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
178 res = response.json()
179 self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'There should be 1 node')
180 self.assertEqual(len(res["output"]["topology"]["node"][0]["owned-node-edge-point"]), 4,
181 'There should be 4 owned-node-edge-points')
183 def test_14_get_tapi_otn_topology(self):
184 url = "{}/operations/tapi-topology:get-topology-details"
186 "tapi-topology:input": {
187 "tapi-topology:topology-id-or-name": "otn-topology"
191 response = test_utils.post_request(url, data)
192 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
193 res = response.json()
194 self.assertEqual(len(res["output"]["topology"]["node"]), 4, 'There should be 4 nodes')
195 self.assertEqual(len(res["output"]["topology"]["link"]), 5, 'There should be 5 links')
196 link_to_check = res["output"]["topology"]["link"][0]
197 # get info from first link to do deeper check
198 node1_uid = link_to_check["node-edge-point"][0]["node-uuid"]
199 node2_uid = link_to_check["node-edge-point"][1]["node-uuid"]
200 node_edge_point1_uid = link_to_check["node-edge-point"][0]["node-edge-point-uuid"]
201 node_edge_point2_uid = link_to_check["node-edge-point"][1]["node-edge-point-uuid"]
202 # get node associated to link info
203 nodes = res["output"]["topology"]["node"]
204 node1 = find_object_with_key(nodes, "uuid", node1_uid)
205 self.assertIsNotNone(node1, 'Node with uuid ' + node1_uid + ' should not be null')
206 node2 = find_object_with_key(nodes, "uuid", node2_uid)
207 self.assertIsNotNone(node2, 'Node with uuid ' + node2_uid + ' should not be null')
208 # get edge-point associated to nodes
209 node1_edge_point = node1["owned-node-edge-point"]
210 node2_edge_point = node2["owned-node-edge-point"]
211 node_edge_point1 = find_object_with_key(node1_edge_point, "uuid", node_edge_point1_uid)
212 self.assertIsNotNone(node_edge_point1, 'Node edge point with uuid ' + node_edge_point1_uid + 'should not be '
214 node_edge_point2 = find_object_with_key(node2_edge_point, "uuid", node_edge_point2_uid)
215 self.assertIsNotNone(node_edge_point2, 'Node edge point with uuid ' + node_edge_point2_uid + 'should not be '
217 self.assertEqual(len(node_edge_point1["name"]), 1, 'There should be 1 name')
218 self.assertEqual(len(node_edge_point2["name"]), 1, 'There should be 1 name')
219 if node_edge_point1["layer-protocol-name"] == 'ODU':
220 self.assertIn('NodeEdgePoint_N', node_edge_point1["name"][0]["value-name"], 'Value name should be '
222 elif node_edge_point1["layer-protocol-name"] == 'PHOTONIC_MEDIA':
223 self.assertIn('iNodeEdgePoint_', node_edge_point1["name"][0]["value-name"], 'Value name should be '
226 self.fail('Wrong layer protocol name')
228 if node_edge_point2["layer-protocol-name"] == 'ODU':
229 self.assertIn('NodeEdgePoint_N', node_edge_point2["name"][0]["value-name"], 'Value name should be '
231 elif node_edge_point2["layer-protocol-name"] == 'PHOTONIC_MEDIA':
232 self.assertIn('iNodeEdgePoint_', node_edge_point2["name"][0]["value-name"], 'Value name should be '
235 self.fail('Wrong layer protocol name')
237 def test_15_disconnect_xpdra(self):
238 response = test_utils.unmount_device("XPDR-A1")
239 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
242 def test_16_disconnect_xpdrc(self):
243 response = test_utils.unmount_device("XPDR-C1")
244 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
247 def test_17_disconnect_roadma(self):
248 response = test_utils.unmount_device("ROADM-A1")
249 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
252 def test_18_disconnect_roadmc(self):
253 response = test_utils.unmount_device("ROADM-C1")
254 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
257 def test_19_disconnect_spdr_sa1(self):
258 response = test_utils.unmount_device("SPDR-SA1")
259 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
262 def find_object_with_key(list_dicts, key, value):
263 for dict_ in list_dicts:
264 if dict_[key] == value:
269 if __name__ == "__main__":
270 unittest.main(verbosity=2)