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 "committed-info-rate": "100000",
55 "committed-burst-size": "64"
60 "port-device-name": "SPDR-SA1-XPDR1",
62 "port-name": "XPDR1-NETWORK1",
63 "port-rack": "000000.00",
64 "port-shelf": "Chassis#1"
67 "lgx-device-name": "Some lgx-device-name",
68 "lgx-port-name": "Some lgx-port-name",
69 "lgx-port-rack": "000000.00",
70 "lgx-port-shelf": "00"
75 "port-device-name": "SPDR-SA1-XPDR1",
77 "port-name": "XPDR1-NETWORK1",
78 "port-rack": "000000.00",
79 "port-shelf": "Chassis#1"
82 "lgx-device-name": "Some lgx-device-name",
83 "lgx-port-name": "Some lgx-port-name",
84 "lgx-port-rack": "000000.00",
85 "lgx-port-shelf": "00"
91 "service-rate": "100",
92 "node-id": "SPDR-SC1",
93 "service-format": "OTU",
94 "otu-service-rate": "org-openroadm-otn-common-types:OTU4",
98 "committed-info-rate": "100000",
99 "committed-burst-size": "64"
104 "port-device-name": "SPDR-SC1-XPDR1",
105 "port-type": "fixed",
106 "port-name": "XPDR1-NETWORK1",
107 "port-rack": "000000.00",
108 "port-shelf": "Chassis#1"
111 "lgx-device-name": "Some lgx-device-name",
112 "lgx-port-name": "Some lgx-port-name",
113 "lgx-port-rack": "000000.00",
114 "lgx-port-shelf": "00"
119 "port-device-name": "SPDR-SC1-XPDR1",
120 "port-type": "fixed",
121 "port-name": "XPDR1-NETWORK1",
122 "port-rack": "000000.00",
123 "port-shelf": "Chassis#1"
126 "lgx-device-name": "Some lgx-device-name",
127 "lgx-port-name": "Some lgx-port-name",
128 "lgx-port-rack": "000000.00",
129 "lgx-port-shelf": "00"
134 "due-date": "2018-06-15T00:00:01Z",
135 "operator-contact": "pw1234"
141 cls.init_failed = False
142 os.environ['JAVA_MIN_MEM'] = '1024M'
143 os.environ['JAVA_MAX_MEM'] = '4096M'
144 cls.processes = test_utils.start_tpce()
145 # TAPI feature is not installed by default in Karaf
146 if "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
147 print("installing tapi feature...")
148 result = test_utils.install_karaf_feature("odl-transportpce-tapi")
149 if result.returncode != 0:
150 cls.init_failed = True
151 print("Restarting OpenDaylight...")
152 test_utils.shutdown_process(cls.processes[0])
153 cls.processes[0] = test_utils.start_karaf()
154 test_utils.process_list[0] = cls.processes[0]
155 cls.init_failed = not test_utils.wait_until_log_contains(
156 test_utils.KARAF_LOG, test_utils.KARAF_OK_START_MSG, time_to_wait=60)
158 print("tapi installation feature failed...")
159 test_utils.shutdown_process(cls.processes[0])
161 cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION),
162 ('roadma', cls.NODE_VERSION),
163 ('roadmb', cls.NODE_VERSION),
164 ('roadmc', cls.NODE_VERSION),
165 ('xpdrc', cls.NODE_VERSION),
166 ('spdra', cls.NODE_VERSION),
167 ('spdrc', cls.NODE_VERSION)])
170 def tearDownClass(cls):
171 # pylint: disable=not-an-iterable
172 for process in cls.processes:
173 test_utils.shutdown_process(process)
174 print("all processes killed")
176 def setUp(self): # instruction executed before each test method
178 self.fail('Feature installation failed')
179 # pylint: disable=consider-using-f-string
180 print("execution of {}".format(self.id().split(".")[-1]))
182 def test_01_get_tapi_topology_T100G(self):
183 url = "{}/operations/tapi-topology:get-topology-details"
185 "tapi-topology:input": {
186 "tapi-topology:topology-id-or-name": "Transponder 100GE"
189 response = test_utils.post_request(url, data)
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"]), 1, 'Topology should contain 1 node')
193 self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
194 self.assertNotIn("owned-node-edge-point", res["output"]["topology"]["node"][0],
195 'Node should contain no owned-node-edge-points')
196 self.assertEqual("Tpdr100g over WDM node", res["output"]["topology"]["node"][0]["name"][0]["value"],
197 'node name should be: Tpdr100g over WDM node')
198 self.assertIn("ETH", res["output"]["topology"]["node"][0]["layer-protocol-name"],
199 'Node layer protocol should contain ETH')
200 self.assertEqual(1, len(res["output"]["topology"]["node"][0]["node-rule-group"]),
201 'node should contain 1 node rule group')
203 def test_02_get_tapi_topology_T0(self):
204 url = "{}/operations/tapi-topology:get-topology-details"
206 "tapi-topology:input": {
207 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
210 response = test_utils.post_request(url, data)
211 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
212 res = response.json()
213 self.assertNotIn("node", res["output"]["topology"], 'Topology should contain no node')
214 self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
216 def test_03_connect_rdmb(self):
217 response = test_utils.mount_device("ROADM-B1", ('roadmb', self.NODE_VERSION))
218 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
221 def test_04_check_tapi_topos(self):
222 url = "{}/operations/tapi-topology:get-topology-details"
224 "tapi-topology:input": {
225 "tapi-topology:topology-id-or-name": "Transponder 100GE"
228 response = test_utils.post_request(url, data)
229 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
230 res = response.json()
231 self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'Topology should contain 1 node')
232 self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
234 url = "{}/operations/tapi-topology:get-topology-details"
236 "tapi-topology:input": {
237 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
240 response = test_utils.post_request(url, data)
241 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
242 res = response.json()
243 self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'Topology should contain 1 node')
244 self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
246 def test_05_disconnect_roadmb(self):
247 response = test_utils.unmount_device("ROADM-B1")
248 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
251 def test_06_connect_xpdra(self):
252 response = test_utils.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION))
253 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
256 def test_07_check_tapi_topos(self):
257 url = "{}/operations/tapi-topology:get-topology-details"
259 "tapi-topology:input": {
260 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
263 response = test_utils.post_request(url, data)
264 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
265 res = response.json()
266 self.assertNotIn("node", res["output"]["topology"], 'Topology should contain no node')
267 self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
269 def test_08_connect_rdma(self):
270 response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION))
271 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
274 def test_09_connect_rdmc(self):
275 response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION))
276 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
279 def test_10_check_tapi_topos(self):
280 self.test_01_get_tapi_topology_T100G()
282 url = "{}/operations/tapi-topology:get-topology-details"
284 "tapi-topology:input": {
285 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
288 response = test_utils.post_request(url, data)
289 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
290 res = response.json()
291 self.assertEqual(1, len(res["output"]["topology"]["node"]), 'Topology should contain 1 node')
292 self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
293 self.assertEqual("ROADM-infra", res["output"]["topology"]["node"][0]["name"][0]["value"],
294 'node name should be: ROADM-infra')
295 self.assertIn("PHOTONIC_MEDIA", res["output"]["topology"]["node"][0]["layer-protocol-name"],
296 'Node layer protocol should contain PHOTONIC_MEDIA')
297 self.assertEqual(1, len(res["output"]["topology"]["node"][0]["node-rule-group"]),
298 'node should contain 1 node rule group')
300 def test_11_connect_xprda_n1_to_roadma_pp1(self):
301 response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "1",
302 "ROADM-A1", "1", "SRG1-PP1-TXRX")
303 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
304 res = response.json()
305 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
306 CREATED_SUCCESSFULLY)
309 def test_12_connect_roadma_pp1_to_xpdra_n1(self):
310 response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "1",
311 "ROADM-A1", "1", "SRG1-PP1-TXRX")
312 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
313 res = response.json()
314 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
315 CREATED_SUCCESSFULLY)
318 def test_13_check_tapi_topology_T100G(self):
319 url = "{}/operations/tapi-topology:get-topology-details"
321 "tapi-topology:input": {
322 "tapi-topology:topology-id-or-name": "Transponder 100GE"
325 response = test_utils.post_request(url, data)
326 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
327 res = response.json()
328 self.assertEqual(1, len(res["output"]["topology"]["node"][0]["owned-node-edge-point"]),
329 'Node should contain 1 owned-node-edge-points')
330 self.assertEqual("XPDR1-CLIENT1",
331 res["output"]["topology"]["node"][0]["owned-node-edge-point"][0]["name"][0]["value"],
332 'name of owned-node-edge-points should be XPDR1-CLIENT1')
334 def test_14_check_tapi_topology_T0(self):
335 url = "{}/operations/tapi-topology:get-topology-details"
337 "tapi-topology:input": {
338 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
341 response = test_utils.post_request(url, data)
342 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
343 res = response.json()
344 nodes = res["output"]["topology"]["node"]
345 links = res["output"]["topology"]["link"]
346 self.assertEqual(3, len(nodes), 'Topology should contain 3 nodes')
347 self.assertEqual(2, len(links), 'Topology should contain 2 links')
348 self.assertEqual(2, count_object_with_double_key(nodes, "name", "value-name", "otsi node name"),
349 'Topology should contain 2 otsi nodes')
350 self.assertEqual(1, count_object_with_double_key(nodes, "name", "value-name", "dsr/odu node name"),
351 'Topology should contain 1 dsr node')
352 self.assertEqual(1, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
353 'Topology should contain 1 transitional link')
354 self.assertEqual(1, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
355 'Topology should contain 1 oms link')
357 def test_15_connect_xpdrc(self):
358 response = test_utils.mount_device("XPDR-C1", ('xpdrc', self.NODE_VERSION))
359 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
362 def test_16_connect_xprdc_n1_to_roadmc_pp1(self):
363 response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "1",
364 "ROADM-C1", "1", "SRG1-PP1-TXRX")
365 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
366 res = response.json()
367 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
368 CREATED_SUCCESSFULLY)
371 def test_17_connect_roadmc_pp1_to_xpdrc_n1(self):
372 response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "1",
373 "ROADM-C1", "1", "SRG1-PP1-TXRX")
374 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
375 res = response.json()
376 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
377 CREATED_SUCCESSFULLY)
380 def test_18_check_tapi_topology_T100G(self):
381 url = "{}/operations/tapi-topology:get-topology-details"
383 "tapi-topology:input": {
384 "tapi-topology:topology-id-or-name": "Transponder 100GE"
387 response = test_utils.post_request(url, data)
388 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
389 res = response.json()
390 self.assertEqual(2, len(res["output"]["topology"]["node"][0]["owned-node-edge-point"]),
391 'Node should contain 2 owned-node-edge-points')
392 self.assertEqual("XPDR1-CLIENT1",
393 res["output"]["topology"]["node"][0]["owned-node-edge-point"][0]["name"][0]["value"],
394 'name of owned-node-edge-points should be XPDR1-CLIENT1')
395 self.assertEqual("XPDR1-CLIENT1",
396 res["output"]["topology"]["node"][0]["owned-node-edge-point"][1]["name"][0]["value"],
397 'name of owned-node-edge-points should be XPDR1-CLIENT1')
399 def test_19_check_tapi_topology_T0(self):
400 url = "{}/operations/tapi-topology:get-topology-details"
402 "tapi-topology:input": {
403 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
406 response = test_utils.post_request(url, data)
407 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
408 res = response.json()
409 nodes = res["output"]["topology"]["node"]
410 links = res["output"]["topology"]["link"]
411 self.assertEqual(5, len(nodes), 'Topology should contain 5 nodes')
412 self.assertEqual(4, len(links), 'Topology should contain 4 links')
413 self.assertEqual(3, count_object_with_double_key(nodes, "name", "value-name", "otsi node name"),
414 'Topology should contain 3 otsi nodes')
415 self.assertEqual(2, count_object_with_double_key(nodes, "name", "value-name", "dsr/odu node name"),
416 'Topology should contain 2 dsr nodes')
417 self.assertEqual(2, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
418 'Topology should contain 2 transitional links')
419 self.assertEqual(2, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
420 'Topology should contain 2 oms links')
422 def test_20_connect_spdr_sa1(self):
423 response = test_utils.mount_device("SPDR-SA1", ('spdra', self.NODE_VERSION))
424 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
426 # TODO replace connect and disconnect timers with test_utils.wait_until_log_contains
428 def test_21_connect_spdr_sc1(self):
429 response = test_utils.mount_device("SPDR-SC1", ('spdrc', self.NODE_VERSION))
430 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
432 # TODO replace connect and disconnect timers with test_utils.wait_until_log_contains
434 def test_22_check_tapi_topology_T100G(self):
435 self.test_18_check_tapi_topology_T100G()
437 def test_23_check_tapi_topology_T0(self):
438 self.test_19_check_tapi_topology_T0()
440 def test_24_connect_sprda_n1_to_roadma_pp2(self):
441 response = test_utils.connect_xpdr_to_rdm_request("SPDR-SA1", "1", "1",
442 "ROADM-A1", "1", "SRG1-PP2-TXRX")
443 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
444 res = response.json()
445 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
446 CREATED_SUCCESSFULLY)
449 def test_25_connect_roadma_pp2_to_spdra_n1(self):
450 response = test_utils.connect_rdm_to_xpdr_request("SPDR-SA1", "1", "1",
451 "ROADM-A1", "1", "SRG1-PP2-TXRX")
452 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
453 res = response.json()
454 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
455 CREATED_SUCCESSFULLY)
458 def test_26_connect_sprdc_n1_to_roadmc_pp2(self):
459 response = test_utils.connect_xpdr_to_rdm_request("SPDR-SC1", "1", "1",
460 "ROADM-C1", "1", "SRG1-PP2-TXRX")
461 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
462 res = response.json()
463 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
464 CREATED_SUCCESSFULLY)
467 def test_27_connect_roadmc_pp2_to_spdrc_n1(self):
468 response = test_utils.connect_rdm_to_xpdr_request("SPDR-SC1", "1", "1",
469 "ROADM-C1", "1", "SRG1-PP2-TXRX")
470 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
471 res = response.json()
472 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
473 CREATED_SUCCESSFULLY)
476 def test_28_check_tapi_topology_T100G(self):
477 self.test_18_check_tapi_topology_T100G()
479 def test_29_check_tapi_topology_T0(self):
480 url = "{}/operations/tapi-topology:get-topology-details"
482 "tapi-topology:input": {
483 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
486 response = test_utils.post_request(url, data)
487 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
488 res = response.json()
489 nodes = res["output"]["topology"]["node"]
490 links = res["output"]["topology"]["link"]
491 self.assertEqual(9, len(nodes), 'Topology should contain 9 nodes')
492 self.assertEqual(8, len(links), 'Topology should contain 8 links')
493 self.assertEqual(5, count_object_with_double_key(nodes, "name", "value-name", "otsi node name"),
494 'Topology should contain 5 otsi nodes')
495 self.assertEqual(4, count_object_with_double_key(nodes, "name", "value-name", "dsr/odu node name"),
496 'Topology should contain 4 dsr nodes')
497 self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
498 'Topology should contain 4 transitional links')
499 self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
500 'Topology should contain 4 oms links')
502 def test_30_add_oms_attributes(self):
503 # Config ROADMA-ROADMC oms-attributes
505 "auto-spanloss": "true",
506 "spanloss-base": 11.4,
507 "spanloss-current": 12,
508 "engineered-spanloss": 12.2,
509 "link-concatenation": [{
512 "SRLG-length": 100000,
514 response = test_utils.add_oms_attr_request("ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", data)
515 self.assertEqual(response.status_code, requests.codes.created)
516 # Config ROADMC-ROADMA oms-attributes
518 "auto-spanloss": "true",
519 "spanloss-base": 11.4,
520 "spanloss-current": 12,
521 "engineered-spanloss": 12.2,
522 "link-concatenation": [{
525 "SRLG-length": 100000,
527 response = test_utils.add_oms_attr_request("ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX", data)
528 self.assertEqual(response.status_code, requests.codes.created)
530 def test_31_create_OCH_OTU4_service(self):
531 response = test_utils.service_create_request(self.cr_serv_sample_data)
532 self.assertEqual(response.status_code, requests.codes.ok)
533 res = response.json()
534 self.assertIn('PCE calculation in progress',
535 res['output']['configuration-response-common']['response-message'])
536 time.sleep(self.WAITING)
538 def test_32_check_tapi_topology_T0(self):
539 url = "{}/operations/tapi-topology:get-topology-details"
541 "tapi-topology:input": {
542 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
545 response = test_utils.post_request(url, data)
546 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
547 res = response.json()
548 nodes = res["output"]["topology"]["node"]
549 links = res["output"]["topology"]["link"]
550 self.assertEqual(9, len(nodes), 'Topology should contain 9 nodes')
551 self.assertEqual(9, len(links), 'Topology should contain 9 links')
552 self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
553 'Topology should contain 4 transitional links')
554 self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
555 'Topology should contain 4 oms links')
556 self.assertEqual(1, count_object_with_double_key(links, "name", "value-name", "otn link name"),
557 'Topology should contain 1 otn link')
559 if link["name"][0]["value"] == "OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1":
560 self.assertEqual(100000, link["available-capacity"]["total-size"]["value"],
561 'OTU4 link should have an available capacity of 100 000 Mbps')
562 elif link["name"][0]["value-name"] == "transitional link name":
563 self.assertEqual(100, link["available-capacity"]["total-size"]["value"],
564 'link should have an available capacity of 100 Gbps')
565 self.assertEqual(2, len(link["node-edge-point"]), 'link should have 2 neps')
567 def test_33_create_ODU4_service(self):
568 self.cr_serv_sample_data["input"]["service-name"] = "service1-ODU4"
569 self.cr_serv_sample_data["input"]["service-a-end"]["service-format"] = "ODU"
570 del self.cr_serv_sample_data["input"]["service-a-end"]["otu-service-rate"]
571 self.cr_serv_sample_data["input"]["service-a-end"]["odu-service-rate"] = "org-openroadm-otn-common-types:ODU4"
572 self.cr_serv_sample_data["input"]["service-z-end"]["service-format"] = "ODU"
573 del self.cr_serv_sample_data["input"]["service-z-end"]["otu-service-rate"]
574 self.cr_serv_sample_data["input"]["service-z-end"]["odu-service-rate"] = "org-openroadm-otn-common-types:ODU4"
576 response = test_utils.service_create_request(self.cr_serv_sample_data)
577 self.assertEqual(response.status_code, requests.codes.ok)
578 res = response.json()
579 self.assertIn('PCE calculation in progress',
580 res['output']['configuration-response-common']['response-message'])
581 time.sleep(self.WAITING)
583 def test_34_check_tapi_topology_T0(self):
584 url = "{}/operations/tapi-topology:get-topology-details"
586 "tapi-topology:input": {
587 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
590 response = test_utils.post_request(url, data)
591 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
592 res = response.json()
593 nodes = res["output"]["topology"]["node"]
594 links = res["output"]["topology"]["link"]
595 self.assertEqual(9, len(nodes), 'Topology should contain 9 nodes')
596 self.assertEqual(10, len(links), 'Topology should contain 10 links')
597 self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
598 'Topology should contain 4 transitional links')
599 self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
600 'Topology should contain 4 oms links')
601 self.assertEqual(2, count_object_with_double_key(links, "name", "value-name", "otn link name"),
602 'Topology should contain 2 otn links')
604 if link["name"][0]["value"] == "OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1":
605 self.assertEqual(0, link["available-capacity"]["total-size"]["value"],
606 'OTU4 link should have an available capacity of 0 Mbps')
607 elif link["name"][0]["value"] == "ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1":
608 self.assertEqual(100000, link["available-capacity"]["total-size"]["value"],
609 'ODU4 link should have an available capacity of 100 000 Mbps')
610 elif link["name"][0]["value-name"] == "transitional link name":
611 self.assertEqual(100, link["available-capacity"]["total-size"]["value"],
612 'link should have an available capacity of 100 Gbps')
613 self.assertEqual(2, len(link["node-edge-point"]), 'link should have 2 neps')
615 def test_35_connect_sprda_2_n2_to_roadma_pp3(self):
616 response = test_utils.connect_xpdr_to_rdm_request("SPDR-SA1", "2", "2",
617 "ROADM-A1", "1", "SRG1-PP3-TXRX")
618 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
619 res = response.json()
620 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
621 CREATED_SUCCESSFULLY)
624 def test_36_connect_roadma_pp3_to_spdra_2_n2(self):
625 response = test_utils.connect_rdm_to_xpdr_request("SPDR-SA1", "2", "2",
626 "ROADM-A1", "1", "SRG1-PP3-TXRX")
627 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
628 res = response.json()
629 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
630 CREATED_SUCCESSFULLY)
633 def test_37_check_tapi_topology_T0(self):
634 url = "{}/operations/tapi-topology:get-topology-details"
636 "tapi-topology:input": {
637 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
640 response = test_utils.post_request(url, data)
641 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
642 res = response.json()
643 nodes = res["output"]["topology"]["node"]
644 links = res["output"]["topology"]["link"]
645 self.assertEqual(11, len(nodes), 'Topology should contain 11 nodes')
646 self.assertEqual(12, len(links), 'Topology should contain 12 links')
647 self.assertEqual(6, count_object_with_double_key(nodes, "name", "value-name", "otsi node name"),
648 'Topology should contain 6 otsi nodes')
649 self.assertEqual(5, count_object_with_double_key(nodes, "name", "value-name", "dsr/odu node name"),
650 'Topology should contain 5 dsr nodes')
651 self.assertEqual(5, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
652 'Topology should contain 5 transitional links')
653 self.assertEqual(5, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
654 'Topology should contain 5 oms links')
655 self.assertEqual(2, count_object_with_double_key(links, "name", "value-name", "otn link name"),
656 'Topology should contain 2 otn links')
658 def test_38_delete_ODU4_service(self):
659 response = test_utils.service_delete_request("service1-ODU4")
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_39_delete_OCH_OTU4_service(self):
667 response = test_utils.service_delete_request("service1-OCH-OTU4")
668 self.assertEqual(response.status_code, requests.codes.ok)
669 res = response.json()
670 self.assertIn('Renderer service delete in progress',
671 res['output']['configuration-response-common']['response-message'])
674 def test_40_check_tapi_topology_T0(self):
675 url = "{}/operations/tapi-topology:get-topology-details"
677 "tapi-topology:input": {
678 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
681 response = test_utils.post_request(url, data)
682 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
683 res = response.json()
684 nodes = res["output"]["topology"]["node"]
685 links = res["output"]["topology"]["link"]
686 self.assertEqual(11, len(nodes), 'Topology should contain 11 nodes')
687 self.assertEqual(10, len(links), 'Topology should contain 10 links')
688 self.assertEqual(0, count_object_with_double_key(links, "name", "value-name", "otn link name"),
689 'Topology should contain 0 otn link')
691 def test_41_disconnect_xponders_from_roadm(self):
692 url = "{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:link/"
693 response = test_utils.get_ordm_topo_request("")
694 self.assertEqual(response.status_code, requests.codes.ok)
695 res = response.json()
696 links = res['network'][0]['ietf-network-topology:link']
698 if (link["org-openroadm-common-network:link-type"] == "XPONDER-OUTPUT" or
699 link["org-openroadm-common-network:link-type"] == "XPONDER-INPUT"):
700 link_name = link["link-id"]
701 response = test_utils.delete_request(url+link_name)
702 self.assertEqual(response.status_code, requests.codes.ok)
704 def test_42_check_tapi_topology_T0(self):
705 url = "{}/operations/tapi-topology:get-topology-details"
707 "tapi-topology:input": {
708 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
711 response = test_utils.post_request(url, data)
712 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
713 res = response.json()
714 nodes = res["output"]["topology"]["node"]
715 self.assertEqual(1, len(nodes), 'Topology should contain 1 node')
716 self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
717 self.assertEqual("ROADM-infra", res["output"]["topology"]["node"][0]["name"][0]["value"],
718 'node name should be: ROADM-infra')
720 def test_43_get_tapi_topology_T100G(self):
721 url = "{}/operations/tapi-topology:get-topology-details"
723 "tapi-topology:input": {
724 "tapi-topology:topology-id-or-name": "Transponder 100GE"
727 response = test_utils.post_request(url, data)
728 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
729 res = response.json()
730 self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'Topology should contain 1 node')
731 self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
732 self.assertNotIn("owned-node-edge-point", res["output"]["topology"]["node"][0],
733 'Node should contain no owned-node-edge-points')
735 def test_44_disconnect_roadma(self):
736 response = test_utils.unmount_device("ROADM-A1")
737 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
740 def test_45_disconnect_roadmc(self):
741 response = test_utils.unmount_device("ROADM-C1")
742 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
745 def test_46_check_tapi_topos(self):
746 self.test_01_get_tapi_topology_T100G()
747 self.test_02_get_tapi_topology_T0()
749 def test_47_disconnect_xpdra(self):
750 response = test_utils.unmount_device("XPDR-A1")
751 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
754 def test_48_disconnect_xpdrc(self):
755 response = test_utils.unmount_device("XPDR-C1")
756 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
759 def test_49_disconnect_spdr_sa1(self):
760 response = test_utils.unmount_device("SPDR-SA1")
761 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
764 def test_50_disconnect_spdr_sc1(self):
765 response = test_utils.unmount_device("SPDR-SC1")
766 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
769 def find_object_with_key(list_dicts, key, value):
770 for dict_ in list_dicts:
771 if dict_[key] == value:
776 def count_object_with_double_key(list_dicts, key1, key2, value):
778 for dict in list_dicts:
779 if dict[key1][0][key2] == value:
784 if __name__ == "__main__":
785 unittest.main(verbosity=2)