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