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 import requests
  22 sys.path.append('transportpce_tests/common/')
  23 import test_utils
  24
  25
  26 CREATED_SUCCESSFULLY = 'Result message should contain Xponder Roadm Link created successfully'
  27
  28 class TransportTapitesting(unittest.TestCase):
  29
  30     processes = None
  31     WAITING = 20
  32     NODE_VERSION = '2.2.1'
  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', cls.NODE_VERSION),
 158                                                ('roadma', cls.NODE_VERSION),
 159                                                ('roadmb', cls.NODE_VERSION),
 160                                                ('roadmc', cls.NODE_VERSION),
 161                                                ('xpdrc', cls.NODE_VERSION),
 162                                                ('spdra', cls.NODE_VERSION),
 163                                                ('spdrc', cls.NODE_VERSION)])
 164
 165     @classmethod
 166     def tearDownClass(cls):
 167         # pylint: disable=not-an-iterable
 168         for process in cls.processes:
 169             test_utils.shutdown_process(process)
 170         print("all processes killed")
 171
 172     def setUp(self):  # instruction executed before each test method
 173         if self.init_failed:
 174             self.fail('Feature installation failed')
 175         print("execution of {}".format(self.id().split(".")[-1]))
 176
 177     def test_01_get_tapi_topology_T100G(self):
 178         url = "{}/operations/tapi-topology:get-topology-details"
 179         data = {
 180             "tapi-topology:input": {
 181                 "tapi-topology:topology-id-or-name": "Transponder 100GE"
 182             }
 183         }
 184         response = test_utils.post_request(url, data)
 185         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 186         res = response.json()
 187         self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'Topology should contain 1 node')
 188         self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
 189         self.assertNotIn("owned-node-edge-point", res["output"]["topology"]["node"][0],
 190                          'Node should contain no owned-node-edge-points')
 191         self.assertEqual("Tpdr100g over WDM node", res["output"]["topology"]["node"][0]["name"][0]["value"],
 192                          'node name should be: Tpdr100g over WDM node')
 193         self.assertIn("ETH", res["output"]["topology"]["node"][0]["layer-protocol-name"],
 194                          'Node layer protocol should contain ETH')
 195         self.assertEqual(1, len(res["output"]["topology"]["node"][0]["node-rule-group"]),
 196                          'node should contain 1 node rule group')
 197
 198     def test_02_get_tapi_topology_T0(self):
 199         url = "{}/operations/tapi-topology:get-topology-details"
 200         data = {
 201             "tapi-topology:input": {
 202                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 203             }
 204         }
 205         response = test_utils.post_request(url, data)
 206         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 207         res = response.json()
 208         self.assertNotIn("node", res["output"]["topology"], 'Topology should contain no node')
 209         self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
 210
 211     def test_03_connect_rdmb(self):
 212         response = test_utils.mount_device("ROADM-B1", ('roadmb', self.NODE_VERSION))
 213         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 214         time.sleep(10)
 215
 216     def test_04_check_tapi_topos(self):
 217         url = "{}/operations/tapi-topology:get-topology-details"
 218         data = {
 219             "tapi-topology:input": {
 220                 "tapi-topology:topology-id-or-name": "Transponder 100GE"
 221             }
 222         }
 223         response = test_utils.post_request(url, data)
 224         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 225         res = response.json()
 226         self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'Topology should contain 1 node')
 227         self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
 228
 229         url = "{}/operations/tapi-topology:get-topology-details"
 230         data = {
 231             "tapi-topology:input": {
 232                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 233             }
 234         }
 235         response = test_utils.post_request(url, data)
 236         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 237         res = response.json()
 238         self.assertNotIn("node", res["output"]["topology"], 'Topology should contain no node')
 239         self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
 240
 241     def test_05_disconnect_roadmb(self):
 242         response = test_utils.unmount_device("ROADM-B1")
 243         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 244         time.sleep(5)
 245
 246     def test_06_connect_xpdra(self):
 247         response = test_utils.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION))
 248         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 249         time.sleep(10)
 250
 251     def test_07_check_tapi_topos(self):
 252         self.test_04_check_tapi_topos()
 253
 254     def test_08_connect_rdma(self):
 255         response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION))
 256         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 257         time.sleep(10)
 258
 259     def test_09_connect_rdmc(self):
 260         response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION))
 261         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 262         time.sleep(10)
 263
 264     def test_10_check_tapi_topos(self):
 265         self.test_01_get_tapi_topology_T100G()
 266
 267         url = "{}/operations/tapi-topology:get-topology-details"
 268         data = {
 269             "tapi-topology:input": {
 270                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 271             }
 272         }
 273         response = test_utils.post_request(url, data)
 274         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 275         res = response.json()
 276         self.assertEqual(1, len(res["output"]["topology"]["node"]), 'Topology should contain 1 node')
 277         self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
 278         self.assertEqual("ROADM-infra", res["output"]["topology"]["node"][0]["name"][0]["value"],
 279                          'node name should be: ROADM-infra')
 280         self.assertIn("PHOTONIC_MEDIA", res["output"]["topology"]["node"][0]["layer-protocol-name"],
 281                          'Node layer protocol should contain PHOTONIC_MEDIA')
 282         self.assertEqual(1, len(res["output"]["topology"]["node"][0]["node-rule-group"]),
 283                          'node should contain 1 node rule group')
 284
 285     def test_11_connect_xprda_n1_to_roadma_pp1(self):
 286         response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "1",
 287                                                           "ROADM-A1", "1", "SRG1-PP1-TXRX")
 288         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 289         res = response.json()
 290         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
 291                       CREATED_SUCCESSFULLY)
 292         time.sleep(2)
 293
 294     def test_12_connect_roadma_pp1_to_xpdra_n1(self):
 295         response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "1",
 296                                                           "ROADM-A1", "1", "SRG1-PP1-TXRX")
 297         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 298         res = response.json()
 299         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
 300                       CREATED_SUCCESSFULLY)
 301         time.sleep(2)
 302
 303     def test_13_check_tapi_topology_T100G(self):
 304         url = "{}/operations/tapi-topology:get-topology-details"
 305         data = {
 306             "tapi-topology:input": {
 307                 "tapi-topology:topology-id-or-name": "Transponder 100GE"
 308             }
 309         }
 310         response = test_utils.post_request(url, data)
 311         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 312         res = response.json()
 313         self.assertEqual(1, len(res["output"]["topology"]["node"][0]["owned-node-edge-point"]),
 314                          'Node should contain 1 owned-node-edge-points')
 315         self.assertEqual("XPDR1-CLIENT1",
 316                          res["output"]["topology"]["node"][0]["owned-node-edge-point"][0]["name"][0]["value"],
 317                          'name of owned-node-edge-points should be XPDR1-CLIENT1')
 318
 319     def test_14_check_tapi_topology_T0(self):
 320         url = "{}/operations/tapi-topology:get-topology-details"
 321         data = {
 322             "tapi-topology:input": {
 323                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 324             }
 325         }
 326         response = test_utils.post_request(url, data)
 327         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 328         res = response.json()
 329         nodes = res["output"]["topology"]["node"]
 330         links = res["output"]["topology"]["link"]
 331         self.assertEqual(3, len(nodes), 'Topology should contain 3 nodes')
 332         self.assertEqual(2, len(links), 'Topology should contain 2 links')
 333         self.assertEqual(2, count_object_with_double_key(nodes, "name", "value-name", "otsi node name"),
 334                          'Topology should contain 2 otsi nodes')
 335         self.assertEqual(1, count_object_with_double_key(nodes, "name", "value-name", "dsr/odu node name"),
 336                          'Topology should contain 1 dsr node')
 337         self.assertEqual(1, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
 338                          'Topology should contain 1 transitional link')
 339         self.assertEqual(1, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
 340                          'Topology should contain 1 oms link')
 341
 342     def test_15_connect_xpdrc(self):
 343         response = test_utils.mount_device("XPDR-C1", ('xpdrc', self.NODE_VERSION))
 344         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 345         time.sleep(10)
 346
 347     def test_16_connect_xprdc_n1_to_roadmc_pp1(self):
 348         response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "1",
 349                                                           "ROADM-C1", "1", "SRG1-PP1-TXRX")
 350         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 351         res = response.json()
 352         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
 353                       CREATED_SUCCESSFULLY)
 354         time.sleep(2)
 355
 356     def test_17_connect_roadmc_pp1_to_xpdrc_n1(self):
 357         response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "1",
 358                                                           "ROADM-C1", "1", "SRG1-PP1-TXRX")
 359         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 360         res = response.json()
 361         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
 362                       CREATED_SUCCESSFULLY)
 363         time.sleep(2)
 364
 365     def test_18_check_tapi_topology_T100G(self):
 366         url = "{}/operations/tapi-topology:get-topology-details"
 367         data = {
 368             "tapi-topology:input": {
 369                 "tapi-topology:topology-id-or-name": "Transponder 100GE"
 370             }
 371         }
 372         response = test_utils.post_request(url, data)
 373         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 374         res = response.json()
 375         self.assertEqual(2, len(res["output"]["topology"]["node"][0]["owned-node-edge-point"]),
 376                          'Node should contain 2 owned-node-edge-points')
 377         self.assertEqual("XPDR1-CLIENT1",
 378                          res["output"]["topology"]["node"][0]["owned-node-edge-point"][0]["name"][0]["value"],
 379                          'name of owned-node-edge-points should be XPDR1-CLIENT1')
 380         self.assertEqual("XPDR1-CLIENT1",
 381                          res["output"]["topology"]["node"][0]["owned-node-edge-point"][1]["name"][0]["value"],
 382                          'name of owned-node-edge-points should be XPDR1-CLIENT1')
 383
 384     def test_19_check_tapi_topology_T0(self):
 385         url = "{}/operations/tapi-topology:get-topology-details"
 386         data = {
 387             "tapi-topology:input": {
 388                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 389             }
 390         }
 391         response = test_utils.post_request(url, data)
 392         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 393         res = response.json()
 394         nodes = res["output"]["topology"]["node"]
 395         links = res["output"]["topology"]["link"]
 396         self.assertEqual(5, len(nodes), 'Topology should contain 5 nodes')
 397         self.assertEqual(4, len(links), 'Topology should contain 4 links')
 398         self.assertEqual(3, count_object_with_double_key(nodes, "name", "value-name", "otsi node name"),
 399                          'Topology should contain 3 otsi nodes')
 400         self.assertEqual(2, count_object_with_double_key(nodes, "name", "value-name", "dsr/odu node name"),
 401                          'Topology should contain 2 dsr nodes')
 402         self.assertEqual(2, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
 403                          'Topology should contain 2 transitional links')
 404         self.assertEqual(2, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
 405                          'Topology should contain 2 oms links')
 406
 407     def test_20_connect_spdr_sa1(self):
 408         response = test_utils.mount_device("SPDR-SA1", ('spdra', self.NODE_VERSION))
 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_21_connect_spdr_sc1(self):
 414         response = test_utils.mount_device("SPDR-SC1", ('spdrc', self.NODE_VERSION))
 415         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 416         time.sleep(10)
 417         # TODO replace connect and disconnect timers with test_utils.wait_until_log_contains
 418
 419     def test_22_check_tapi_topology_T100G(self):
 420         self.test_18_check_tapi_topology_T100G()
 421
 422     def test_23_check_tapi_topology_T0(self):
 423         self.test_19_check_tapi_topology_T0()
 424
 425     def test_24_connect_sprda_n1_to_roadma_pp2(self):
 426         response = test_utils.connect_xpdr_to_rdm_request("SPDR-SA1", "1", "1",
 427                                                           "ROADM-A1", "1", "SRG1-PP2-TXRX")
 428         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 429         res = response.json()
 430         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
 431                       CREATED_SUCCESSFULLY)
 432         time.sleep(2)
 433
 434     def test_25_connect_roadma_pp2_to_spdra_n1(self):
 435         response = test_utils.connect_rdm_to_xpdr_request("SPDR-SA1", "1", "1",
 436                                                           "ROADM-A1", "1", "SRG1-PP2-TXRX")
 437         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 438         res = response.json()
 439         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
 440                       CREATED_SUCCESSFULLY)
 441         time.sleep(2)
 442
 443     def test_26_connect_sprdc_n1_to_roadmc_pp2(self):
 444         response = test_utils.connect_xpdr_to_rdm_request("SPDR-SC1", "1", "1",
 445                                                           "ROADM-C1", "1", "SRG1-PP2-TXRX")
 446         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 447         res = response.json()
 448         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
 449                       CREATED_SUCCESSFULLY)
 450         time.sleep(2)
 451
 452     def test_27_connect_roadmc_pp2_to_spdrc_n1(self):
 453         response = test_utils.connect_rdm_to_xpdr_request("SPDR-SC1", "1", "1",
 454                                                           "ROADM-C1", "1", "SRG1-PP2-TXRX")
 455         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 456         res = response.json()
 457         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
 458                       CREATED_SUCCESSFULLY)
 459         time.sleep(2)
 460
 461     def test_28_check_tapi_topology_T100G(self):
 462         self.test_18_check_tapi_topology_T100G()
 463
 464     def test_29_check_tapi_topology_T0(self):
 465         url = "{}/operations/tapi-topology:get-topology-details"
 466         data = {
 467             "tapi-topology:input": {
 468                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 469             }
 470         }
 471         response = test_utils.post_request(url, data)
 472         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 473         res = response.json()
 474         nodes = res["output"]["topology"]["node"]
 475         links = res["output"]["topology"]["link"]
 476         self.assertEqual(9, len(nodes), 'Topology should contain 9 nodes')
 477         self.assertEqual(8, len(links), 'Topology should contain 8 links')
 478         self.assertEqual(5, count_object_with_double_key(nodes, "name", "value-name", "otsi node name"),
 479                          'Topology should contain 5 otsi nodes')
 480         self.assertEqual(4, count_object_with_double_key(nodes, "name", "value-name", "dsr/odu node name"),
 481                          'Topology should contain 4 dsr nodes')
 482         self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
 483                          'Topology should contain 4 transitional links')
 484         self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
 485                          'Topology should contain 4 oms links')
 486
 487     def test_30_add_oms_attributes(self):
 488         # Config ROADMA-ROADMC oms-attributes
 489         data = {"span": {
 490             "auto-spanloss": "true",
 491             "spanloss-base": 11.4,
 492             "spanloss-current": 12,
 493             "engineered-spanloss": 12.2,
 494             "link-concatenation": [{
 495                 "SRLG-Id": 0,
 496                 "fiber-type": "smf",
 497                 "SRLG-length": 100000,
 498                 "pmd": 0.5}]}}
 499         response = test_utils.add_oms_attr_request("ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", data)
 500         self.assertEqual(response.status_code, requests.codes.created)
 501         # Config ROADMC-ROADMA oms-attributes
 502         data = {"span": {
 503             "auto-spanloss": "true",
 504             "spanloss-base": 11.4,
 505             "spanloss-current": 12,
 506             "engineered-spanloss": 12.2,
 507             "link-concatenation": [{
 508                 "SRLG-Id": 0,
 509                 "fiber-type": "smf",
 510                 "SRLG-length": 100000,
 511                 "pmd": 0.5}]}}
 512         response = test_utils.add_oms_attr_request("ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX", data)
 513         self.assertEqual(response.status_code, requests.codes.created)
 514
 515     def test_31_create_OCH_OTU4_service(self):
 516         response = test_utils.service_create_request(self.cr_serv_sample_data)
 517         self.assertEqual(response.status_code, requests.codes.ok)
 518         res = response.json()
 519         self.assertIn('PCE calculation in progress',
 520                       res['output']['configuration-response-common']['response-message'])
 521         time.sleep(self.WAITING)
 522
 523     def test_32_check_tapi_topology_T0(self):
 524         url = "{}/operations/tapi-topology:get-topology-details"
 525         data = {
 526             "tapi-topology:input": {
 527                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 528             }
 529         }
 530         response = test_utils.post_request(url, data)
 531         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 532         res = response.json()
 533         nodes = res["output"]["topology"]["node"]
 534         links = res["output"]["topology"]["link"]
 535         self.assertEqual(9, len(nodes), 'Topology should contain 9 nodes')
 536         self.assertEqual(9, len(links), 'Topology should contain 9 links')
 537         self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
 538                          'Topology should contain 4 transitional links')
 539         self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
 540                          'Topology should contain 4 oms links')
 541         self.assertEqual(1, count_object_with_double_key(links, "name", "value-name", "otn link name"),
 542                          'Topology should contain 1 otn link')
 543         for link in links:
 544             if link["name"][0]["value"] == "OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1":
 545                 self.assertEqual(100000, link["available-capacity"]["total-size"]["value"],
 546                          'OTU4 link should have an available capacity of 100 000 Mbps')
 547             elif link["name"][0]["value-name"] == "transitional link name":
 548                 self.assertEqual(100, link["available-capacity"]["total-size"]["value"],
 549                          'link should have an available capacity of 100 Gbps')
 550             self.assertEqual(2, len(link["node-edge-point"]), 'link should have 2 neps')
 551
 552     def test_33_create_ODU4_service(self):
 553         self.cr_serv_sample_data["input"]["service-name"] = "service1-ODU4"
 554         self.cr_serv_sample_data["input"]["service-a-end"]["service-format"] = "ODU"
 555         del self.cr_serv_sample_data["input"]["service-a-end"]["otu-service-rate"]
 556         self.cr_serv_sample_data["input"]["service-a-end"]["odu-service-rate"] = "org-openroadm-otn-common-types:ODU4"
 557         self.cr_serv_sample_data["input"]["service-z-end"]["service-format"] = "ODU"
 558         del self.cr_serv_sample_data["input"]["service-z-end"]["otu-service-rate"]
 559         self.cr_serv_sample_data["input"]["service-z-end"]["odu-service-rate"] = "org-openroadm-otn-common-types:ODU4"
 560
 561         response = test_utils.service_create_request(self.cr_serv_sample_data)
 562         self.assertEqual(response.status_code, requests.codes.ok)
 563         res = response.json()
 564         self.assertIn('PCE calculation in progress',
 565                       res['output']['configuration-response-common']['response-message'])
 566         time.sleep(self.WAITING)
 567
 568     def test_34_check_tapi_topology_T0(self):
 569         url = "{}/operations/tapi-topology:get-topology-details"
 570         data = {
 571             "tapi-topology:input": {
 572                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 573             }
 574         }
 575         response = test_utils.post_request(url, data)
 576         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 577         res = response.json()
 578         nodes = res["output"]["topology"]["node"]
 579         links = res["output"]["topology"]["link"]
 580         self.assertEqual(9, len(nodes), 'Topology should contain 9 nodes')
 581         self.assertEqual(10, len(links), 'Topology should contain 10 links')
 582         self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
 583                          'Topology should contain 4 transitional links')
 584         self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
 585                          'Topology should contain 4 oms links')
 586         self.assertEqual(2, count_object_with_double_key(links, "name", "value-name", "otn link name"),
 587                          'Topology should contain 2 otn links')
 588         for link in links:
 589             if link["name"][0]["value"] == "OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1":
 590                 self.assertEqual(0, link["available-capacity"]["total-size"]["value"],
 591                          'OTU4 link should have an available capacity of 0 Mbps')
 592             elif link["name"][0]["value"] == "ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1":
 593                 self.assertEqual(100000, link["available-capacity"]["total-size"]["value"],
 594                          'ODU4 link should have an available capacity of 100 000 Mbps')
 595             elif link["name"][0]["value-name"] == "transitional link name":
 596                 self.assertEqual(100, link["available-capacity"]["total-size"]["value"],
 597                          'link should have an available capacity of 100 Gbps')
 598             self.assertEqual(2, len(link["node-edge-point"]), 'link should have 2 neps')
 599
 600     def test_35_connect_sprda_2_n2_to_roadma_pp3(self):
 601         response = test_utils.connect_xpdr_to_rdm_request("SPDR-SA1", "2", "2",
 602                                                           "ROADM-A1", "1", "SRG1-PP3-TXRX")
 603         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 604         res = response.json()
 605         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
 606                       CREATED_SUCCESSFULLY)
 607         time.sleep(2)
 608
 609     def test_36_connect_roadma_pp3_to_spdra_2_n2(self):
 610         response = test_utils.connect_rdm_to_xpdr_request("SPDR-SA1", "2", "2",
 611                                                           "ROADM-A1", "1", "SRG1-PP3-TXRX")
 612         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 613         res = response.json()
 614         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
 615                       CREATED_SUCCESSFULLY)
 616         time.sleep(2)
 617
 618     def test_37_check_tapi_topology_T0(self):
 619         url = "{}/operations/tapi-topology:get-topology-details"
 620         data = {
 621             "tapi-topology:input": {
 622                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 623             }
 624         }
 625         response = test_utils.post_request(url, data)
 626         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 627         res = response.json()
 628         nodes = res["output"]["topology"]["node"]
 629         links = res["output"]["topology"]["link"]
 630         self.assertEqual(11, len(nodes), 'Topology should contain 11 nodes')
 631         self.assertEqual(12, len(links), 'Topology should contain 12 links')
 632         self.assertEqual(6, count_object_with_double_key(nodes, "name", "value-name", "otsi node name"),
 633                          'Topology should contain 6 otsi nodes')
 634         self.assertEqual(5, count_object_with_double_key(nodes, "name", "value-name", "dsr/odu node name"),
 635                          'Topology should contain 5 dsr nodes')
 636         self.assertEqual(5, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
 637                          'Topology should contain 5 transitional links')
 638         self.assertEqual(5, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
 639                          'Topology should contain 5 oms links')
 640         self.assertEqual(2, count_object_with_double_key(links, "name", "value-name", "otn link name"),
 641                          'Topology should contain 2 otn links')
 642
 643     def test_38_delete_ODU4_service(self):
 644         response = test_utils.service_delete_request("service1-ODU4")
 645         self.assertEqual(response.status_code, requests.codes.ok)
 646         res = response.json()
 647         self.assertIn('Renderer service delete in progress',
 648                       res['output']['configuration-response-common']['response-message'])
 649         time.sleep(20)
 650
 651     def test_39_delete_OCH_OTU4_service(self):
 652         response = test_utils.service_delete_request("service1-OCH-OTU4")
 653         self.assertEqual(response.status_code, requests.codes.ok)
 654         res = response.json()
 655         self.assertIn('Renderer service delete in progress',
 656                       res['output']['configuration-response-common']['response-message'])
 657         time.sleep(20)
 658
 659     def test_40_check_tapi_topology_T0(self):
 660         url = "{}/operations/tapi-topology:get-topology-details"
 661         data = {
 662             "tapi-topology:input": {
 663                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 664             }
 665         }
 666         response = test_utils.post_request(url, data)
 667         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 668         res = response.json()
 669         nodes = res["output"]["topology"]["node"]
 670         links = res["output"]["topology"]["link"]
 671         self.assertEqual(11, len(nodes), 'Topology should contain 11 nodes')
 672         self.assertEqual(10, len(links), 'Topology should contain 10 links')
 673         self.assertEqual(0, count_object_with_double_key(links, "name", "value-name", "otn link name"),
 674                          'Topology should contain 0 otn link')
 675
 676     def test_41_disconnect_xponders_from_roadm(self):
 677         url = "{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:link/"
 678         response = test_utils.get_ordm_topo_request("")
 679         self.assertEqual(response.status_code, requests.codes.ok)
 680         res = response.json()
 681         links = res['network'][0]['ietf-network-topology:link']
 682         for link in links:
 683             if (link["org-openroadm-common-network:link-type"] == "XPONDER-OUTPUT" or
 684             link["org-openroadm-common-network:link-type"] == "XPONDER-INPUT"):
 685                  link_name = link["link-id"]
 686                  response = test_utils.delete_request(url+link_name)
 687                  self.assertEqual(response.status_code, requests.codes.ok)
 688
 689     def test_42_check_tapi_topology_T0(self):
 690         url = "{}/operations/tapi-topology:get-topology-details"
 691         data = {
 692             "tapi-topology:input": {
 693                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 694             }
 695         }
 696         response = test_utils.post_request(url, data)
 697         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 698         res = response.json()
 699         nodes = res["output"]["topology"]["node"]
 700         self.assertEqual(1, len(nodes), 'Topology should contain 1 node')
 701         self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
 702         self.assertEqual("ROADM-infra", res["output"]["topology"]["node"][0]["name"][0]["value"],
 703                          'node name should be: ROADM-infra')
 704
 705     def test_43_get_tapi_topology_T100G(self):
 706         url = "{}/operations/tapi-topology:get-topology-details"
 707         data = {
 708             "tapi-topology:input": {
 709                 "tapi-topology:topology-id-or-name": "Transponder 100GE"
 710             }
 711         }
 712         response = test_utils.post_request(url, data)
 713         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 714         res = response.json()
 715         self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'Topology should contain 1 node')
 716         self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
 717         self.assertNotIn("owned-node-edge-point", res["output"]["topology"]["node"][0],
 718                          'Node should contain no owned-node-edge-points')
 719
 720     def test_44_disconnect_roadma(self):
 721         response = test_utils.unmount_device("ROADM-A1")
 722         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 723         time.sleep(5)
 724
 725     def test_45_disconnect_roadmc(self):
 726         response = test_utils.unmount_device("ROADM-C1")
 727         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 728         time.sleep(5)
 729
 730     def test_46_check_tapi_topos(self):
 731         self.test_01_get_tapi_topology_T100G()
 732         self.test_02_get_tapi_topology_T0()
 733
 734     def test_47_disconnect_xpdra(self):
 735         response = test_utils.unmount_device("XPDR-A1")
 736         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 737         time.sleep(5)
 738
 739     def test_48_disconnect_xpdrc(self):
 740         response = test_utils.unmount_device("XPDR-C1")
 741         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 742         time.sleep(5)
 743
 744     def test_49_disconnect_spdr_sa1(self):
 745         response = test_utils.unmount_device("SPDR-SA1")
 746         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 747         time.sleep(5)
 748
 749     def test_50_disconnect_spdr_sc1(self):
 750         response = test_utils.unmount_device("SPDR-SC1")
 751         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 752
 753
 754 def find_object_with_key(list_dicts, key, value):
 755     for dict_ in list_dicts:
 756         if dict_[key] == value:
 757             return dict_
 758     return None
 759
 760 def count_object_with_double_key(list_dicts, key1, key2, value):
 761     nb = 0
 762     for dict in list_dicts:
 763         if dict[key1][0][key2] == value:
 764             nb += 1
 765     return nb
 766
 767
 768 if __name__ == "__main__":
 769     unittest.main(verbosity=2)