tests/transportpce_tests/1.2.1/test_olm.py

   1 #!/usr/bin/env python
   2
   3 #############################################################################
   4 # Copyright (c) 2017 Orange, Inc. and others.  All rights reserved.
   5 #
   6 # All rights reserved. This program and the accompanying materials
   7 # are made available under the terms of the Apache License, Version 2.0
   8 # which accompanies this distribution, and is available at
   9 # http://www.apache.org/licenses/LICENSE-2.0
  10 #############################################################################
  11
  12 import unittest
  13 import time
  14 import json
  15 #from unittest.result import failfast
  16 import requests
  17 from common import test_utils
  18
  19
  20 class TransportOlmTesting(unittest.TestCase):
  21
  22     processes = None
  23
  24     @classmethod
  25     def setUpClass(cls):
  26         cls.processes = test_utils.start_tpce()
  27         cls.processes = test_utils.start_sims(['xpdra', 'roadma-full', 'roadmc-full', 'xpdrc'])
  28
  29     @classmethod
  30     def tearDownClass(cls):
  31         for process in cls.processes:
  32             test_utils.shutdown_process(process)
  33         print("all processes killed")
  34
  35     def setUp(self):
  36         print("execution of {}".format(self.id().split(".")[-1]))
  37         time.sleep(1)
  38
  39     def test_01_xpdrA_device_connected(self):
  40         response = test_utils.mount_device("XPDRA01", 'xpdra')
  41         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  42
  43     def test_02_xpdrC_device_connected(self):
  44         response = test_utils.mount_device("XPDRC01", 'xpdrc')
  45         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  46
  47     def test_03_rdmA_device_connected(self):
  48         response = test_utils.mount_device("ROADMA01", 'roadma-full')
  49         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  50
  51     def test_04_rdmC_device_connected(self):
  52         response = test_utils.mount_device("ROADMC01", 'roadmc-full')
  53         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  54
  55     def test_05_connect_xprdA_to_roadmA(self):
  56         url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL)
  57         data = {
  58             "networkutils:input": {
  59                 "networkutils:links-input": {
  60                     "networkutils:xpdr-node": "XPDRA01",
  61                     "networkutils:xpdr-num": "1",
  62                     "networkutils:network-num": "1",
  63                     "networkutils:rdm-node": "ROADMA01",
  64                     "networkutils:srg-num": "1",
  65                     "networkutils:termination-point-num": "SRG1-PP1-TXRX"
  66                 }
  67             }
  68         }
  69         response = requests.request(
  70             "POST", url, data=json.dumps(data),
  71             headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
  72         self.assertEqual(response.status_code, requests.codes.ok)
  73         res = response.json()
  74         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
  75
  76     def test_06_connect_roadmA_to_xpdrA(self):
  77         url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL)
  78         data = {
  79             "networkutils:input": {
  80                 "networkutils:links-input": {
  81                     "networkutils:xpdr-node": "XPDRA01",
  82                     "networkutils:xpdr-num": "1",
  83                     "networkutils:network-num": "1",
  84                     "networkutils:rdm-node": "ROADMA01",
  85                     "networkutils:srg-num": "1",
  86                     "networkutils:termination-point-num": "SRG1-PP1-TXRX"
  87                 }
  88             }
  89         }
  90         response = requests.request(
  91             "POST", url, data=json.dumps(data),
  92             headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
  93         self.assertEqual(response.status_code, requests.codes.ok)
  94         res = response.json()
  95         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
  96
  97     def test_07_connect_xprdC_to_roadmC(self):
  98         url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL)
  99         data = {
 100             "networkutils:input": {
 101                 "networkutils:links-input": {
 102                     "networkutils:xpdr-node": "XPDRC01",
 103                     "networkutils:xpdr-num": "1",
 104                     "networkutils:network-num": "1",
 105                     "networkutils:rdm-node": "ROADMC01",
 106                     "networkutils:srg-num": "1",
 107                     "networkutils:termination-point-num": "SRG1-PP1-TXRX"
 108                 }
 109             }
 110         }
 111         response = requests.request(
 112             "POST", url, data=json.dumps(data),
 113             headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 114         self.assertEqual(response.status_code, requests.codes.ok)
 115         res = response.json()
 116         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
 117
 118     def test_08_connect_roadmC_to_xpdrC(self):
 119         url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL)
 120         data = {
 121             "networkutils:input": {
 122                 "networkutils:links-input": {
 123                     "networkutils:xpdr-node": "XPDRC01",
 124                     "networkutils:xpdr-num": "1",
 125                     "networkutils:network-num": "1",
 126                     "networkutils:rdm-node": "ROADMC01",
 127                     "networkutils:srg-num": "1",
 128                     "networkutils:termination-point-num": "SRG1-PP1-TXRX"
 129                 }
 130             }
 131         }
 132         response = requests.request(
 133             "POST", url, data=json.dumps(data),
 134             headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 135         self.assertEqual(response.status_code, requests.codes.ok)
 136         res = response.json()
 137         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
 138
 139     def test_09_create_OTS_ROADMA(self):
 140         url = "{}/operations/transportpce-device-renderer:create-ots-oms".format(test_utils.RESTCONF_BASE_URL)
 141         data = {
 142             "input": {
 143                 "node-id": "ROADMA01",
 144                 "logical-connection-point": "DEG1-TTP-TXRX"
 145             }
 146         }
 147         response = requests.request(
 148             "POST", url, data=json.dumps(data),
 149             headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 150         self.assertEqual(response.status_code, requests.codes.ok)
 151         res = response.json()
 152         self.assertIn('Interfaces OTS-DEG1-TTP-TXRX - OMS-DEG1-TTP-TXRX successfully created on node ROADMA01',
 153                       res["output"]["result"])
 154
 155     def test_10_create_OTS_ROADMC(self):
 156         url = "{}/operations/transportpce-device-renderer:create-ots-oms".format(test_utils.RESTCONF_BASE_URL)
 157         data = {
 158             "input": {
 159                 "node-id": "ROADMC01",
 160                 "logical-connection-point": "DEG2-TTP-TXRX"
 161             }
 162         }
 163         response = requests.request(
 164             "POST", url, data=json.dumps(data),
 165             headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 166         self.assertEqual(response.status_code, requests.codes.ok)
 167         res = response.json()
 168         self.assertIn('Interfaces OTS-DEG2-TTP-TXRX - OMS-DEG2-TTP-TXRX successfully created on node ROADMC01',
 169                       res["output"]["result"])
 170
 171     def test_11_get_PM_ROADMA(self):
 172         url = "{}/operations/transportpce-olm:get-pm".format(test_utils.RESTCONF_BASE_URL)
 173         data = {
 174             "input": {
 175                 "node-id": "ROADMA01",
 176                 "resource-type": "interface",
 177                 "granularity": "15min",
 178                 "resource-identifier": {
 179                     "resource-name": "OTS-DEG1-TTP-TXRX"
 180                 }
 181             }
 182         }
 183         response = requests.request(
 184             "POST", url, data=json.dumps(data),
 185             headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 186         self.assertEqual(response.status_code, requests.codes.ok)
 187         res = response.json()
 188         self.assertIn({
 189             "pmparameter-name": "OpticalPowerOutput",
 190             "pmparameter-value": "2.5"
 191         }, res["output"]["measurements"])
 192         self.assertIn({
 193             "pmparameter-name": "OpticalReturnLoss",
 194             "pmparameter-value": "49.9"
 195         }, res["output"]["measurements"])
 196         self.assertIn({
 197             "pmparameter-name": "OpticalPowerInput",
 198             "pmparameter-value": "3"
 199         }, res["output"]["measurements"])
 200
 201     def test_12_get_PM_ROADMC(self):
 202         url = "{}/operations/transportpce-olm:get-pm".format(test_utils.RESTCONF_BASE_URL)
 203         data = {
 204             "input": {
 205                 "node-id": "ROADMC01",
 206                 "resource-type": "interface",
 207                 "granularity": "15min",
 208                 "resource-identifier": {
 209                     "resource-name": "OTS-DEG2-TTP-TXRX"
 210                 }
 211             }
 212         }
 213         response = requests.request(
 214             "POST", url, data=json.dumps(data),
 215             headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 216         self.assertEqual(response.status_code, requests.codes.ok)
 217         res = response.json()
 218         self.assertIn({
 219             "pmparameter-name": "OpticalPowerOutput",
 220             "pmparameter-value": "18.1"
 221         }, res["output"]["measurements"])
 222         self.assertIn({
 223             "pmparameter-name": "OpticalReturnLoss",
 224             "pmparameter-value": "48.8"
 225         }, res["output"]["measurements"])
 226         self.assertIn({
 227             "pmparameter-name": "OpticalPowerInput",
 228             "pmparameter-value": "-3.2"
 229         }, res["output"]["measurements"])
 230
 231     def test_13_calculate_span_loss_base_ROADMA_ROADMC(self):
 232         url = "{}/operations/transportpce-olm:calculate-spanloss-base".format(test_utils.RESTCONF_BASE_URL)
 233         data = {
 234             "input": {
 235                 "src-type": "link",
 236                 "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
 237             }
 238         }
 239         response = requests.request(
 240             "POST", url, data=json.dumps(data),
 241             headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 242         self.assertEqual(response.status_code, requests.codes.ok)
 243         res = response.json()
 244         self.assertIn('Success',
 245                       res["output"]["result"])
 246         self.assertIn({
 247             "spanloss": "5.7",
 248             "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
 249         }, res["output"]["spans"])
 250         time.sleep(5)
 251
 252     def test_14_calculate_span_loss_base_all(self):
 253         url = "{}/operations/transportpce-olm:calculate-spanloss-base".format(test_utils.RESTCONF_BASE_URL)
 254         data = {
 255             "input": {
 256                 "src-type": "all"
 257             }
 258         }
 259         response = requests.request(
 260             "POST", url, data=json.dumps(data),
 261             headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 262         self.assertEqual(response.status_code, requests.codes.ok)
 263         res = response.json()
 264         self.assertIn('Success',
 265                       res["output"]["result"])
 266         self.assertIn({
 267             "spanloss": "15.1",
 268             "link-id": "ROADMC01-DEG2-DEG2-TTP-TXRXtoROADMA01-DEG1-DEG1-TTP-TXRX"
 269         }, res["output"]["spans"])
 270         self.assertIn({
 271             "spanloss": "5.7",
 272             "link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
 273         }, res["output"]["spans"])
 274         time.sleep(5)
 275
 276     def test_15_get_OTS_DEG1_TTP_TXRX_ROADMA(self):
 277         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 278                "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/interface/OTS-DEG1-TTP-TXRX/"
 279                "org-openroadm-optical-transport-interfaces:ots".format(test_utils.RESTCONF_BASE_URL))
 280         response = requests.request(
 281             "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 282         self.assertEqual(response.status_code, requests.codes.ok)
 283         res = response.json()
 284         self.assertEqual(5.7, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit'])
 285         self.assertEqual(15.1, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-receive'])
 286
 287     def test_16_get_OTS_DEG2_TTP_TXRX_ROADMC(self):
 288         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 289                "node/ROADMC01/yang-ext:mount/org-openroadm-device:org-openroadm-device/interface/OTS-DEG2-TTP-TXRX/"
 290                "org-openroadm-optical-transport-interfaces:ots".format(test_utils.RESTCONF_BASE_URL))
 291         response = requests.request(
 292             "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 293         self.assertEqual(response.status_code, requests.codes.ok)
 294         res = response.json()
 295         self.assertEqual(15.1, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit'])
 296         self.assertEqual(5.7, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-receive'])
 297
 298     def test_17_servicePath_create_AToZ(self):
 299         url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
 300         data = {
 301             "input": {
 302                 "service-name": "test",
 303                 "wave-number": "1",
 304                 "modulation-format": "qpsk",
 305                 "operation": "create",
 306                 "nodes": [
 307                     {
 308                         "dest-tp": "XPDR1-NETWORK1",
 309                         "src-tp": "XPDR1-CLIENT1",
 310                         "node-id": "XPDRA01"
 311                     },
 312                     {
 313                         "dest-tp": "DEG1-TTP-TXRX",
 314                         "src-tp": "SRG1-PP1-TXRX",
 315                         "node-id": "ROADMA01"
 316                     },
 317                     {
 318                         "dest-tp": "SRG1-PP1-TXRX",
 319                         "src-tp": "DEG2-TTP-TXRX",
 320                         "node-id": "ROADMC01"
 321                     },
 322                     {
 323                         "dest-tp": "XPDR1-CLIENT1",
 324                         "src-tp": "XPDR1-NETWORK1",
 325                         "node-id": "XPDRC01"
 326                     }
 327                 ]
 328             }
 329         }
 330         response = requests.request(
 331             "POST", url, data=json.dumps(data),
 332             headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 333         self.assertEqual(response.status_code, requests.codes.ok)
 334         res = response.json()
 335         self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
 336         # time.sleep(40)
 337         time.sleep(10)
 338
 339     def test_18_servicePath_create_ZToA(self):
 340         url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
 341         data = {
 342             "input": {
 343                 "service-name": "test",
 344                 "wave-number": "1",
 345                 "modulation-format": "qpsk",
 346                 "operation": "create",
 347                 "nodes": [
 348                     {
 349                         "dest-tp": "XPDR1-NETWORK1",
 350                         "src-tp": "XPDR1-CLIENT1",
 351                         "node-id": "XPDRC01"
 352                     },
 353                     {
 354                         "dest-tp": "DEG2-TTP-TXRX",
 355                         "src-tp": "SRG1-PP1-TXRX",
 356                         "node-id": "ROADMC01"
 357                     },
 358                     {
 359                         "src-tp": "DEG1-TTP-TXRX",
 360                         "dest-tp": "SRG1-PP1-TXRX",
 361                         "node-id": "ROADMA01"
 362                     },
 363                     {
 364                         "src-tp": "XPDR1-NETWORK1",
 365                         "dest-tp": "XPDR1-CLIENT1",
 366                         "node-id": "XPDRA01"
 367                     }
 368                 ]
 369             }
 370         }
 371         response = requests.request(
 372             "POST", url, data=json.dumps(data),
 373             headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 374         self.assertEqual(response.status_code, requests.codes.ok)
 375         res = response.json()
 376         self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
 377         # time.sleep(40)
 378         time.sleep(10)
 379
 380     def test_19_service_power_setup_XPDRA_XPDRC(self):
 381         url = "{}/operations/transportpce-olm:service-power-setup".format(test_utils.RESTCONF_BASE_URL)
 382         data = {
 383             "input": {
 384                 "service-name": "test",
 385                 "wave-number": 1,
 386                 "nodes": [
 387                     {
 388                         "dest-tp": "XPDR1-NETWORK1",
 389                         "src-tp": "XPDR1-CLIENT1",
 390                         "node-id": "XPDRA01"
 391                     },
 392                     {
 393                         "dest-tp": "DEG1-TTP-TXRX",
 394                         "src-tp": "SRG1-PP1-TXRX",
 395                         "node-id": "ROADMA01"
 396                     },
 397                     {
 398                         "dest-tp": "SRG1-PP1-TXRX",
 399                         "src-tp": "DEG2-TTP-TXRX",
 400                         "node-id": "ROADMC01"
 401                     },
 402                     {
 403                         "dest-tp": "XPDR1-CLIENT1",
 404                         "src-tp": "XPDR1-NETWORK1",
 405                         "node-id": "XPDRC01"
 406                     }
 407                 ]
 408             }
 409         }
 410         response = requests.request(
 411             "POST", url, data=json.dumps(data),
 412             headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 413         self.assertEqual(response.status_code, requests.codes.ok)
 414         res = response.json()
 415         self.assertIn('Success', res["output"]["result"])
 416
 417     def test_20_get_interface_XPDRA_XPDR1_NETWORK1(self):
 418         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/XPDRA01/yang-ext:mount/"
 419                "org-openroadm-device:org-openroadm-device/interface/XPDR1-NETWORK1-1/"
 420                "org-openroadm-optical-channel-interfaces:och".format(test_utils.RESTCONF_BASE_URL))
 421         response = requests.request(
 422             "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 423         self.assertEqual(response.status_code, requests.codes.ok)
 424         res = response.json()
 425         self.assertEqual(0, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
 426         self.assertEqual(1, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number'])
 427
 428     def test_21_get_roadmconnection_ROADMA(self):
 429         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADMA01/yang-ext:mount/"
 430                "org-openroadm-device:org-openroadm-device/roadm-connections/"
 431                "SRG1-PP1-TXRX-DEG1-TTP-TXRX-1".format(test_utils.RESTCONF_BASE_URL))
 432         response = requests.request(
 433             "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 434         self.assertEqual(response.status_code, requests.codes.ok)
 435         res = response.json()
 436         self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode'])
 437         self.assertEqual(-3.3, res['roadm-connections'][0]['target-output-power'])
 438
 439     def test_22_get_roadmconnection_ROADMC(self):
 440         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADMC01/yang-ext:mount/"
 441                "org-openroadm-device:org-openroadm-device/roadm-connections/"
 442                "DEG2-TTP-TXRX-SRG1-PP1-TXRX-1".format(test_utils.RESTCONF_BASE_URL))
 443         response = requests.request(
 444             "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 445         self.assertEqual(response.status_code, requests.codes.ok)
 446         res = response.json()
 447         self.assertEqual("power", res['roadm-connections'][0]['opticalControlMode'])
 448
 449     def test_23_service_power_setup_XPDRC_XPDRA(self):
 450         url = "{}/operations/transportpce-olm:service-power-setup".format(test_utils.RESTCONF_BASE_URL)
 451         data = {
 452             "input": {
 453                 "service-name": "test",
 454                 "wave-number": 1,
 455                 "nodes": [
 456                     {
 457                         "dest-tp": "XPDR1-NETWORK1",
 458                         "src-tp": "XPDR1-CLIENT1",
 459                         "node-id": "XPDRC01"
 460                     },
 461                     {
 462                         "dest-tp": "DEG2-TTP-TXRX",
 463                         "src-tp": "SRG1-PP1-TXRX",
 464                         "node-id": "ROADMC01"
 465                     },
 466                     {
 467                         "src-tp": "DEG1-TTP-TXRX",
 468                         "dest-tp": "SRG1-PP1-TXRX",
 469                         "node-id": "ROADMA01"
 470                     },
 471                     {
 472                         "src-tp": "XPDR1-NETWORK1",
 473                         "dest-tp": "XPDR1-CLIENT1",
 474                         "node-id": "XPDRA01"
 475                     }
 476                 ]
 477             }
 478         }
 479         response = requests.request(
 480             "POST", url, data=json.dumps(data),
 481             headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 482         self.assertEqual(response.status_code, requests.codes.ok)
 483         res = response.json()
 484         self.assertIn('Success', res["output"]["result"])
 485
 486     def test_24_get_interface_XPDRC_XPDR1_NETWORK1(self):
 487         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/XPDRC01/yang-ext:mount/"
 488                "org-openroadm-device:org-openroadm-device/interface/XPDR1-NETWORK1-1/"
 489                "org-openroadm-optical-channel-interfaces:och".format(test_utils.RESTCONF_BASE_URL))
 490         response = requests.request(
 491             "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 492         self.assertEqual(response.status_code, requests.codes.ok)
 493         res = response.json()
 494         self.assertEqual(0, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
 495         self.assertEqual(1, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number'])
 496
 497     def test_25_get_roadmconnection_ROADMC(self):
 498         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADMC01/yang-ext:mount/"
 499                "org-openroadm-device:org-openroadm-device/roadm-connections/"
 500                "SRG1-PP1-TXRX-DEG2-TTP-TXRX-1".format(test_utils.RESTCONF_BASE_URL))
 501         response = requests.request(
 502             "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 503         self.assertEqual(response.status_code, requests.codes.ok)
 504         res = response.json()
 505         self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode'])
 506         self.assertEqual(2, res['roadm-connections'][0]['target-output-power'])
 507
 508     def test_26_service_power_turndown_XPDRA_XPDRC(self):
 509         url = "{}/operations/transportpce-olm:service-power-turndown".format(test_utils.RESTCONF_BASE_URL)
 510         data = {
 511             "input": {
 512                 "service-name": "test",
 513                 "wave-number": 1,
 514                 "nodes": [
 515                     {
 516                         "dest-tp": "XPDR1-NETWORK1",
 517                         "src-tp": "XPDR1-CLIENT1",
 518                         "node-id": "XPDRA01"
 519                     },
 520                     {
 521                         "dest-tp": "DEG1-TTP-TXRX",
 522                         "src-tp": "SRG1-PP1-TXRX",
 523                         "node-id": "ROADMA01"
 524                     },
 525                     {
 526                         "dest-tp": "SRG1-PP1-TXRX",
 527                         "src-tp": "DEG2-TTP-TXRX",
 528                         "node-id": "ROADMC01"
 529                     },
 530                     {
 531                         "dest-tp": "XPDR1-CLIENT1",
 532                         "src-tp": "XPDR1-NETWORK1",
 533                         "node-id": "XPDRC01"
 534                     }
 535                 ]
 536             }
 537         }
 538         response = requests.request(
 539             "POST", url, data=json.dumps(data),
 540             headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 541         self.assertEqual(response.status_code, requests.codes.ok)
 542         res = response.json()
 543         self.assertIn('Success', res["output"]["result"])
 544
 545     def test_27_get_roadmconnection_ROADMA(self):
 546         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADMA01/yang-ext:mount/"
 547                "org-openroadm-device:org-openroadm-device/roadm-connections/"
 548                "SRG1-PP1-TXRX-DEG1-TTP-TXRX-1".format(test_utils.RESTCONF_BASE_URL))
 549         response = requests.request(
 550             "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 551         self.assertEqual(response.status_code, requests.codes.ok)
 552         res = response.json()
 553         self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode'])
 554         self.assertEqual(-60, res['roadm-connections'][0]['target-output-power'])
 555
 556     def test_28_get_roadmconnection_ROADMC(self):
 557         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADMC01/yang-ext:mount/"
 558                "org-openroadm-device:org-openroadm-device/roadm-connections/"
 559                "DEG2-TTP-TXRX-SRG1-PP1-TXRX-1".format(test_utils.RESTCONF_BASE_URL))
 560         response = requests.request(
 561             "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 562         self.assertEqual(response.status_code, requests.codes.ok)
 563         res = response.json()
 564         self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode'])
 565
 566     def test_29_servicePath_delete_AToZ(self):
 567         url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
 568         data = {
 569             "input": {
 570                 "service-name": "test",
 571                 "wave-number": "1",
 572                 "modulation-format": "qpsk",
 573                 "operation": "delete",
 574                 "nodes": [
 575                     {
 576                         "dest-tp": "XPDR1-NETWORK1",
 577                         "src-tp": "XPDR1-CLIENT1",
 578                         "node-id": "XPDRA01"
 579                     },
 580                     {
 581                         "dest-tp": "DEG1-TTP-TXRX",
 582                         "src-tp": "SRG1-PP1-TXRX",
 583                         "node-id": "ROADMA01"
 584                     },
 585                     {
 586                         "dest-tp": "SRG1-PP1-TXRX",
 587                         "src-tp": "DEG2-TTP-TXRX",
 588                         "node-id": "ROADMC01"
 589                     },
 590                     {
 591                         "dest-tp": "XPDR1-CLIENT1",
 592                         "src-tp": "XPDR1-NETWORK1",
 593                         "node-id": "XPDRC01"
 594                     }
 595                 ]
 596             }
 597         }
 598         response = requests.request(
 599             "POST", url, data=json.dumps(data),
 600             headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 601         self.assertEqual(response.status_code, requests.codes.ok)
 602         res = response.json()
 603         self.assertIn('Request processed', res["output"]["result"])
 604         time.sleep(10)
 605
 606     def test_30_servicePath_delete_ZToA(self):
 607         url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
 608         data = {
 609             "input": {
 610                 "service-name": "test",
 611                 "wave-number": "1",
 612                 "modulation-format": "qpsk",
 613                 "operation": "delete",
 614                 "nodes": [
 615                     {
 616                         "dest-tp": "XPDR1-NETWORK1",
 617                         "src-tp": "XPDR1-CLIENT1",
 618                         "node-id": "XPDRC01"
 619                     },
 620                     {
 621                         "dest-tp": "DEG2-TTP-TXRX",
 622                         "src-tp": "SRG1-PP1-TXRX",
 623                         "node-id": "ROADMC01"
 624                     },
 625                     {
 626                         "src-tp": "DEG1-TTP-TXRX",
 627                         "dest-tp": "SRG1-PP1-TXRX",
 628                         "node-id": "ROADMA01"
 629                     },
 630                     {
 631                         "src-tp": "XPDR1-NETWORK1",
 632                         "dest-tp": "XPDR1-CLIENT1",
 633                         "node-id": "XPDRA01"
 634                     }
 635                 ]
 636             }
 637         }
 638         response = requests.request(
 639             "POST", url, data=json.dumps(data),
 640             headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 641         self.assertEqual(response.status_code, requests.codes.ok)
 642         res = response.json()
 643         self.assertIn('Request processed', res["output"]["result"])
 644         time.sleep(10)
 645
 646     """to test case where SRG where the xpdr is connected to has no optical range data"""
 647
 648     def test_31_connect_xprdA_to_roadmA(self):
 649         url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL)
 650         data = {
 651             "networkutils:input": {
 652                 "networkutils:links-input": {
 653                     "networkutils:xpdr-node": "XPDRA01",
 654                     "networkutils:xpdr-num": "1",
 655                     "networkutils:network-num": "2",
 656                     "networkutils:rdm-node": "ROADMA01",
 657                     "networkutils:srg-num": "1",
 658                     "networkutils:termination-point-num": "SRG1-PP2-TXRX"
 659                 }
 660             }
 661         }
 662         response = requests.request(
 663             "POST", url, data=json.dumps(data),
 664             headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 665         self.assertEqual(response.status_code, requests.codes.ok)
 666         res = response.json()
 667         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
 668
 669     def test_32_connect_roadmA_to_xpdrA(self):
 670         url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL)
 671         data = {
 672             "networkutils:input": {
 673                 "networkutils:links-input": {
 674                     "networkutils:xpdr-node": "XPDRA01",
 675                     "networkutils:xpdr-num": "1",
 676                     "networkutils:network-num": "2",
 677                     "networkutils:rdm-node": "ROADMA01",
 678                     "networkutils:srg-num": "1",
 679                     "networkutils:termination-point-num": "SRG1-PP2-TXRX"
 680                 }
 681             }
 682         }
 683         response = requests.request(
 684             "POST", url, data=json.dumps(data),
 685             headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 686         self.assertEqual(response.status_code, requests.codes.ok)
 687         res = response.json()
 688         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
 689
 690     def test_33_servicePath_create_AToZ(self):
 691         url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
 692         data = {
 693             "input": {
 694                 "service-name": "test2",
 695                 "wave-number": "2",
 696                 "modulation-format": "qpsk",
 697                 "operation": "create",
 698                 "nodes": [
 699                     {
 700                         "dest-tp": "XPDR1-NETWORK2",
 701                         "src-tp": "XPDR1-CLIENT2",
 702                         "node-id": "XPDRA01"
 703                     },
 704                     {
 705                         "dest-tp": "DEG1-TTP-TXRX",
 706                         "src-tp": "SRG1-PP2-TXRX",
 707                         "node-id": "ROADMA01"
 708                     }
 709                 ]
 710             }
 711         }
 712         response = requests.request(
 713             "POST", url, data=json.dumps(data),
 714             headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 715         self.assertEqual(response.status_code, requests.codes.ok)
 716         res = response.json()
 717         self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
 718         # time.sleep(40)
 719         time.sleep(10)
 720
 721     def test_34_get_interface_XPDRA_XPDR1_NETWORK2(self):
 722         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/XPDRA01/yang-ext:mount/"
 723                "org-openroadm-device:org-openroadm-device/interface/XPDR1-NETWORK2-2/"
 724                "org-openroadm-optical-channel-interfaces:och".format(test_utils.RESTCONF_BASE_URL))
 725         response = requests.request(
 726             "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 727         self.assertEqual(response.status_code, requests.codes.ok)
 728         res = response.json()
 729         self.assertEqual(-5, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
 730         self.assertEqual(2, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number'])
 731
 732     def test_35_servicePath_delete_AToZ(self):
 733         url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
 734         data = {
 735             "input": {
 736                 "service-name": "test",
 737                 "wave-number": "1",
 738                 "modulation-format": "qpsk",
 739                 "operation": "delete",
 740                 "nodes": [
 741                     {
 742                         "dest-tp": "XPDR1-NETWORK2",
 743                         "src-tp": "XPDR1-CLIENT2",
 744                         "node-id": "XPDRA01"
 745                     },
 746                     {
 747                         "dest-tp": "DEG1-TTP-TXRX",
 748                         "src-tp": "SRG1-PP2-TXRX",
 749                         "node-id": "ROADMA01"
 750                     }
 751                 ]
 752             }
 753         }
 754         response = requests.request(
 755             "POST", url, data=json.dumps(data),
 756             headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 757         self.assertEqual(response.status_code, requests.codes.ok)
 758         res = response.json()
 759         self.assertIn('Request processed', res["output"]["result"])
 760         time.sleep(10)
 761
 762     def test_36_xpdrA_device_disconnected(self):
 763         response = test_utils.unmount_device("XPDRA01")
 764         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 765
 766     def test_37_xpdrC_device_disconnected(self):
 767         response = test_utils.unmount_device("XPDRC01")
 768         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 769
 770     def test_38_calculate_span_loss_current(self):
 771         url = "{}/operations/transportpce-olm:calculate-spanloss-current".format(test_utils.RESTCONF_BASE_URL)
 772         response = requests.request(
 773             "POST", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 774         self.assertEqual(response.status_code, requests.codes.ok)
 775         res = response.json()
 776         self.assertIn('Success',
 777                       res["output"]["result"])
 778         time.sleep(5)
 779
 780     def test_39_rdmA_device_disconnected(self):
 781         response = test_utils.unmount_device("ROADMA01")
 782         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 783
 784     def test_40_rdmC_device_disconnected(self):
 785         response = test_utils.unmount_device("ROADMC01")
 786         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 787
 788
 789 if __name__ == "__main__":
 790     unittest.main(verbosity=2)