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