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