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