tests/transportpce_tests/2.2.1/test_tapi.py

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