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