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
22 sys.path.append('transportpce_tests/common/')
26 CREATED_SUCCESSFULLY = 'Result message should contain Xponder Roadm Link created successfully'
29 class TransportTapitesting(unittest.TestCase):
33 NODE_VERSION = '2.2.1'
34 cr_serv_sample_data = {"input": {
35 "sdnc-request-header": {
36 "request-id": "request-1",
37 "rpc-action": "service-create",
38 "request-system-id": "appname"
40 "service-name": "service1-OCH-OTU4",
41 "common-id": "commonId",
42 "connection-type": "infrastructure",
44 "service-rate": "100",
45 "node-id": "SPDR-SA1",
46 "service-format": "OTU",
47 "otu-service-rate": "org-openroadm-otn-common-types:OTU4",
51 "committed-info-rate": "100000",
52 "committed-burst-size": "64"
57 "port-device-name": "SPDR-SA1-XPDR1",
59 "port-name": "XPDR1-NETWORK1",
60 "port-rack": "000000.00",
61 "port-shelf": "Chassis#1"
64 "lgx-device-name": "Some lgx-device-name",
65 "lgx-port-name": "Some lgx-port-name",
66 "lgx-port-rack": "000000.00",
67 "lgx-port-shelf": "00"
72 "port-device-name": "SPDR-SA1-XPDR1",
74 "port-name": "XPDR1-NETWORK1",
75 "port-rack": "000000.00",
76 "port-shelf": "Chassis#1"
79 "lgx-device-name": "Some lgx-device-name",
80 "lgx-port-name": "Some lgx-port-name",
81 "lgx-port-rack": "000000.00",
82 "lgx-port-shelf": "00"
88 "service-rate": "100",
89 "node-id": "SPDR-SC1",
90 "service-format": "OTU",
91 "otu-service-rate": "org-openroadm-otn-common-types:OTU4",
95 "committed-info-rate": "100000",
96 "committed-burst-size": "64"
101 "port-device-name": "SPDR-SC1-XPDR1",
102 "port-type": "fixed",
103 "port-name": "XPDR1-NETWORK1",
104 "port-rack": "000000.00",
105 "port-shelf": "Chassis#1"
108 "lgx-device-name": "Some lgx-device-name",
109 "lgx-port-name": "Some lgx-port-name",
110 "lgx-port-rack": "000000.00",
111 "lgx-port-shelf": "00"
116 "port-device-name": "SPDR-SC1-XPDR1",
117 "port-type": "fixed",
118 "port-name": "XPDR1-NETWORK1",
119 "port-rack": "000000.00",
120 "port-shelf": "Chassis#1"
123 "lgx-device-name": "Some lgx-device-name",
124 "lgx-port-name": "Some lgx-port-name",
125 "lgx-port-rack": "000000.00",
126 "lgx-port-shelf": "00"
131 "due-date": "2018-06-15T00:00:01Z",
132 "operator-contact": "pw1234"
138 cls.init_failed = False
139 os.environ['JAVA_MIN_MEM'] = '1024M'
140 os.environ['JAVA_MAX_MEM'] = '4096M'
141 cls.processes = test_utils.start_tpce()
142 # TAPI feature is not installed by default in Karaf
143 if "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
144 print("installing tapi feature...")
145 result = test_utils.install_karaf_feature("odl-transportpce-tapi")
146 if result.returncode != 0:
147 cls.init_failed = True
148 print("Restarting OpenDaylight...")
149 test_utils.shutdown_process(cls.processes[0])
150 cls.processes[0] = test_utils.start_karaf()
151 test_utils.process_list[0] = cls.processes[0]
152 cls.init_failed = not test_utils.wait_until_log_contains(
153 test_utils.KARAF_LOG, test_utils.KARAF_OK_START_MSG, time_to_wait=60)
155 print("tapi installation feature failed...")
156 test_utils.shutdown_process(cls.processes[0])
158 cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION),
159 ('roadma', cls.NODE_VERSION),
160 ('roadmb', cls.NODE_VERSION),
161 ('roadmc', cls.NODE_VERSION),
162 ('xpdrc', cls.NODE_VERSION),
163 ('spdra', cls.NODE_VERSION),
164 ('spdrc', cls.NODE_VERSION)])
167 def tearDownClass(cls):
168 # pylint: disable=not-an-iterable
169 for process in cls.processes:
170 test_utils.shutdown_process(process)
171 print("all processes killed")
173 def setUp(self): # instruction executed before each test method
175 self.fail('Feature installation failed')
176 print("execution of {}".format(self.id().split(".")[-1]))
178 def test_01_get_tapi_topology_T100G(self):
179 url = "{}/operations/tapi-topology:get-topology-details"
181 "tapi-topology:input": {
182 "tapi-topology:topology-id-or-name": "Transponder 100GE"
185 response = test_utils.post_request(url, data)
186 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
187 res = response.json()
188 self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'Topology should contain 1 node')
189 self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
190 self.assertNotIn("owned-node-edge-point", res["output"]["topology"]["node"][0],
191 'Node should contain no owned-node-edge-points')
192 self.assertEqual("Tpdr100g over WDM node", res["output"]["topology"]["node"][0]["name"][0]["value"],
193 'node name should be: Tpdr100g over WDM node')
194 self.assertIn("ETH", res["output"]["topology"]["node"][0]["layer-protocol-name"],
195 'Node layer protocol should contain ETH')
196 self.assertEqual(1, len(res["output"]["topology"]["node"][0]["node-rule-group"]),
197 'node should contain 1 node rule group')
199 def test_02_get_tapi_topology_T0(self):
200 url = "{}/operations/tapi-topology:get-topology-details"
202 "tapi-topology:input": {
203 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
206 response = test_utils.post_request(url, data)
207 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
208 res = response.json()
209 self.assertNotIn("node", res["output"]["topology"], 'Topology should contain no node')
210 self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
212 def test_03_connect_rdmb(self):
213 response = test_utils.mount_device("ROADM-B1", ('roadmb', self.NODE_VERSION))
214 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
217 def test_04_check_tapi_topos(self):
218 url = "{}/operations/tapi-topology:get-topology-details"
220 "tapi-topology:input": {
221 "tapi-topology:topology-id-or-name": "Transponder 100GE"
224 response = test_utils.post_request(url, data)
225 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
226 res = response.json()
227 self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'Topology should contain 1 node')
228 self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
230 url = "{}/operations/tapi-topology:get-topology-details"
232 "tapi-topology:input": {
233 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
236 response = test_utils.post_request(url, data)
237 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
238 res = response.json()
239 self.assertNotIn("node", res["output"]["topology"], 'Topology should contain no node')
240 self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
242 def test_05_disconnect_roadmb(self):
243 response = test_utils.unmount_device("ROADM-B1")
244 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
247 def test_06_connect_xpdra(self):
248 response = test_utils.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION))
249 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
252 def test_07_check_tapi_topos(self):
253 self.test_04_check_tapi_topos()
255 def test_08_connect_rdma(self):
256 response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION))
257 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
260 def test_09_connect_rdmc(self):
261 response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION))
262 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
265 def test_10_check_tapi_topos(self):
266 self.test_01_get_tapi_topology_T100G()
268 url = "{}/operations/tapi-topology:get-topology-details"
270 "tapi-topology:input": {
271 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
274 response = test_utils.post_request(url, data)
275 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
276 res = response.json()
277 self.assertEqual(1, len(res["output"]["topology"]["node"]), 'Topology should contain 1 node')
278 self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
279 self.assertEqual("ROADM-infra", res["output"]["topology"]["node"][0]["name"][0]["value"],
280 'node name should be: ROADM-infra')
281 self.assertIn("PHOTONIC_MEDIA", res["output"]["topology"]["node"][0]["layer-protocol-name"],
282 'Node layer protocol should contain PHOTONIC_MEDIA')
283 self.assertEqual(1, len(res["output"]["topology"]["node"][0]["node-rule-group"]),
284 'node should contain 1 node rule group')
286 def test_11_connect_xprda_n1_to_roadma_pp1(self):
287 response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "1",
288 "ROADM-A1", "1", "SRG1-PP1-TXRX")
289 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
290 res = response.json()
291 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
292 CREATED_SUCCESSFULLY)
295 def test_12_connect_roadma_pp1_to_xpdra_n1(self):
296 response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "1",
297 "ROADM-A1", "1", "SRG1-PP1-TXRX")
298 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
299 res = response.json()
300 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
301 CREATED_SUCCESSFULLY)
304 def test_13_check_tapi_topology_T100G(self):
305 url = "{}/operations/tapi-topology:get-topology-details"
307 "tapi-topology:input": {
308 "tapi-topology:topology-id-or-name": "Transponder 100GE"
311 response = test_utils.post_request(url, data)
312 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
313 res = response.json()
314 self.assertEqual(1, len(res["output"]["topology"]["node"][0]["owned-node-edge-point"]),
315 'Node should contain 1 owned-node-edge-points')
316 self.assertEqual("XPDR1-CLIENT1",
317 res["output"]["topology"]["node"][0]["owned-node-edge-point"][0]["name"][0]["value"],
318 'name of owned-node-edge-points should be XPDR1-CLIENT1')
320 def test_14_check_tapi_topology_T0(self):
321 url = "{}/operations/tapi-topology:get-topology-details"
323 "tapi-topology:input": {
324 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
327 response = test_utils.post_request(url, data)
328 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
329 res = response.json()
330 nodes = res["output"]["topology"]["node"]
331 links = res["output"]["topology"]["link"]
332 self.assertEqual(3, len(nodes), 'Topology should contain 3 nodes')
333 self.assertEqual(2, len(links), 'Topology should contain 2 links')
334 self.assertEqual(2, count_object_with_double_key(nodes, "name", "value-name", "otsi node name"),
335 'Topology should contain 2 otsi nodes')
336 self.assertEqual(1, count_object_with_double_key(nodes, "name", "value-name", "dsr/odu node name"),
337 'Topology should contain 1 dsr node')
338 self.assertEqual(1, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
339 'Topology should contain 1 transitional link')
340 self.assertEqual(1, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
341 'Topology should contain 1 oms link')
343 def test_15_connect_xpdrc(self):
344 response = test_utils.mount_device("XPDR-C1", ('xpdrc', self.NODE_VERSION))
345 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
348 def test_16_connect_xprdc_n1_to_roadmc_pp1(self):
349 response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "1",
350 "ROADM-C1", "1", "SRG1-PP1-TXRX")
351 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
352 res = response.json()
353 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
354 CREATED_SUCCESSFULLY)
357 def test_17_connect_roadmc_pp1_to_xpdrc_n1(self):
358 response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "1",
359 "ROADM-C1", "1", "SRG1-PP1-TXRX")
360 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
361 res = response.json()
362 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
363 CREATED_SUCCESSFULLY)
366 def test_18_check_tapi_topology_T100G(self):
367 url = "{}/operations/tapi-topology:get-topology-details"
369 "tapi-topology:input": {
370 "tapi-topology:topology-id-or-name": "Transponder 100GE"
373 response = test_utils.post_request(url, data)
374 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
375 res = response.json()
376 self.assertEqual(2, len(res["output"]["topology"]["node"][0]["owned-node-edge-point"]),
377 'Node should contain 2 owned-node-edge-points')
378 self.assertEqual("XPDR1-CLIENT1",
379 res["output"]["topology"]["node"][0]["owned-node-edge-point"][0]["name"][0]["value"],
380 'name of owned-node-edge-points should be XPDR1-CLIENT1')
381 self.assertEqual("XPDR1-CLIENT1",
382 res["output"]["topology"]["node"][0]["owned-node-edge-point"][1]["name"][0]["value"],
383 'name of owned-node-edge-points should be XPDR1-CLIENT1')
385 def test_19_check_tapi_topology_T0(self):
386 url = "{}/operations/tapi-topology:get-topology-details"
388 "tapi-topology:input": {
389 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
392 response = test_utils.post_request(url, data)
393 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
394 res = response.json()
395 nodes = res["output"]["topology"]["node"]
396 links = res["output"]["topology"]["link"]
397 self.assertEqual(5, len(nodes), 'Topology should contain 5 nodes')
398 self.assertEqual(4, len(links), 'Topology should contain 4 links')
399 self.assertEqual(3, count_object_with_double_key(nodes, "name", "value-name", "otsi node name"),
400 'Topology should contain 3 otsi nodes')
401 self.assertEqual(2, count_object_with_double_key(nodes, "name", "value-name", "dsr/odu node name"),
402 'Topology should contain 2 dsr nodes')
403 self.assertEqual(2, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
404 'Topology should contain 2 transitional links')
405 self.assertEqual(2, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
406 'Topology should contain 2 oms links')
408 def test_20_connect_spdr_sa1(self):
409 response = test_utils.mount_device("SPDR-SA1", ('spdra', self.NODE_VERSION))
410 self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
412 # TODO replace connect and disconnect timers with test_utils.wait_until_log_contains
414 def test_21_connect_spdr_sc1(self):
415 response = test_utils.mount_device("SPDR-SC1", ('spdrc', 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_22_check_tapi_topology_T100G(self):
421 self.test_18_check_tapi_topology_T100G()
423 def test_23_check_tapi_topology_T0(self):
424 self.test_19_check_tapi_topology_T0()
426 def test_24_connect_sprda_n1_to_roadma_pp2(self):
427 response = test_utils.connect_xpdr_to_rdm_request("SPDR-SA1", "1", "1",
428 "ROADM-A1", "1", "SRG1-PP2-TXRX")
429 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
430 res = response.json()
431 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
432 CREATED_SUCCESSFULLY)
435 def test_25_connect_roadma_pp2_to_spdra_n1(self):
436 response = test_utils.connect_rdm_to_xpdr_request("SPDR-SA1", "1", "1",
437 "ROADM-A1", "1", "SRG1-PP2-TXRX")
438 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
439 res = response.json()
440 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
441 CREATED_SUCCESSFULLY)
444 def test_26_connect_sprdc_n1_to_roadmc_pp2(self):
445 response = test_utils.connect_xpdr_to_rdm_request("SPDR-SC1", "1", "1",
446 "ROADM-C1", "1", "SRG1-PP2-TXRX")
447 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
448 res = response.json()
449 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
450 CREATED_SUCCESSFULLY)
453 def test_27_connect_roadmc_pp2_to_spdrc_n1(self):
454 response = test_utils.connect_rdm_to_xpdr_request("SPDR-SC1", "1", "1",
455 "ROADM-C1", "1", "SRG1-PP2-TXRX")
456 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
457 res = response.json()
458 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
459 CREATED_SUCCESSFULLY)
462 def test_28_check_tapi_topology_T100G(self):
463 self.test_18_check_tapi_topology_T100G()
465 def test_29_check_tapi_topology_T0(self):
466 url = "{}/operations/tapi-topology:get-topology-details"
468 "tapi-topology:input": {
469 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
472 response = test_utils.post_request(url, data)
473 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
474 res = response.json()
475 nodes = res["output"]["topology"]["node"]
476 links = res["output"]["topology"]["link"]
477 self.assertEqual(9, len(nodes), 'Topology should contain 9 nodes')
478 self.assertEqual(8, len(links), 'Topology should contain 8 links')
479 self.assertEqual(5, count_object_with_double_key(nodes, "name", "value-name", "otsi node name"),
480 'Topology should contain 5 otsi nodes')
481 self.assertEqual(4, count_object_with_double_key(nodes, "name", "value-name", "dsr/odu node name"),
482 'Topology should contain 4 dsr nodes')
483 self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
484 'Topology should contain 4 transitional links')
485 self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
486 'Topology should contain 4 oms links')
488 def test_30_add_oms_attributes(self):
489 # Config ROADMA-ROADMC oms-attributes
491 "auto-spanloss": "true",
492 "spanloss-base": 11.4,
493 "spanloss-current": 12,
494 "engineered-spanloss": 12.2,
495 "link-concatenation": [{
498 "SRLG-length": 100000,
500 response = test_utils.add_oms_attr_request("ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", data)
501 self.assertEqual(response.status_code, requests.codes.created)
502 # Config ROADMC-ROADMA oms-attributes
504 "auto-spanloss": "true",
505 "spanloss-base": 11.4,
506 "spanloss-current": 12,
507 "engineered-spanloss": 12.2,
508 "link-concatenation": [{
511 "SRLG-length": 100000,
513 response = test_utils.add_oms_attr_request("ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX", data)
514 self.assertEqual(response.status_code, requests.codes.created)
516 def test_31_create_OCH_OTU4_service(self):
517 response = test_utils.service_create_request(self.cr_serv_sample_data)
518 self.assertEqual(response.status_code, requests.codes.ok)
519 res = response.json()
520 self.assertIn('PCE calculation in progress',
521 res['output']['configuration-response-common']['response-message'])
522 time.sleep(self.WAITING)
524 def test_32_check_tapi_topology_T0(self):
525 url = "{}/operations/tapi-topology:get-topology-details"
527 "tapi-topology:input": {
528 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
531 response = test_utils.post_request(url, data)
532 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
533 res = response.json()
534 nodes = res["output"]["topology"]["node"]
535 links = res["output"]["topology"]["link"]
536 self.assertEqual(9, len(nodes), 'Topology should contain 9 nodes')
537 self.assertEqual(9, len(links), 'Topology should contain 9 links')
538 self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
539 'Topology should contain 4 transitional links')
540 self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
541 'Topology should contain 4 oms links')
542 self.assertEqual(1, count_object_with_double_key(links, "name", "value-name", "otn link name"),
543 'Topology should contain 1 otn link')
545 if link["name"][0]["value"] == "OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1":
546 self.assertEqual(100000, link["available-capacity"]["total-size"]["value"],
547 'OTU4 link should have an available capacity of 100 000 Mbps')
548 elif link["name"][0]["value-name"] == "transitional link name":
549 self.assertEqual(100, link["available-capacity"]["total-size"]["value"],
550 'link should have an available capacity of 100 Gbps')
551 self.assertEqual(2, len(link["node-edge-point"]), 'link should have 2 neps')
553 def test_33_create_ODU4_service(self):
554 self.cr_serv_sample_data["input"]["service-name"] = "service1-ODU4"
555 self.cr_serv_sample_data["input"]["service-a-end"]["service-format"] = "ODU"
556 del self.cr_serv_sample_data["input"]["service-a-end"]["otu-service-rate"]
557 self.cr_serv_sample_data["input"]["service-a-end"]["odu-service-rate"] = "org-openroadm-otn-common-types:ODU4"
558 self.cr_serv_sample_data["input"]["service-z-end"]["service-format"] = "ODU"
559 del self.cr_serv_sample_data["input"]["service-z-end"]["otu-service-rate"]
560 self.cr_serv_sample_data["input"]["service-z-end"]["odu-service-rate"] = "org-openroadm-otn-common-types:ODU4"
562 response = test_utils.service_create_request(self.cr_serv_sample_data)
563 self.assertEqual(response.status_code, requests.codes.ok)
564 res = response.json()
565 self.assertIn('PCE calculation in progress',
566 res['output']['configuration-response-common']['response-message'])
567 time.sleep(self.WAITING)
569 def test_34_check_tapi_topology_T0(self):
570 url = "{}/operations/tapi-topology:get-topology-details"
572 "tapi-topology:input": {
573 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
576 response = test_utils.post_request(url, data)
577 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
578 res = response.json()
579 nodes = res["output"]["topology"]["node"]
580 links = res["output"]["topology"]["link"]
581 self.assertEqual(9, len(nodes), 'Topology should contain 9 nodes')
582 self.assertEqual(10, len(links), 'Topology should contain 10 links')
583 self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
584 'Topology should contain 4 transitional links')
585 self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
586 'Topology should contain 4 oms links')
587 self.assertEqual(2, count_object_with_double_key(links, "name", "value-name", "otn link name"),
588 'Topology should contain 2 otn links')
590 if link["name"][0]["value"] == "OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1":
591 self.assertEqual(0, link["available-capacity"]["total-size"]["value"],
592 'OTU4 link should have an available capacity of 0 Mbps')
593 elif link["name"][0]["value"] == "ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1":
594 self.assertEqual(100000, link["available-capacity"]["total-size"]["value"],
595 'ODU4 link should have an available capacity of 100 000 Mbps')
596 elif link["name"][0]["value-name"] == "transitional link name":
597 self.assertEqual(100, link["available-capacity"]["total-size"]["value"],
598 'link should have an available capacity of 100 Gbps')
599 self.assertEqual(2, len(link["node-edge-point"]), 'link should have 2 neps')
601 def test_35_connect_sprda_2_n2_to_roadma_pp3(self):
602 response = test_utils.connect_xpdr_to_rdm_request("SPDR-SA1", "2", "2",
603 "ROADM-A1", "1", "SRG1-PP3-TXRX")
604 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
605 res = response.json()
606 self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
607 CREATED_SUCCESSFULLY)
610 def test_36_connect_roadma_pp3_to_spdra_2_n2(self):
611 response = test_utils.connect_rdm_to_xpdr_request("SPDR-SA1", "2", "2",
612 "ROADM-A1", "1", "SRG1-PP3-TXRX")
613 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
614 res = response.json()
615 self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
616 CREATED_SUCCESSFULLY)
619 def test_37_check_tapi_topology_T0(self):
620 url = "{}/operations/tapi-topology:get-topology-details"
622 "tapi-topology:input": {
623 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
626 response = test_utils.post_request(url, data)
627 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
628 res = response.json()
629 nodes = res["output"]["topology"]["node"]
630 links = res["output"]["topology"]["link"]
631 self.assertEqual(11, len(nodes), 'Topology should contain 11 nodes')
632 self.assertEqual(12, len(links), 'Topology should contain 12 links')
633 self.assertEqual(6, count_object_with_double_key(nodes, "name", "value-name", "otsi node name"),
634 'Topology should contain 6 otsi nodes')
635 self.assertEqual(5, count_object_with_double_key(nodes, "name", "value-name", "dsr/odu node name"),
636 'Topology should contain 5 dsr nodes')
637 self.assertEqual(5, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
638 'Topology should contain 5 transitional links')
639 self.assertEqual(5, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
640 'Topology should contain 5 oms links')
641 self.assertEqual(2, count_object_with_double_key(links, "name", "value-name", "otn link name"),
642 'Topology should contain 2 otn links')
644 def test_38_delete_ODU4_service(self):
645 response = test_utils.service_delete_request("service1-ODU4")
646 self.assertEqual(response.status_code, requests.codes.ok)
647 res = response.json()
648 self.assertIn('Renderer service delete in progress',
649 res['output']['configuration-response-common']['response-message'])
652 def test_39_delete_OCH_OTU4_service(self):
653 response = test_utils.service_delete_request("service1-OCH-OTU4")
654 self.assertEqual(response.status_code, requests.codes.ok)
655 res = response.json()
656 self.assertIn('Renderer service delete in progress',
657 res['output']['configuration-response-common']['response-message'])
660 def test_40_check_tapi_topology_T0(self):
661 url = "{}/operations/tapi-topology:get-topology-details"
663 "tapi-topology:input": {
664 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
667 response = test_utils.post_request(url, data)
668 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
669 res = response.json()
670 nodes = res["output"]["topology"]["node"]
671 links = res["output"]["topology"]["link"]
672 self.assertEqual(11, len(nodes), 'Topology should contain 11 nodes')
673 self.assertEqual(10, len(links), 'Topology should contain 10 links')
674 self.assertEqual(0, count_object_with_double_key(links, "name", "value-name", "otn link name"),
675 'Topology should contain 0 otn link')
677 def test_41_disconnect_xponders_from_roadm(self):
678 url = "{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:link/"
679 response = test_utils.get_ordm_topo_request("")
680 self.assertEqual(response.status_code, requests.codes.ok)
681 res = response.json()
682 links = res['network'][0]['ietf-network-topology:link']
684 if (link["org-openroadm-common-network:link-type"] == "XPONDER-OUTPUT" or
685 link["org-openroadm-common-network:link-type"] == "XPONDER-INPUT"):
686 link_name = link["link-id"]
687 response = test_utils.delete_request(url+link_name)
688 self.assertEqual(response.status_code, requests.codes.ok)
690 def test_42_check_tapi_topology_T0(self):
691 url = "{}/operations/tapi-topology:get-topology-details"
693 "tapi-topology:input": {
694 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
697 response = test_utils.post_request(url, data)
698 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
699 res = response.json()
700 nodes = res["output"]["topology"]["node"]
701 self.assertEqual(1, len(nodes), 'Topology should contain 1 node')
702 self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
703 self.assertEqual("ROADM-infra", res["output"]["topology"]["node"][0]["name"][0]["value"],
704 'node name should be: ROADM-infra')
706 def test_43_get_tapi_topology_T100G(self):
707 url = "{}/operations/tapi-topology:get-topology-details"
709 "tapi-topology:input": {
710 "tapi-topology:topology-id-or-name": "Transponder 100GE"
713 response = test_utils.post_request(url, data)
714 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
715 res = response.json()
716 self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'Topology should contain 1 node')
717 self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
718 self.assertNotIn("owned-node-edge-point", res["output"]["topology"]["node"][0],
719 'Node should contain no owned-node-edge-points')
721 def test_44_disconnect_roadma(self):
722 response = test_utils.unmount_device("ROADM-A1")
723 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
726 def test_45_disconnect_roadmc(self):
727 response = test_utils.unmount_device("ROADM-C1")
728 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
731 def test_46_check_tapi_topos(self):
732 self.test_01_get_tapi_topology_T100G()
733 self.test_02_get_tapi_topology_T0()
735 def test_47_disconnect_xpdra(self):
736 response = test_utils.unmount_device("XPDR-A1")
737 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
740 def test_48_disconnect_xpdrc(self):
741 response = test_utils.unmount_device("XPDR-C1")
742 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
745 def test_49_disconnect_spdr_sa1(self):
746 response = test_utils.unmount_device("SPDR-SA1")
747 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
750 def test_50_disconnect_spdr_sc1(self):
751 response = test_utils.unmount_device("SPDR-SC1")
752 self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
755 def find_object_with_key(list_dicts, key, value):
756 for dict_ in list_dicts:
757 if dict_[key] == value:
762 def count_object_with_double_key(list_dicts, key1, key2, value):
764 for dict in list_dicts:
765 if dict[key1][0][key2] == value:
770 if __name__ == "__main__":
771 unittest.main(verbosity=2)