tests/transportpce_tests/2.2.1/test10_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
  29 class TransportTapitesting(unittest.TestCase):
  30
  31     processes = None
  32     WAITING = 20
  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"
  39         },
  40         "service-name": "service1-OCH-OTU4",
  41         "common-id": "commonId",
  42         "connection-type": "infrastructure",
  43         "service-a-end": {
  44             "service-rate": "100",
  45             "node-id": "SPDR-SA1",
  46             "service-format": "OTU",
  47             "otu-service-rate": "org-openroadm-otn-common-types:OTU4",
  48             "clli": "NodeSA",
  49             "subrate-eth-sla": {
  50                     "subrate-eth-sla": {
  51                         "committed-info-rate": "100000",
  52                         "committed-burst-size": "64"
  53                     }
  54             },
  55             "tx-direction": {
  56                 "port": {
  57                     "port-device-name": "SPDR-SA1-XPDR1",
  58                     "port-type": "fixed",
  59                     "port-name": "XPDR1-NETWORK1",
  60                     "port-rack": "000000.00",
  61                     "port-shelf": "Chassis#1"
  62                 },
  63                 "lgx": {
  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"
  68                 }
  69             },
  70             "rx-direction": {
  71                 "port": {
  72                     "port-device-name": "SPDR-SA1-XPDR1",
  73                     "port-type": "fixed",
  74                     "port-name": "XPDR1-NETWORK1",
  75                     "port-rack": "000000.00",
  76                     "port-shelf": "Chassis#1"
  77                 },
  78                 "lgx": {
  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"
  83                 }
  84             },
  85             "optic-type": "gray"
  86         },
  87         "service-z-end": {
  88             "service-rate": "100",
  89             "node-id": "SPDR-SC1",
  90             "service-format": "OTU",
  91             "otu-service-rate": "org-openroadm-otn-common-types:OTU4",
  92             "clli": "NodeSC",
  93             "subrate-eth-sla": {
  94                     "subrate-eth-sla": {
  95                         "committed-info-rate": "100000",
  96                         "committed-burst-size": "64"
  97                     }
  98             },
  99             "tx-direction": {
 100                 "port": {
 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"
 106                 },
 107                 "lgx": {
 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"
 112                 }
 113             },
 114             "rx-direction": {
 115                 "port": {
 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"
 121                 },
 122                 "lgx": {
 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"
 127                 }
 128             },
 129             "optic-type": "gray"
 130         },
 131         "due-date": "2018-06-15T00:00:01Z",
 132         "operator-contact": "pw1234"
 133     }
 134     }
 135
 136     @classmethod
 137     def setUpClass(cls):
 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)
 154         if cls.init_failed:
 155             print("tapi installation feature failed...")
 156             test_utils.shutdown_process(cls.processes[0])
 157             sys.exit(2)
 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)])
 165
 166     @classmethod
 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")
 172
 173     def setUp(self):  # instruction executed before each test method
 174         if self.init_failed:
 175             self.fail('Feature installation failed')
 176         print("execution of {}".format(self.id().split(".")[-1]))
 177
 178     def test_01_get_tapi_topology_T100G(self):
 179         url = "{}/operations/tapi-topology:get-topology-details"
 180         data = {
 181             "tapi-topology:input": {
 182                 "tapi-topology:topology-id-or-name": "Transponder 100GE"
 183             }
 184         }
 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')
 198
 199     def test_02_get_tapi_topology_T0(self):
 200         url = "{}/operations/tapi-topology:get-topology-details"
 201         data = {
 202             "tapi-topology:input": {
 203                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 204             }
 205         }
 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')
 211
 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)
 215         time.sleep(10)
 216
 217     def test_04_check_tapi_topos(self):
 218         url = "{}/operations/tapi-topology:get-topology-details"
 219         data = {
 220             "tapi-topology:input": {
 221                 "tapi-topology:topology-id-or-name": "Transponder 100GE"
 222             }
 223         }
 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')
 229
 230         url = "{}/operations/tapi-topology:get-topology-details"
 231         data = {
 232             "tapi-topology:input": {
 233                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 234             }
 235         }
 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')
 241
 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)
 245         time.sleep(5)
 246
 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)
 250         time.sleep(10)
 251
 252     def test_07_check_tapi_topos(self):
 253         self.test_04_check_tapi_topos()
 254
 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)
 258         time.sleep(10)
 259
 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)
 263         time.sleep(10)
 264
 265     def test_10_check_tapi_topos(self):
 266         self.test_01_get_tapi_topology_T100G()
 267
 268         url = "{}/operations/tapi-topology:get-topology-details"
 269         data = {
 270             "tapi-topology:input": {
 271                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 272             }
 273         }
 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')
 285
 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)
 293         time.sleep(2)
 294
 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)
 302         time.sleep(2)
 303
 304     def test_13_check_tapi_topology_T100G(self):
 305         url = "{}/operations/tapi-topology:get-topology-details"
 306         data = {
 307             "tapi-topology:input": {
 308                 "tapi-topology:topology-id-or-name": "Transponder 100GE"
 309             }
 310         }
 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')
 319
 320     def test_14_check_tapi_topology_T0(self):
 321         url = "{}/operations/tapi-topology:get-topology-details"
 322         data = {
 323             "tapi-topology:input": {
 324                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 325             }
 326         }
 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')
 342
 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)
 346         time.sleep(10)
 347
 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)
 355         time.sleep(2)
 356
 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)
 364         time.sleep(2)
 365
 366     def test_18_check_tapi_topology_T100G(self):
 367         url = "{}/operations/tapi-topology:get-topology-details"
 368         data = {
 369             "tapi-topology:input": {
 370                 "tapi-topology:topology-id-or-name": "Transponder 100GE"
 371             }
 372         }
 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')
 384
 385     def test_19_check_tapi_topology_T0(self):
 386         url = "{}/operations/tapi-topology:get-topology-details"
 387         data = {
 388             "tapi-topology:input": {
 389                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 390             }
 391         }
 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')
 407
 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)
 411         time.sleep(10)
 412         # TODO replace connect and disconnect timers with test_utils.wait_until_log_contains
 413
 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)
 417         time.sleep(10)
 418         # TODO replace connect and disconnect timers with test_utils.wait_until_log_contains
 419
 420     def test_22_check_tapi_topology_T100G(self):
 421         self.test_18_check_tapi_topology_T100G()
 422
 423     def test_23_check_tapi_topology_T0(self):
 424         self.test_19_check_tapi_topology_T0()
 425
 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)
 433         time.sleep(2)
 434
 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)
 442         time.sleep(2)
 443
 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)
 451         time.sleep(2)
 452
 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)
 460         time.sleep(2)
 461
 462     def test_28_check_tapi_topology_T100G(self):
 463         self.test_18_check_tapi_topology_T100G()
 464
 465     def test_29_check_tapi_topology_T0(self):
 466         url = "{}/operations/tapi-topology:get-topology-details"
 467         data = {
 468             "tapi-topology:input": {
 469                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 470             }
 471         }
 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')
 487
 488     def test_30_add_oms_attributes(self):
 489         # Config ROADMA-ROADMC oms-attributes
 490         data = {"span": {
 491             "auto-spanloss": "true",
 492             "spanloss-base": 11.4,
 493             "spanloss-current": 12,
 494             "engineered-spanloss": 12.2,
 495             "link-concatenation": [{
 496                 "SRLG-Id": 0,
 497                 "fiber-type": "smf",
 498                 "SRLG-length": 100000,
 499                 "pmd": 0.5}]}}
 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
 503         data = {"span": {
 504             "auto-spanloss": "true",
 505             "spanloss-base": 11.4,
 506             "spanloss-current": 12,
 507             "engineered-spanloss": 12.2,
 508             "link-concatenation": [{
 509                 "SRLG-Id": 0,
 510                 "fiber-type": "smf",
 511                 "SRLG-length": 100000,
 512                 "pmd": 0.5}]}}
 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)
 515
 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)
 523
 524     def test_32_check_tapi_topology_T0(self):
 525         url = "{}/operations/tapi-topology:get-topology-details"
 526         data = {
 527             "tapi-topology:input": {
 528                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 529             }
 530         }
 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')
 544         for link in links:
 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')
 552
 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"
 561
 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)
 568
 569     def test_34_check_tapi_topology_T0(self):
 570         url = "{}/operations/tapi-topology:get-topology-details"
 571         data = {
 572             "tapi-topology:input": {
 573                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 574             }
 575         }
 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')
 589         for link in 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')
 600
 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)
 608         time.sleep(2)
 609
 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)
 617         time.sleep(2)
 618
 619     def test_37_check_tapi_topology_T0(self):
 620         url = "{}/operations/tapi-topology:get-topology-details"
 621         data = {
 622             "tapi-topology:input": {
 623                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 624             }
 625         }
 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')
 643
 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'])
 650         time.sleep(20)
 651
 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'])
 658         time.sleep(20)
 659
 660     def test_40_check_tapi_topology_T0(self):
 661         url = "{}/operations/tapi-topology:get-topology-details"
 662         data = {
 663             "tapi-topology:input": {
 664                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 665             }
 666         }
 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')
 676
 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']
 683         for link in links:
 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)
 689
 690     def test_42_check_tapi_topology_T0(self):
 691         url = "{}/operations/tapi-topology:get-topology-details"
 692         data = {
 693             "tapi-topology:input": {
 694                 "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
 695             }
 696         }
 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')
 705
 706     def test_43_get_tapi_topology_T100G(self):
 707         url = "{}/operations/tapi-topology:get-topology-details"
 708         data = {
 709             "tapi-topology:input": {
 710                 "tapi-topology:topology-id-or-name": "Transponder 100GE"
 711             }
 712         }
 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')
 720
 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)
 724         time.sleep(5)
 725
 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)
 729         time.sleep(5)
 730
 731     def test_46_check_tapi_topos(self):
 732         self.test_01_get_tapi_topology_T100G()
 733         self.test_02_get_tapi_topology_T0()
 734
 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)
 738         time.sleep(5)
 739
 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)
 743         time.sleep(5)
 744
 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)
 748         time.sleep(5)
 749
 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)
 753
 754
 755 def find_object_with_key(list_dicts, key, value):
 756     for dict_ in list_dicts:
 757         if dict_[key] == value:
 758             return dict_
 759     return None
 760
 761
 762 def count_object_with_double_key(list_dicts, key1, key2, value):
 763     nb = 0
 764     for dict in list_dicts:
 765         if dict[key1][0][key2] == value:
 766             nb += 1
 767     return nb
 768
 769
 770 if __name__ == "__main__":
 771     unittest.main(verbosity=2)