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