tests/transportpce_tests/2.2.1/test_tapi.py

   1 #!/usr/bin/env python
   2 ##############################################################################
   3 # Copyright (c) 2020 Orange, Inc. and others.  All rights reserved.
   4 #
   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 ##############################################################################
  10
  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
  16
  17 import os
  18 import sys
  19 import time
  20 import unittest
  21
  22 import requests
  23
  24 from common import test_utils
  25
  26
  27 CREATED_SUCCESSFULLY = 'Result message should contain Xponder Roadm Link created successfully'
  28
  29 class TransportTapitesting(unittest.TestCase):
  30
  31     processes = None
  32     WAITING = 20
  33     cr_serv_sample_data = {"input": {
  34         "sdnc-request-header": {
  35             "request-id": "request-1",
  36             "rpc-action": "service-create",
  37             "request-system-id": "appname"
  38         },
  39         "service-name": "service1-OCH-OTU4",
  40         "common-id": "commonId",
  41         "connection-type": "infrastructure",
  42         "service-a-end": {
  43             "service-rate": "100",
  44             "node-id": "SPDR-SA1",
  45             "service-format": "OTU",
  46             "otu-service-rate": "org-openroadm-otn-common-types:OTU4",
  47             "clli": "NodeSA",
  48             "subrate-eth-sla": {
  49                     "subrate-eth-sla": {
  50                         "committed-info-rate": "100000",
  51                         "committed-burst-size": "64"
  52                     }
  53             },
  54             "tx-direction": {
  55                 "port": {
  56                     "port-device-name": "SPDR-SA1-XPDR1",
  57                     "port-type": "fixed",
  58                     "port-name": "XPDR1-NETWORK1",
  59                     "port-rack": "000000.00",
  60                     "port-shelf": "Chassis#1"
  61                 },
  62                 "lgx": {
  63                     "lgx-device-name": "Some lgx-device-name",
  64                     "lgx-port-name": "Some lgx-port-name",
  65                     "lgx-port-rack": "000000.00",
  66                     "lgx-port-shelf": "00"
  67                 }
  68             },
  69             "rx-direction": {
  70                 "port": {
  71                     "port-device-name": "SPDR-SA1-XPDR1",
  72                     "port-type": "fixed",
  73                     "port-name": "XPDR1-NETWORK1",
  74                     "port-rack": "000000.00",
  75                     "port-shelf": "Chassis#1"
  76                 },
  77                 "lgx": {
  78                     "lgx-device-name": "Some lgx-device-name",
  79                     "lgx-port-name": "Some lgx-port-name",
  80                     "lgx-port-rack": "000000.00",
  81                     "lgx-port-shelf": "00"
  82                 }
  83             },
  84             "optic-type": "gray"
  85         },
  86         "service-z-end": {
  87             "service-rate": "100",
  88             "node-id": "SPDR-SC1",
  89             "service-format": "OTU",
  90             "otu-service-rate": "org-openroadm-otn-common-types:OTU4",
  91             "clli": "NodeSC",
  92             "subrate-eth-sla": {
  93                     "subrate-eth-sla": {
  94                         "committed-info-rate": "100000",
  95                         "committed-burst-size": "64"
  96                     }
  97             },
  98             "tx-direction": {
  99                 "port": {
 100                     "port-device-name": "SPDR-SC1-XPDR1",
 101                     "port-type": "fixed",
 102                     "port-name": "XPDR1-NETWORK1",
 103                     "port-rack": "000000.00",
 104                     "port-shelf": "Chassis#1"
 105                 },
 106                 "lgx": {
 107                     "lgx-device-name": "Some lgx-device-name",
 108                     "lgx-port-name": "Some lgx-port-name",
 109                     "lgx-port-rack": "000000.00",
 110                     "lgx-port-shelf": "00"
 111                 }
 112             },
 113             "rx-direction": {
 114                 "port": {
 115                     "port-device-name": "SPDR-SC1-XPDR1",
 116                     "port-type": "fixed",
 117                     "port-name": "XPDR1-NETWORK1",
 118                     "port-rack": "000000.00",
 119                     "port-shelf": "Chassis#1"
 120                 },
 121                 "lgx": {
 122                     "lgx-device-name": "Some lgx-device-name",
 123                     "lgx-port-name": "Some lgx-port-name",
 124                     "lgx-port-rack": "000000.00",
 125                     "lgx-port-shelf": "00"
 126                 }
 127             },
 128             "optic-type": "gray"
 129         },
 130         "due-date": "2018-06-15T00:00:01Z",
 131         "operator-contact": "pw1234"
 132     }
 133     }
 134
 135     @classmethod
 136     def setUpClass(cls):
 137         cls.init_failed = False
 138         os.environ['JAVA_MIN_MEM'] = '1024M'
 139         os.environ['JAVA_MAX_MEM'] = '4096M'
 140         cls.processes = test_utils.start_tpce()
 141         # TAPI feature is not installed by default in Karaf
 142         if "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
 143             print("installing tapi feature...")
 144             result = test_utils.install_karaf_feature("odl-transportpce-tapi")
 145             if result.returncode != 0:
 146                 cls.init_failed = True
 147             print("Restarting OpenDaylight...")
 148             test_utils.shutdown_process(cls.processes[0])
 149             cls.processes[0] = test_utils.start_karaf()
 150             test_utils.process_list[0] = cls.processes[0]
 151             cls.init_failed = not test_utils.wait_until_log_contains(
 152                 test_utils.KARAF_LOG, test_utils.KARAF_OK_START_MSG, time_to_wait=60)
 153         if cls.init_failed:
 154             print("tapi installation feature failed...")
 155             test_utils.shutdown_process(cls.processes[0])
 156             sys.exit(2)
 157         cls.processes = test_utils.start_sims(['xpdra', 'roadma', 'roadmb', 'roadmc', 'xpdrc', 'spdra', 'spdrc'])
 158
 159     @classmethod
 160     def tearDownClass(cls):
 161         # pylint: disable=not-an-iterable
 162         for process in cls.processes:
 163             test_utils.shutdown_process(process)
 164         print("all processes killed")
 165
 166     def setUp(self):  # instruction executed before each test method
 167         if self.init_failed:
 168             self.fail('Feature installation failed')
 169         print("execution of {}".format(self.id().split(".")[-1]))
 170
 171     def test_01_get_tapi_topology_T100G(self):
 172         url = "{}/operations/tapi-topology:get-topology-details"
 173         data = {
 174             "tapi-topology:input": {
 175                 "tapi-topology:topology-id-or-name": "Transponder 100GE"
 176             }
 177         }
 178         response = test_utils.post_request(url, data)
 179         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 180         res = response.json()
 181         self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'Topology should contain 1 node')
 182         self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
 183         self.assertNotIn("owned-node-edge-point", res["output"]["topology"]["node"][0],
 184                          'Node should contain no owned-node-edge-points')
 185         self.assertEqual("Tpdr100g over WDM node", res["output"]["topology"]["node"][0]["name"][0]["value"],
 186                          'node name should be: Tpdr100g over WDM node')
 187         self.assertIn("ETH", res["output"]["topology"]["node"][0]["layer-protocol-name"],
 188                          'Node layer protocol should contain ETH')
 189         self.assertEqual(1, len(res["output"]["topology"]["node"][0]["node-rule-group"]),
 190                          'node should contain 1 node rule group')
 191
 192     def test_02_get_tapi_topology_T0(self):
 193         url = "{}/operations/tapi-topology:get-topology-details"
 194         data = {
 195             "tapi-topology:input": {
 196                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 197             }
 198         }
 199         response = test_utils.post_request(url, data)
 200         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 201         res = response.json()
 202         self.assertNotIn("node", res["output"]["topology"], 'Topology should contain no node')
 203         self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
 204
 205     def test_03_connect_rdmb(self):
 206         response = test_utils.mount_device("ROADM-B1", 'roadmb')
 207         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 208         time.sleep(10)
 209
 210     def test_04_check_tapi_topos(self):
 211         url = "{}/operations/tapi-topology:get-topology-details"
 212         data = {
 213             "tapi-topology:input": {
 214                 "tapi-topology:topology-id-or-name": "Transponder 100GE"
 215             }
 216         }
 217         response = test_utils.post_request(url, data)
 218         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 219         res = response.json()
 220         self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'Topology should contain 1 node')
 221         self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
 222
 223         url = "{}/operations/tapi-topology:get-topology-details"
 224         data = {
 225             "tapi-topology:input": {
 226                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 227             }
 228         }
 229         response = test_utils.post_request(url, data)
 230         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 231         res = response.json()
 232         self.assertNotIn("node", res["output"]["topology"], 'Topology should contain no node')
 233         self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
 234
 235     def test_05_disconnect_roadmb(self):
 236         response = test_utils.unmount_device("ROADM-B1")
 237         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 238         time.sleep(5)
 239
 240     def test_06_connect_xpdra(self):
 241         response = test_utils.mount_device("XPDR-A1", 'xpdra')
 242         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 243         time.sleep(10)
 244
 245     def test_07_check_tapi_topos(self):
 246         self.test_04_check_tapi_topos()
 247
 248     def test_08_connect_rdma(self):
 249         response = test_utils.mount_device("ROADM-A1", 'roadma')
 250         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 251         time.sleep(10)
 252
 253     def test_09_connect_rdmc(self):
 254         response = test_utils.mount_device("ROADM-C1", 'roadmc')
 255         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 256         time.sleep(10)
 257
 258     def test_10_check_tapi_topos(self):
 259         self.test_01_get_tapi_topology_T100G()
 260
 261         url = "{}/operations/tapi-topology:get-topology-details"
 262         data = {
 263             "tapi-topology:input": {
 264                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 265             }
 266         }
 267         response = test_utils.post_request(url, data)
 268         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 269         res = response.json()
 270         self.assertEqual(1, len(res["output"]["topology"]["node"]), 'Topology should contain 1 node')
 271         self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
 272         self.assertEqual("ROADM-infra", res["output"]["topology"]["node"][0]["name"][0]["value"],
 273                          'node name should be: ROADM-infra')
 274         self.assertIn("PHOTONIC_MEDIA", res["output"]["topology"]["node"][0]["layer-protocol-name"],
 275                          'Node layer protocol should contain PHOTONIC_MEDIA')
 276         self.assertEqual(1, len(res["output"]["topology"]["node"][0]["node-rule-group"]),
 277                          'node should contain 1 node rule group')
 278
 279     def test_11_connect_xprda_n1_to_roadma_pp1(self):
 280         response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "1",
 281                                                           "ROADM-A1", "1", "SRG1-PP1-TXRX")
 282         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 283         res = response.json()
 284         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
 285                       CREATED_SUCCESSFULLY)
 286         time.sleep(2)
 287
 288     def test_12_connect_roadma_pp1_to_xpdra_n1(self):
 289         response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "1",
 290                                                           "ROADM-A1", "1", "SRG1-PP1-TXRX")
 291         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 292         res = response.json()
 293         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
 294                       CREATED_SUCCESSFULLY)
 295         time.sleep(2)
 296
 297     def test_13_check_tapi_topology_T100G(self):
 298         url = "{}/operations/tapi-topology:get-topology-details"
 299         data = {
 300             "tapi-topology:input": {
 301                 "tapi-topology:topology-id-or-name": "Transponder 100GE"
 302             }
 303         }
 304         response = test_utils.post_request(url, data)
 305         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 306         res = response.json()
 307         self.assertEqual(1, len(res["output"]["topology"]["node"][0]["owned-node-edge-point"]),
 308                          'Node should contain 1 owned-node-edge-points')
 309         self.assertEqual("XPDR1-CLIENT1",
 310                          res["output"]["topology"]["node"][0]["owned-node-edge-point"][0]["name"][0]["value"],
 311                          'name of owned-node-edge-points should be XPDR1-CLIENT1')
 312
 313     def test_14_check_tapi_topology_T0(self):
 314         url = "{}/operations/tapi-topology:get-topology-details"
 315         data = {
 316             "tapi-topology:input": {
 317                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 318             }
 319         }
 320         response = test_utils.post_request(url, data)
 321         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 322         res = response.json()
 323         nodes = res["output"]["topology"]["node"]
 324         links = res["output"]["topology"]["link"]
 325         self.assertEqual(3, len(nodes), 'Topology should contain 3 nodes')
 326         self.assertEqual(2, len(links), 'Topology should contain 2 links')
 327         self.assertEqual(2, count_object_with_double_key(nodes, "name", "value-name", "otsi node name"),
 328                          'Topology should contain 2 otsi nodes')
 329         self.assertEqual(1, count_object_with_double_key(nodes, "name", "value-name", "dsr/odu node name"),
 330                          'Topology should contain 1 dsr node')
 331         self.assertEqual(1, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
 332                          'Topology should contain 1 transitional link')
 333         self.assertEqual(1, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
 334                          'Topology should contain 1 oms link')
 335
 336     def test_15_connect_xpdrc(self):
 337         response = test_utils.mount_device("XPDR-C1", 'xpdrc')
 338         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 339         time.sleep(10)
 340
 341     def test_16_connect_xprdc_n1_to_roadmc_pp1(self):
 342         response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "1",
 343                                                           "ROADM-C1", "1", "SRG1-PP1-TXRX")
 344         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 345         res = response.json()
 346         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
 347                       CREATED_SUCCESSFULLY)
 348         time.sleep(2)
 349
 350     def test_17_connect_roadmc_pp1_to_xpdrc_n1(self):
 351         response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "1",
 352                                                           "ROADM-C1", "1", "SRG1-PP1-TXRX")
 353         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 354         res = response.json()
 355         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
 356                       CREATED_SUCCESSFULLY)
 357         time.sleep(2)
 358
 359     def test_18_check_tapi_topology_T100G(self):
 360         url = "{}/operations/tapi-topology:get-topology-details"
 361         data = {
 362             "tapi-topology:input": {
 363                 "tapi-topology:topology-id-or-name": "Transponder 100GE"
 364             }
 365         }
 366         response = test_utils.post_request(url, data)
 367         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 368         res = response.json()
 369         self.assertEqual(2, len(res["output"]["topology"]["node"][0]["owned-node-edge-point"]),
 370                          'Node should contain 2 owned-node-edge-points')
 371         self.assertEqual("XPDR1-CLIENT1",
 372                          res["output"]["topology"]["node"][0]["owned-node-edge-point"][0]["name"][0]["value"],
 373                          'name of owned-node-edge-points should be XPDR1-CLIENT1')
 374         self.assertEqual("XPDR1-CLIENT1",
 375                          res["output"]["topology"]["node"][0]["owned-node-edge-point"][1]["name"][0]["value"],
 376                          'name of owned-node-edge-points should be XPDR1-CLIENT1')
 377
 378     def test_19_check_tapi_topology_T0(self):
 379         url = "{}/operations/tapi-topology:get-topology-details"
 380         data = {
 381             "tapi-topology:input": {
 382                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 383             }
 384         }
 385         response = test_utils.post_request(url, data)
 386         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 387         res = response.json()
 388         nodes = res["output"]["topology"]["node"]
 389         links = res["output"]["topology"]["link"]
 390         self.assertEqual(5, len(nodes), 'Topology should contain 5 nodes')
 391         self.assertEqual(4, len(links), 'Topology should contain 4 links')
 392         self.assertEqual(3, count_object_with_double_key(nodes, "name", "value-name", "otsi node name"),
 393                          'Topology should contain 3 otsi nodes')
 394         self.assertEqual(2, count_object_with_double_key(nodes, "name", "value-name", "dsr/odu node name"),
 395                          'Topology should contain 2 dsr nodes')
 396         self.assertEqual(2, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
 397                          'Topology should contain 2 transitional links')
 398         self.assertEqual(2, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
 399                          'Topology should contain 2 oms links')
 400
 401     def test_20_connect_spdr_sa1(self):
 402         response = test_utils.mount_device("SPDR-SA1", 'spdra')
 403         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 404         time.sleep(10)
 405         # TODO replace connect and disconnect timers with test_utils.wait_until_log_contains
 406
 407     def test_21_connect_spdr_sc1(self):
 408         response = test_utils.mount_device("SPDR-SC1", 'spdrc')
 409         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 410         time.sleep(10)
 411         # TODO replace connect and disconnect timers with test_utils.wait_until_log_contains
 412
 413     def test_22_check_tapi_topology_T100G(self):
 414         self.test_18_check_tapi_topology_T100G()
 415
 416     def test_23_check_tapi_topology_T0(self):
 417         self.test_19_check_tapi_topology_T0()
 418
 419     def test_24_connect_sprda_n1_to_roadma_pp2(self):
 420         response = test_utils.connect_xpdr_to_rdm_request("SPDR-SA1", "1", "1",
 421                                                           "ROADM-A1", "1", "SRG1-PP2-TXRX")
 422         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 423         res = response.json()
 424         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
 425                       CREATED_SUCCESSFULLY)
 426         time.sleep(2)
 427
 428     def test_25_connect_roadma_pp2_to_spdra_n1(self):
 429         response = test_utils.connect_rdm_to_xpdr_request("SPDR-SA1", "1", "1",
 430                                                           "ROADM-A1", "1", "SRG1-PP2-TXRX")
 431         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 432         res = response.json()
 433         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
 434                       CREATED_SUCCESSFULLY)
 435         time.sleep(2)
 436
 437     def test_26_connect_sprdc_n1_to_roadmc_pp2(self):
 438         response = test_utils.connect_xpdr_to_rdm_request("SPDR-SC1", "1", "1",
 439                                                           "ROADM-C1", "1", "SRG1-PP2-TXRX")
 440         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 441         res = response.json()
 442         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
 443                       CREATED_SUCCESSFULLY)
 444         time.sleep(2)
 445
 446     def test_27_connect_roadmc_pp2_to_spdrc_n1(self):
 447         response = test_utils.connect_rdm_to_xpdr_request("SPDR-SC1", "1", "1",
 448                                                           "ROADM-C1", "1", "SRG1-PP2-TXRX")
 449         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 450         res = response.json()
 451         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
 452                       CREATED_SUCCESSFULLY)
 453         time.sleep(2)
 454
 455     def test_28_check_tapi_topology_T100G(self):
 456         self.test_18_check_tapi_topology_T100G()
 457
 458     def test_29_check_tapi_topology_T0(self):
 459         url = "{}/operations/tapi-topology:get-topology-details"
 460         data = {
 461             "tapi-topology:input": {
 462                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 463             }
 464         }
 465         response = test_utils.post_request(url, data)
 466         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 467         res = response.json()
 468         nodes = res["output"]["topology"]["node"]
 469         links = res["output"]["topology"]["link"]
 470         self.assertEqual(9, len(nodes), 'Topology should contain 9 nodes')
 471         self.assertEqual(8, len(links), 'Topology should contain 8 links')
 472         self.assertEqual(5, count_object_with_double_key(nodes, "name", "value-name", "otsi node name"),
 473                          'Topology should contain 5 otsi nodes')
 474         self.assertEqual(4, count_object_with_double_key(nodes, "name", "value-name", "dsr/odu node name"),
 475                          'Topology should contain 4 dsr nodes')
 476         self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
 477                          'Topology should contain 4 transitional links')
 478         self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
 479                          'Topology should contain 4 oms links')
 480
 481     def test_30_add_oms_attributes(self):
 482         # Config ROADMA-ROADMC oms-attributes
 483         data = {"span": {
 484             "auto-spanloss": "true",
 485             "spanloss-base": 11.4,
 486             "spanloss-current": 12,
 487             "engineered-spanloss": 12.2,
 488             "link-concatenation": [{
 489                 "SRLG-Id": 0,
 490                 "fiber-type": "smf",
 491                 "SRLG-length": 100000,
 492                 "pmd": 0.5}]}}
 493         response = test_utils.add_oms_attr_request("ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", data)
 494         self.assertEqual(response.status_code, requests.codes.created)
 495         # Config ROADMC-ROADMA oms-attributes
 496         data = {"span": {
 497             "auto-spanloss": "true",
 498             "spanloss-base": 11.4,
 499             "spanloss-current": 12,
 500             "engineered-spanloss": 12.2,
 501             "link-concatenation": [{
 502                 "SRLG-Id": 0,
 503                 "fiber-type": "smf",
 504                 "SRLG-length": 100000,
 505                 "pmd": 0.5}]}}
 506         response = test_utils.add_oms_attr_request("ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX", data)
 507         self.assertEqual(response.status_code, requests.codes.created)
 508
 509     def test_31_create_OCH_OTU4_service(self):
 510         response = test_utils.service_create_request(self.cr_serv_sample_data)
 511         self.assertEqual(response.status_code, requests.codes.ok)
 512         res = response.json()
 513         self.assertIn('PCE calculation in progress',
 514                       res['output']['configuration-response-common']['response-message'])
 515         time.sleep(self.WAITING)
 516
 517     def test_32_check_tapi_topology_T0(self):
 518         url = "{}/operations/tapi-topology:get-topology-details"
 519         data = {
 520             "tapi-topology:input": {
 521                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 522             }
 523         }
 524         response = test_utils.post_request(url, data)
 525         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 526         res = response.json()
 527         nodes = res["output"]["topology"]["node"]
 528         links = res["output"]["topology"]["link"]
 529         self.assertEqual(9, len(nodes), 'Topology should contain 9 nodes')
 530         self.assertEqual(9, len(links), 'Topology should contain 9 links')
 531         self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
 532                          'Topology should contain 4 transitional links')
 533         self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
 534                          'Topology should contain 4 oms links')
 535         self.assertEqual(1, count_object_with_double_key(links, "name", "value-name", "otn link name"),
 536                          'Topology should contain 1 otn link')
 537         for link in links:
 538             if link["name"][0]["value"] == "OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1":
 539                 self.assertEqual(100000, link["available-capacity"]["total-size"]["value"],
 540                          'OTU4 link should have an available capacity of 100 000 Mbps')
 541             elif link["name"][0]["value-name"] == "transitional link name":
 542                 self.assertEqual(100, link["available-capacity"]["total-size"]["value"],
 543                          'link should have an available capacity of 100 Gbps')
 544             self.assertEqual(2, len(link["node-edge-point"]), 'link should have 2 neps')
 545
 546     def test_33_create_ODU4_service(self):
 547         self.cr_serv_sample_data["input"]["service-name"] = "service1-ODU4"
 548         self.cr_serv_sample_data["input"]["service-a-end"]["service-format"] = "ODU"
 549         del self.cr_serv_sample_data["input"]["service-a-end"]["otu-service-rate"]
 550         self.cr_serv_sample_data["input"]["service-a-end"]["odu-service-rate"] = "org-openroadm-otn-common-types:ODU4"
 551         self.cr_serv_sample_data["input"]["service-z-end"]["service-format"] = "ODU"
 552         del self.cr_serv_sample_data["input"]["service-z-end"]["otu-service-rate"]
 553         self.cr_serv_sample_data["input"]["service-z-end"]["odu-service-rate"] = "org-openroadm-otn-common-types:ODU4"
 554
 555         response = test_utils.service_create_request(self.cr_serv_sample_data)
 556         self.assertEqual(response.status_code, requests.codes.ok)
 557         res = response.json()
 558         self.assertIn('PCE calculation in progress',
 559                       res['output']['configuration-response-common']['response-message'])
 560         time.sleep(self.WAITING)
 561
 562     def test_34_check_tapi_topology_T0(self):
 563         url = "{}/operations/tapi-topology:get-topology-details"
 564         data = {
 565             "tapi-topology:input": {
 566                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 567             }
 568         }
 569         response = test_utils.post_request(url, data)
 570         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 571         res = response.json()
 572         nodes = res["output"]["topology"]["node"]
 573         links = res["output"]["topology"]["link"]
 574         self.assertEqual(9, len(nodes), 'Topology should contain 9 nodes')
 575         self.assertEqual(10, len(links), 'Topology should contain 10 links')
 576         self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
 577                          'Topology should contain 4 transitional links')
 578         self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
 579                          'Topology should contain 4 oms links')
 580         self.assertEqual(2, count_object_with_double_key(links, "name", "value-name", "otn link name"),
 581                          'Topology should contain 2 otn links')
 582         for link in links:
 583             if link["name"][0]["value"] == "OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1":
 584                 self.assertEqual(0, link["available-capacity"]["total-size"]["value"],
 585                          'OTU4 link should have an available capacity of 0 Mbps')
 586             elif link["name"][0]["value"] == "ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1":
 587                 self.assertEqual(100000, link["available-capacity"]["total-size"]["value"],
 588                          'ODU4 link should have an available capacity of 100 000 Mbps')
 589             elif link["name"][0]["value-name"] == "transitional link name":
 590                 self.assertEqual(100, link["available-capacity"]["total-size"]["value"],
 591                          'link should have an available capacity of 100 Gbps')
 592             self.assertEqual(2, len(link["node-edge-point"]), 'link should have 2 neps')
 593
 594     def test_35_connect_sprda_2_n2_to_roadma_pp3(self):
 595         response = test_utils.connect_xpdr_to_rdm_request("SPDR-SA1", "2", "2",
 596                                                           "ROADM-A1", "1", "SRG1-PP3-TXRX")
 597         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 598         res = response.json()
 599         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
 600                       CREATED_SUCCESSFULLY)
 601         time.sleep(2)
 602
 603     def test_36_connect_roadma_pp3_to_spdra_2_n2(self):
 604         response = test_utils.connect_rdm_to_xpdr_request("SPDR-SA1", "2", "2",
 605                                                           "ROADM-A1", "1", "SRG1-PP3-TXRX")
 606         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 607         res = response.json()
 608         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
 609                       CREATED_SUCCESSFULLY)
 610         time.sleep(2)
 611
 612     def test_37_check_tapi_topology_T0(self):
 613         url = "{}/operations/tapi-topology:get-topology-details"
 614         data = {
 615             "tapi-topology:input": {
 616                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 617             }
 618         }
 619         response = test_utils.post_request(url, data)
 620         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 621         res = response.json()
 622         nodes = res["output"]["topology"]["node"]
 623         links = res["output"]["topology"]["link"]
 624         self.assertEqual(11, len(nodes), 'Topology should contain 11 nodes')
 625         self.assertEqual(12, len(links), 'Topology should contain 12 links')
 626         self.assertEqual(6, count_object_with_double_key(nodes, "name", "value-name", "otsi node name"),
 627                          'Topology should contain 6 otsi nodes')
 628         self.assertEqual(5, count_object_with_double_key(nodes, "name", "value-name", "dsr/odu node name"),
 629                          'Topology should contain 5 dsr nodes')
 630         self.assertEqual(5, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
 631                          'Topology should contain 5 transitional links')
 632         self.assertEqual(5, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
 633                          'Topology should contain 5 oms links')
 634         self.assertEqual(2, count_object_with_double_key(links, "name", "value-name", "otn link name"),
 635                          'Topology should contain 2 otn links')
 636
 637     def test_38_delete_ODU4_service(self):
 638         response = test_utils.service_delete_request("service1-ODU4")
 639         self.assertEqual(response.status_code, requests.codes.ok)
 640         res = response.json()
 641         self.assertIn('Renderer service delete in progress',
 642                       res['output']['configuration-response-common']['response-message'])
 643         time.sleep(20)
 644
 645     def test_39_delete_OCH_OTU4_service(self):
 646         response = test_utils.service_delete_request("service1-OCH-OTU4")
 647         self.assertEqual(response.status_code, requests.codes.ok)
 648         res = response.json()
 649         self.assertIn('Renderer service delete in progress',
 650                       res['output']['configuration-response-common']['response-message'])
 651         time.sleep(20)
 652
 653     def test_40_check_tapi_topology_T0(self):
 654         url = "{}/operations/tapi-topology:get-topology-details"
 655         data = {
 656             "tapi-topology:input": {
 657                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 658             }
 659         }
 660         response = test_utils.post_request(url, data)
 661         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 662         res = response.json()
 663         nodes = res["output"]["topology"]["node"]
 664         links = res["output"]["topology"]["link"]
 665         self.assertEqual(11, len(nodes), 'Topology should contain 11 nodes')
 666         self.assertEqual(10, len(links), 'Topology should contain 10 links')
 667         self.assertEqual(0, count_object_with_double_key(links, "name", "value-name", "otn link name"),
 668                          'Topology should contain 0 otn link')
 669
 670     def test_41_disconnect_xponders_from_roadm(self):
 671         url = "{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:link/"
 672         response = test_utils.get_ordm_topo_request("")
 673         self.assertEqual(response.status_code, requests.codes.ok)
 674         res = response.json()
 675         links = res['network'][0]['ietf-network-topology:link']
 676         for link in links:
 677             if (link["org-openroadm-common-network:link-type"] == "XPONDER-OUTPUT" or
 678             link["org-openroadm-common-network:link-type"] == "XPONDER-INPUT"):
 679                  link_name = link["link-id"]
 680                  response = test_utils.delete_request(url+link_name)
 681                  self.assertEqual(response.status_code, requests.codes.ok)
 682
 683     def test_42_check_tapi_topology_T0(self):
 684         url = "{}/operations/tapi-topology:get-topology-details"
 685         data = {
 686             "tapi-topology:input": {
 687                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 688             }
 689         }
 690         response = test_utils.post_request(url, data)
 691         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 692         res = response.json()
 693         nodes = res["output"]["topology"]["node"]
 694         self.assertEqual(1, len(nodes), 'Topology should contain 1 node')
 695         self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
 696         self.assertEqual("ROADM-infra", res["output"]["topology"]["node"][0]["name"][0]["value"],
 697                          'node name should be: ROADM-infra')
 698
 699     def test_43_get_tapi_topology_T100G(self):
 700         url = "{}/operations/tapi-topology:get-topology-details"
 701         data = {
 702             "tapi-topology:input": {
 703                 "tapi-topology:topology-id-or-name": "Transponder 100GE"
 704             }
 705         }
 706         response = test_utils.post_request(url, data)
 707         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 708         res = response.json()
 709         self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'Topology should contain 1 node')
 710         self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
 711         self.assertNotIn("owned-node-edge-point", res["output"]["topology"]["node"][0],
 712                          'Node should contain no owned-node-edge-points')
 713
 714     def test_44_disconnect_roadma(self):
 715         response = test_utils.unmount_device("ROADM-A1")
 716         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 717         time.sleep(5)
 718
 719     def test_45_disconnect_roadmc(self):
 720         response = test_utils.unmount_device("ROADM-C1")
 721         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 722         time.sleep(5)
 723
 724     def test_46_check_tapi_topos(self):
 725         self.test_01_get_tapi_topology_T100G()
 726         self.test_02_get_tapi_topology_T0()
 727
 728     def test_47_disconnect_xpdra(self):
 729         response = test_utils.unmount_device("XPDR-A1")
 730         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 731         time.sleep(5)
 732
 733     def test_48_disconnect_xpdrc(self):
 734         response = test_utils.unmount_device("XPDR-C1")
 735         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 736         time.sleep(5)
 737
 738     def test_49_disconnect_spdr_sa1(self):
 739         response = test_utils.unmount_device("SPDR-SA1")
 740         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 741         time.sleep(5)
 742
 743     def test_50_disconnect_spdr_sc1(self):
 744         response = test_utils.unmount_device("SPDR-SC1")
 745         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 746
 747
 748 def find_object_with_key(list_dicts, key, value):
 749     for dict_ in list_dicts:
 750         if dict_[key] == value:
 751             return dict_
 752     return None
 753
 754 def count_object_with_double_key(list_dicts, key1, key2, value):
 755     nb = 0
 756     for dict in list_dicts:
 757         if dict[key1][0][key2] == value:
 758             nb += 1
 759     return nb
 760
 761
 762 if __name__ == "__main__":
 763     unittest.main(verbosity=2)