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
18 # pylint: disable=wrong-import-order
23 sys.path.append('transportpce_tests/common/')
24 # pylint: disable=wrong-import-position
25 # pylint: disable=import-error
26 import test_utils # nopep8
29 CREATED_SUCCESSFULLY = 'Result message should contain Xponder Roadm Link created successfully'
32 class TransportTapitesting(unittest.TestCase):
36 NODE_VERSION = '2.2.1'
37 cr_serv_sample_data = {"input": {
38 "sdnc-request-header": {
39 "request-id": "request-1",
40 "rpc-action": "service-create",
41 "request-system-id": "appname"
43 "service-name": "service1-OCH-OTU4",
44 "common-id": "commonId",
45 "connection-type": "infrastructure",
47 "service-rate": "100",
48 "node-id": "SPDR-SA1",
49 "service-format": "OTU",
50 "otu-service-rate": "org-openroadm-otn-common-types:OTU4",
54 "port-device-name": "SPDR-SA1-XPDR1",
56 "port-name": "XPDR1-NETWORK1",
57 "port-rack": "000000.00",
58 "port-shelf": "Chassis#1"
61 "lgx-device-name": "Some lgx-device-name",
62 "lgx-port-name": "Some lgx-port-name",
63 "lgx-port-rack": "000000.00",
64 "lgx-port-shelf": "00"
70 "port-device-name": "SPDR-SA1-XPDR1",
72 "port-name": "XPDR1-NETWORK1",
73 "port-rack": "000000.00",
74 "port-shelf": "Chassis#1"
77 "lgx-device-name": "Some lgx-device-name",
78 "lgx-port-name": "Some lgx-port-name",
79 "lgx-port-rack": "000000.00",
80 "lgx-port-shelf": "00"
87 "service-rate": "100",
88 "node-id": "SPDR-SC1",
89 "service-format": "OTU",
90 "otu-service-rate": "org-openroadm-otn-common-types:OTU4",
94 "port-device-name": "SPDR-SC1-XPDR1",
96 "port-name": "XPDR1-NETWORK1",
97 "port-rack": "000000.00",
98 "port-shelf": "Chassis#1"
101 "lgx-device-name": "Some lgx-device-name",
102 "lgx-port-name": "Some lgx-port-name",
103 "lgx-port-rack": "000000.00",
104 "lgx-port-shelf": "00"
110 "port-device-name": "SPDR-SC1-XPDR1",
111 "port-type": "fixed",
112 "port-name": "XPDR1-NETWORK1",
113 "port-rack": "000000.00",
114 "port-shelf": "Chassis#1"
117 "lgx-device-name": "Some lgx-device-name",
118 "lgx-port-name": "Some lgx-port-name",
119 "lgx-port-rack": "000000.00",
120 "lgx-port-shelf": "00"
126 "due-date": "2018-06-15T00:00:01Z",
127 "operator-contact": "pw1234"
133 cls.init_failed = False
134 os.environ['JAVA_MIN_MEM'] = '1024M'
135 os.environ['JAVA_MAX_MEM'] = '4096M'
136 cls.processes = test_utils.start_tpce()
137 # TAPI feature is not installed by default in Karaf
138 if "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
139 print("installing tapi feature...")
140 result = test_utils.install_karaf_feature("odl-transportpce-tapi")
141 if result.returncode != 0:
142 cls.init_failed = True
143 print("Restarting OpenDaylight...")
144 test_utils.shutdown_process(cls.processes[0])
145 cls.processes[0] = test_utils.start_karaf()
146 test_utils.process_list[0] = cls.processes[0]
147 cls.init_failed = not test_utils.wait_until_log_contains(
148 test_utils.KARAF_LOG, test_utils.KARAF_OK_START_MSG, time_to_wait=60)
150 print("tapi installation feature failed...")
151 test_utils.shutdown_process(cls.processes[0])
153 cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION),
154 ('roadma', cls.NODE_VERSION),
155 ('roadmb', cls.NODE_VERSION),
156 ('roadmc', cls.NODE_VERSION),
157 ('xpdrc', cls.NODE_VERSION),
158 ('spdra', cls.NODE_VERSION),
159 ('spdrc', cls.NODE_VERSION)])
162 def tearDownClass(cls):
163 # pylint: disable=not-an-iterable
164 for process in cls.processes:
165 test_utils.shutdown_process(process)
166 print("all processes killed")
168 def setUp(self): # instruction executed before each test method
170 self.fail('Feature installation failed')
171 # pylint: disable=consider-using-f-string
172 print("execution of {}".format(self.id().split(".")[-1]))
174 def test_01_get_tapi_topology_T100G(self):
175 url = "{}/operations/tapi-topology:get-topology-details"
177 "tapi-topology:input": {
178 "tapi-topology:topology-id-or-name": "Transponder 100GE"
181 response = test_utils.post_request(url, data)
182 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
183 res = response.json()
184 self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'Topology should contain 1 node')
185 self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
186 self.assertNotIn("owned-node-edge-point", res["output"]["topology"]["node"][0],
187 'Node should contain no owned-node-edge-points')
188 self.assertEqual("Tpdr100g over WDM node", res["output"]["topology"]["node"][0]["name"][0]["value"],
189 'node name should be: Tpdr100g over WDM node')
190 self.assertIn("ETH", res["output"]["topology"]["node"][0]["layer-protocol-name"],
191 'Node layer protocol should contain ETH')
192 self.assertEqual(1, len(res["output"]["topology"]["node"][0]["node-rule-group"]),
193 'node should contain 1 node rule group')
195 def test_02_get_tapi_topology_T0(self):
196 url = "{}/operations/tapi-topology:get-topology-details"
198 "tapi-topology:input": {
199 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
202 response = test_utils.post_request(url, data)
203 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
204 res = response.json()
205 self.assertNotIn("node", res["output"]["topology"], 'Topology should contain no node')
206 self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
208 def test_03_connect_rdmb(self):
209 response = test_utils.mount_device("ROADM-B1", ('roadmb', self.NODE_VERSION))
210 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
213 def test_04_check_tapi_topos(self):
214 url = "{}/operations/tapi-topology:get-topology-details"
216 "tapi-topology:input": {
217 "tapi-topology:topology-id-or-name": "Transponder 100GE"
220 response = test_utils.post_request(url, data)
221 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
222 res = response.json()
223 self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'Topology should contain 1 node')
224 self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
226 url = "{}/operations/tapi-topology:get-topology-details"
228 "tapi-topology:input": {
229 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
232 response = test_utils.post_request(url, data)
233 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
234 res = response.json()
235 self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'Topology should contain 1 node')
236 self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
238 def test_05_disconnect_roadmb(self):
239 response = test_utils.unmount_device("ROADM-B1")
240 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
243 def test_06_connect_xpdra(self):
244 response = test_utils.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION))
245 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
248 def test_07_check_tapi_topos(self):
249 url = "{}/operations/tapi-topology:get-topology-details"
251 "tapi-topology:input": {
252 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
255 response = test_utils.post_request(url, data)
256 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
257 res = response.json()
258 self.assertNotIn("node", res["output"]["topology"], 'Topology should contain no node')
259 self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
261 def test_08_connect_rdma(self):
262 response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION))
263 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
266 def test_09_connect_rdmc(self):
267 response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION))
268 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
271 def test_10_check_tapi_topos(self):
272 self.test_01_get_tapi_topology_T100G()
274 url = "{}/operations/tapi-topology:get-topology-details"
276 "tapi-topology:input": {
277 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
280 response = test_utils.post_request(url, data)
281 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
282 res = response.json()
283 self.assertEqual(1, len(res["output"]["topology"]["node"]), 'Topology should contain 1 node')
284 self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
285 self.assertEqual("ROADM-infra", res["output"]["topology"]["node"][0]["name"][0]["value"],
286 'node name should be: ROADM-infra')
287 self.assertIn("PHOTONIC_MEDIA", res["output"]["topology"]["node"][0]["layer-protocol-name"],
288 'Node layer protocol should contain PHOTONIC_MEDIA')
289 self.assertEqual(1, len(res["output"]["topology"]["node"][0]["node-rule-group"]),
290 'node should contain 1 node rule group')
292 def test_11_connect_xprda_n1_to_roadma_pp1(self):
293 response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "1",
294 "ROADM-A1", "1", "SRG1-PP1-TXRX")
295 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
296 res = response.json()
297 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
298 CREATED_SUCCESSFULLY)
301 def test_12_connect_roadma_pp1_to_xpdra_n1(self):
302 response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "1",
303 "ROADM-A1", "1", "SRG1-PP1-TXRX")
304 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
305 res = response.json()
306 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
307 CREATED_SUCCESSFULLY)
310 def test_13_check_tapi_topology_T100G(self):
311 url = "{}/operations/tapi-topology:get-topology-details"
313 "tapi-topology:input": {
314 "tapi-topology:topology-id-or-name": "Transponder 100GE"
317 response = test_utils.post_request(url, data)
318 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
319 res = response.json()
320 self.assertEqual(1, len(res["output"]["topology"]["node"][0]["owned-node-edge-point"]),
321 'Node should contain 1 owned-node-edge-points')
322 self.assertEqual("XPDR-A1-XPDR1+DSR+XPDR1-CLIENT1",
323 res["output"]["topology"]["node"][0]["owned-node-edge-point"][0]["name"][0]["value"],
324 'name of owned-node-edge-points should be XPDR-A1-XPDR1+DSR+XPDR1-CLIENT1')
326 def test_14_check_tapi_topology_T0(self):
327 url = "{}/operations/tapi-topology:get-topology-details"
329 "tapi-topology:input": {
330 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
333 response = test_utils.post_request(url, data)
334 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
335 res = response.json()
336 nodes = res["output"]["topology"]["node"]
337 links = res["output"]["topology"]["link"]
338 self.assertEqual(3, len(nodes), 'Topology should contain 3 nodes')
339 self.assertEqual(2, len(links), 'Topology should contain 2 links')
340 self.assertEqual(2, count_object_with_double_key(nodes, "name", "value-name", "otsi node name"),
341 'Topology should contain 2 otsi nodes')
342 self.assertEqual(1, count_object_with_double_key(nodes, "name", "value-name", "dsr/odu node name"),
343 'Topology should contain 1 dsr node')
344 self.assertEqual(1, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
345 'Topology should contain 1 transitional link')
346 self.assertEqual(1, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
347 'Topology should contain 1 oms link')
349 def test_15_connect_xpdrc(self):
350 response = test_utils.mount_device("XPDR-C1", ('xpdrc', self.NODE_VERSION))
351 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
354 def test_16_connect_xprdc_n1_to_roadmc_pp1(self):
355 response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "1",
356 "ROADM-C1", "1", "SRG1-PP1-TXRX")
357 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
358 res = response.json()
359 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
360 CREATED_SUCCESSFULLY)
363 def test_17_connect_roadmc_pp1_to_xpdrc_n1(self):
364 response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "1",
365 "ROADM-C1", "1", "SRG1-PP1-TXRX")
366 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
367 res = response.json()
368 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
369 CREATED_SUCCESSFULLY)
372 def test_18_check_tapi_topology_T100G(self):
373 url = "{}/operations/tapi-topology:get-topology-details"
375 "tapi-topology:input": {
376 "tapi-topology:topology-id-or-name": "Transponder 100GE"
379 response = test_utils.post_request(url, data)
380 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
381 res = response.json()
382 self.assertEqual(2, len(res["output"]["topology"]["node"][0]["owned-node-edge-point"]),
383 'Node should contain 2 owned-node-edge-points')
384 self.assertEqual("XPDR-C1-XPDR1+DSR+XPDR1-CLIENT1",
385 res["output"]["topology"]["node"][0]["owned-node-edge-point"][0]["name"][0]["value"],
386 'name of owned-node-edge-points should be XPDR-C1-XPDR1+DSR+XPDR1-CLIENT1')
387 self.assertEqual("XPDR-A1-XPDR1+DSR+XPDR1-CLIENT1",
388 res["output"]["topology"]["node"][0]["owned-node-edge-point"][1]["name"][0]["value"],
389 'name of owned-node-edge-points should be XPDR-A1-XPDR1+DSR+XPDR1-CLIENT1')
391 def test_19_check_tapi_topology_T0(self):
392 url = "{}/operations/tapi-topology:get-topology-details"
394 "tapi-topology:input": {
395 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
398 response = test_utils.post_request(url, data)
399 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
400 res = response.json()
401 nodes = res["output"]["topology"]["node"]
402 links = res["output"]["topology"]["link"]
403 self.assertEqual(5, len(nodes), 'Topology should contain 5 nodes')
404 self.assertEqual(4, len(links), 'Topology should contain 4 links')
405 self.assertEqual(3, count_object_with_double_key(nodes, "name", "value-name", "otsi node name"),
406 'Topology should contain 3 otsi nodes')
407 self.assertEqual(2, count_object_with_double_key(nodes, "name", "value-name", "dsr/odu node name"),
408 'Topology should contain 2 dsr nodes')
409 self.assertEqual(2, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
410 'Topology should contain 2 transitional links')
411 self.assertEqual(2, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
412 'Topology should contain 2 oms links')
414 def test_20_connect_spdr_sa1(self):
415 response = test_utils.mount_device("SPDR-SA1", ('spdra', self.NODE_VERSION))
416 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
418 # TODO replace connect and disconnect timers with test_utils.wait_until_log_contains
420 def test_21_connect_spdr_sc1(self):
421 response = test_utils.mount_device("SPDR-SC1", ('spdrc', self.NODE_VERSION))
422 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
424 # TODO replace connect and disconnect timers with test_utils.wait_until_log_contains
426 def test_22_check_tapi_topology_T100G(self):
427 self.test_18_check_tapi_topology_T100G()
429 def test_23_check_tapi_topology_T0(self):
430 self.test_19_check_tapi_topology_T0()
432 def test_24_connect_sprda_n1_to_roadma_pp2(self):
433 response = test_utils.connect_xpdr_to_rdm_request("SPDR-SA1", "1", "1",
434 "ROADM-A1", "1", "SRG1-PP2-TXRX")
435 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
436 res = response.json()
437 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
438 CREATED_SUCCESSFULLY)
441 def test_25_connect_roadma_pp2_to_spdra_n1(self):
442 response = test_utils.connect_rdm_to_xpdr_request("SPDR-SA1", "1", "1",
443 "ROADM-A1", "1", "SRG1-PP2-TXRX")
444 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
445 res = response.json()
446 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
447 CREATED_SUCCESSFULLY)
450 def test_26_connect_sprdc_n1_to_roadmc_pp2(self):
451 response = test_utils.connect_xpdr_to_rdm_request("SPDR-SC1", "1", "1",
452 "ROADM-C1", "1", "SRG1-PP2-TXRX")
453 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
454 res = response.json()
455 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
456 CREATED_SUCCESSFULLY)
459 def test_27_connect_roadmc_pp2_to_spdrc_n1(self):
460 response = test_utils.connect_rdm_to_xpdr_request("SPDR-SC1", "1", "1",
461 "ROADM-C1", "1", "SRG1-PP2-TXRX")
462 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
463 res = response.json()
464 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
465 CREATED_SUCCESSFULLY)
468 def test_28_check_tapi_topology_T100G(self):
469 self.test_18_check_tapi_topology_T100G()
471 def test_29_check_tapi_topology_T0(self):
472 url = "{}/operations/tapi-topology:get-topology-details"
474 "tapi-topology:input": {
475 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
478 response = test_utils.post_request(url, data)
479 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
480 res = response.json()
481 nodes = res["output"]["topology"]["node"]
482 links = res["output"]["topology"]["link"]
483 self.assertEqual(9, len(nodes), 'Topology should contain 9 nodes')
484 self.assertEqual(8, len(links), 'Topology should contain 8 links')
485 self.assertEqual(5, count_object_with_double_key(nodes, "name", "value-name", "otsi node name"),
486 'Topology should contain 5 otsi nodes')
487 self.assertEqual(4, count_object_with_double_key(nodes, "name", "value-name", "dsr/odu node name"),
488 'Topology should contain 4 dsr nodes')
489 self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
490 'Topology should contain 4 transitional links')
491 self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
492 'Topology should contain 4 oms links')
494 def test_30_add_oms_attributes(self):
495 # Config ROADMA-ROADMC oms-attributes
497 "auto-spanloss": "true",
498 "spanloss-base": 11.4,
499 "spanloss-current": 12,
500 "engineered-spanloss": 12.2,
501 "link-concatenation": [{
504 "SRLG-length": 100000,
506 response = test_utils.add_oms_attr_request("ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", data)
507 self.assertEqual(response.status_code, requests.codes.created)
508 # Config ROADMC-ROADMA oms-attributes
510 "auto-spanloss": "true",
511 "spanloss-base": 11.4,
512 "spanloss-current": 12,
513 "engineered-spanloss": 12.2,
514 "link-concatenation": [{
517 "SRLG-length": 100000,
519 response = test_utils.add_oms_attr_request("ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX", data)
520 self.assertEqual(response.status_code, requests.codes.created)
522 def test_31_create_OCH_OTU4_service(self):
523 response = test_utils.service_create_request(self.cr_serv_sample_data)
524 self.assertEqual(response.status_code, requests.codes.ok)
525 res = response.json()
526 self.assertIn('PCE calculation in progress',
527 res['output']['configuration-response-common']['response-message'])
528 time.sleep(self.WAITING)
530 def test_32_check_tapi_topology_T0(self):
531 url = "{}/operations/tapi-topology:get-topology-details"
533 "tapi-topology:input": {
534 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
537 response = test_utils.post_request(url, data)
538 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
539 res = response.json()
540 nodes = res["output"]["topology"]["node"]
541 links = res["output"]["topology"]["link"]
542 self.assertEqual(9, len(nodes), 'Topology should contain 9 nodes')
543 self.assertEqual(9, len(links), 'Topology should contain 9 links')
544 self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
545 'Topology should contain 4 transitional links')
546 self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
547 'Topology should contain 4 oms links')
548 self.assertEqual(1, count_object_with_double_key(links, "name", "value-name", "otn link name"),
549 'Topology should contain 1 otn link')
551 if link["name"][0]["value"] == "OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1":
552 self.assertEqual(100000, link["available-capacity"]["total-size"]["value"],
553 'OTU4 link should have an available capacity of 100 000 Mbps')
554 elif link["name"][0]["value-name"] == "transitional link name":
555 self.assertEqual(100, link["available-capacity"]["total-size"]["value"],
556 'link should have an available capacity of 100 Gbps')
557 self.assertEqual(2, len(link["node-edge-point"]), 'link should have 2 neps')
559 def test_33_create_ODU4_service(self):
560 self.cr_serv_sample_data["input"]["service-name"] = "service1-ODU4"
561 self.cr_serv_sample_data["input"]["service-a-end"]["service-format"] = "ODU"
562 del self.cr_serv_sample_data["input"]["service-a-end"]["otu-service-rate"]
563 self.cr_serv_sample_data["input"]["service-a-end"]["odu-service-rate"] = "org-openroadm-otn-common-types:ODU4"
564 self.cr_serv_sample_data["input"]["service-z-end"]["service-format"] = "ODU"
565 del self.cr_serv_sample_data["input"]["service-z-end"]["otu-service-rate"]
566 self.cr_serv_sample_data["input"]["service-z-end"]["odu-service-rate"] = "org-openroadm-otn-common-types:ODU4"
568 response = test_utils.service_create_request(self.cr_serv_sample_data)
569 self.assertEqual(response.status_code, requests.codes.ok)
570 res = response.json()
571 self.assertIn('PCE calculation in progress',
572 res['output']['configuration-response-common']['response-message'])
573 time.sleep(self.WAITING)
575 def test_34_check_tapi_topology_T0(self):
576 url = "{}/operations/tapi-topology:get-topology-details"
578 "tapi-topology:input": {
579 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
582 response = test_utils.post_request(url, data)
583 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
584 res = response.json()
585 nodes = res["output"]["topology"]["node"]
586 links = res["output"]["topology"]["link"]
587 self.assertEqual(9, len(nodes), 'Topology should contain 9 nodes')
588 self.assertEqual(10, len(links), 'Topology should contain 10 links')
589 self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
590 'Topology should contain 4 transitional links')
591 self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
592 'Topology should contain 4 oms links')
593 self.assertEqual(2, count_object_with_double_key(links, "name", "value-name", "otn link name"),
594 'Topology should contain 2 otn links')
596 if link["name"][0]["value"] == "OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1":
597 self.assertEqual(0, link["available-capacity"]["total-size"]["value"],
598 'OTU4 link should have an available capacity of 0 Mbps')
599 elif link["name"][0]["value"] == "ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1":
600 self.assertEqual(100000, link["available-capacity"]["total-size"]["value"],
601 'ODU4 link should have an available capacity of 100 000 Mbps')
602 elif link["name"][0]["value-name"] == "transitional link name":
603 self.assertEqual(100, link["available-capacity"]["total-size"]["value"],
604 'link should have an available capacity of 100 Gbps')
605 self.assertEqual(2, len(link["node-edge-point"]), 'link should have 2 neps')
607 def test_35_connect_sprda_2_n2_to_roadma_pp3(self):
608 response = test_utils.connect_xpdr_to_rdm_request("SPDR-SA1", "2", "2",
609 "ROADM-A1", "1", "SRG1-PP3-TXRX")
610 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
611 res = response.json()
612 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
613 CREATED_SUCCESSFULLY)
616 def test_36_connect_roadma_pp3_to_spdra_2_n2(self):
617 response = test_utils.connect_rdm_to_xpdr_request("SPDR-SA1", "2", "2",
618 "ROADM-A1", "1", "SRG1-PP3-TXRX")
619 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
620 res = response.json()
621 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
622 CREATED_SUCCESSFULLY)
625 def test_37_check_tapi_topology_T0(self):
626 url = "{}/operations/tapi-topology:get-topology-details"
628 "tapi-topology:input": {
629 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
632 response = test_utils.post_request(url, data)
633 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
634 res = response.json()
635 nodes = res["output"]["topology"]["node"]
636 links = res["output"]["topology"]["link"]
637 self.assertEqual(11, len(nodes), 'Topology should contain 11 nodes')
638 self.assertEqual(12, len(links), 'Topology should contain 12 links')
639 self.assertEqual(6, count_object_with_double_key(nodes, "name", "value-name", "otsi node name"),
640 'Topology should contain 6 otsi nodes')
641 self.assertEqual(5, count_object_with_double_key(nodes, "name", "value-name", "dsr/odu node name"),
642 'Topology should contain 5 dsr nodes')
643 self.assertEqual(5, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
644 'Topology should contain 5 transitional links')
645 self.assertEqual(5, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
646 'Topology should contain 5 oms links')
647 self.assertEqual(2, count_object_with_double_key(links, "name", "value-name", "otn link name"),
648 'Topology should contain 2 otn links')
650 def test_38_delete_ODU4_service(self):
651 response = test_utils.service_delete_request("service1-ODU4")
652 self.assertEqual(response.status_code, requests.codes.ok)
653 res = response.json()
654 self.assertIn('Renderer service delete in progress',
655 res['output']['configuration-response-common']['response-message'])
658 def test_39_delete_OCH_OTU4_service(self):
659 response = test_utils.service_delete_request("service1-OCH-OTU4")
660 self.assertEqual(response.status_code, requests.codes.ok)
661 res = response.json()
662 self.assertIn('Renderer service delete in progress',
663 res['output']['configuration-response-common']['response-message'])
666 def test_40_check_tapi_topology_T0(self):
667 url = "{}/operations/tapi-topology:get-topology-details"
669 "tapi-topology:input": {
670 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
673 response = test_utils.post_request(url, data)
674 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
675 res = response.json()
676 nodes = res["output"]["topology"]["node"]
677 links = res["output"]["topology"]["link"]
678 self.assertEqual(11, len(nodes), 'Topology should contain 11 nodes')
679 self.assertEqual(10, len(links), 'Topology should contain 10 links')
680 self.assertEqual(0, count_object_with_double_key(links, "name", "value-name", "otn link name"),
681 'Topology should contain 0 otn link')
683 def test_41_disconnect_xponders_from_roadm(self):
684 url = "{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:link/"
685 response = test_utils.get_ordm_topo_request("")
686 self.assertEqual(response.status_code, requests.codes.ok)
687 res = response.json()
688 links = res['network'][0]['ietf-network-topology:link']
690 if link["org-openroadm-common-network:link-type"] in ('XPONDER-OUTPUT', 'XPONDER-INPUT'):
691 link_name = link["link-id"]
692 response = test_utils.delete_request(url+link_name)
693 self.assertEqual(response.status_code, requests.codes.ok)
695 def test_42_check_tapi_topology_T0(self):
696 url = "{}/operations/tapi-topology:get-topology-details"
698 "tapi-topology:input": {
699 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
702 response = test_utils.post_request(url, data)
703 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
704 res = response.json()
705 nodes = res["output"]["topology"]["node"]
706 self.assertEqual(1, len(nodes), 'Topology should contain 1 node')
707 self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
708 self.assertEqual("ROADM-infra", res["output"]["topology"]["node"][0]["name"][0]["value"],
709 'node name should be: ROADM-infra')
711 def test_43_get_tapi_topology_T100G(self):
712 url = "{}/operations/tapi-topology:get-topology-details"
714 "tapi-topology:input": {
715 "tapi-topology:topology-id-or-name": "Transponder 100GE"
718 response = test_utils.post_request(url, data)
719 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
720 res = response.json()
721 self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'Topology should contain 1 node')
722 self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
723 self.assertNotIn("owned-node-edge-point", res["output"]["topology"]["node"][0],
724 'Node should contain no owned-node-edge-points')
726 def test_44_disconnect_roadma(self):
727 response = test_utils.unmount_device("ROADM-A1")
728 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
731 def test_45_disconnect_roadmc(self):
732 response = test_utils.unmount_device("ROADM-C1")
733 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
736 def test_46_check_tapi_topos(self):
737 self.test_01_get_tapi_topology_T100G()
738 self.test_02_get_tapi_topology_T0()
740 def test_47_disconnect_xpdra(self):
741 response = test_utils.unmount_device("XPDR-A1")
742 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
745 def test_48_disconnect_xpdrc(self):
746 response = test_utils.unmount_device("XPDR-C1")
747 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
750 def test_49_disconnect_spdr_sa1(self):
751 response = test_utils.unmount_device("SPDR-SA1")
752 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
755 def test_50_disconnect_spdr_sc1(self):
756 response = test_utils.unmount_device("SPDR-SC1")
757 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
760 def find_object_with_key(list_dicts, key, value):
761 for dict_ in list_dicts:
762 if dict_[key] == value:
767 def count_object_with_double_key(list_dicts, key1, key2, value):
769 for dictio in list_dicts:
770 if dictio[key1][0][key2] == value:
775 if __name__ == "__main__":
776 unittest.main(verbosity=2)