tests/transportpce_tests/1.2.1/test_renderer_service_path_nominal.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 TransportPCERendererTesting(unittest.TestCase):
  25
  26     honeynode_process1 = None
  27     honeynode_process2 = None
  28     odl_process = None
  29     restconf_baseurl = "http://localhost:8181/restconf"
  30
  31 #START_IGNORE_XTESTING
  32
  33     @classmethod
  34     def __start_honeynode1(cls):
  35         executable = ("./honeynode/2.1/honeynode-distribution/target/honeynode-distribution-1.18.01-hc"
  36                       "/honeynode-distribution-1.18.01/honeycomb-tpce")
  37         if os.path.isfile(executable):
  38             with open('honeynode1.log', 'w') as outfile:
  39                 cls.honeynode_process1 = subprocess.Popen(
  40                     [executable, "17830", "sample_configs/openroadm/2.1/oper-ROADMA.xml"],
  41                     stdout=outfile)
  42
  43     @classmethod
  44     def __start_honeynode2(cls):
  45         executable = ("./honeynode/2.1/honeynode-distribution/target/honeynode-distribution-1.18.01-hc"
  46                       "/honeynode-distribution-1.18.01/honeycomb-tpce")
  47         if os.path.isfile(executable):
  48             with open('honeynode2.log', 'w') as outfile:
  49                 cls.honeynode_process2 = subprocess.Popen(
  50                     [executable, "17831", "sample_configs/openroadm/2.1/oper-XPDRA.xml"],
  51                     stdout=outfile)
  52
  53     @classmethod
  54     def __start_odl(cls):
  55         executable = "../karaf/target/assembly/bin/karaf"
  56         with open('odl.log', 'w') as outfile:
  57             cls.odl_process = subprocess.Popen(
  58                 ["bash", executable, "server"], stdout=outfile,
  59                 stdin=open(os.devnull))
  60
  61     @classmethod
  62     def setUpClass(cls):
  63         cls.__start_honeynode1()
  64         time.sleep(40)
  65         cls.__start_honeynode2()
  66         time.sleep(40)
  67         cls.__start_odl()
  68         time.sleep(60)
  69
  70     @classmethod
  71     def tearDownClass(cls):
  72         for child in psutil.Process(cls.odl_process.pid).children():
  73             child.send_signal(signal.SIGINT)
  74             child.wait()
  75         cls.odl_process.send_signal(signal.SIGINT)
  76         cls.odl_process.wait()
  77         for child in psutil.Process(cls.honeynode_process1.pid).children():
  78             child.send_signal(signal.SIGINT)
  79             child.wait()
  80         cls.honeynode_process1.send_signal(signal.SIGINT)
  81         cls.honeynode_process1.wait()
  82         for child in psutil.Process(cls.honeynode_process2.pid).children():
  83             child.send_signal(signal.SIGINT)
  84             child.wait()
  85         cls.honeynode_process2.send_signal(signal.SIGINT)
  86         cls.honeynode_process2.wait()
  87
  88     def setUp(self):
  89         print ("execution of {}".format(self.id().split(".")[-1]))
  90         time.sleep(10)
  91
  92 #END_IGNORE_XTESTING
  93
  94     def test_01_rdm_device_connected(self):
  95         url = ("{}/config/network-topology:"
  96                "network-topology/topology/topology-netconf/node/ROADMA"
  97                .format(self.restconf_baseurl))
  98         data = {"node": [{
  99              "node-id": "ROADMA",
 100              "netconf-node-topology:username": "admin",
 101              "netconf-node-topology:password": "admin",
 102              "netconf-node-topology:host": "127.0.0.1",
 103              "netconf-node-topology:port": "17830",
 104              "netconf-node-topology:tcp-only": "false",
 105              "netconf-node-topology:pass-through": {}}]}
 106         headers = {'content-type': 'application/json'}
 107         response = requests.request(
 108              "PUT", url, data=json.dumps(data), headers=headers,
 109               auth=('admin', 'admin'))
 110         self.assertEqual(response.status_code, requests.codes.created)
 111         time.sleep(20)
 112
 113     def test_02_xpdr_device_connected(self):
 114         url = ("{}/config/network-topology:"
 115                "network-topology/topology/topology-netconf/node/XPDRA"
 116               .format(self.restconf_baseurl))
 117         data = {"node": [{
 118             "node-id": "XPDRA",
 119             "netconf-node-topology:username": "admin",
 120             "netconf-node-topology:password": "admin",
 121             "netconf-node-topology:host": "127.0.0.1",
 122             "netconf-node-topology:port": "17831",
 123             "netconf-node-topology:tcp-only": "false",
 124             "netconf-node-topology:pass-through": {}}]}
 125         headers = {'content-type': 'application/json'}
 126         response = requests.request(
 127             "PUT", url, data=json.dumps(data), headers=headers,
 128             auth=('admin', 'admin'))
 129         self.assertEqual(response.status_code, requests.codes.created)
 130         time.sleep(20)
 131
 132     def test_03_rdm_portmapping(self):
 133         url = ("{}/config/transportpce-portmapping:network/"
 134                "nodes/ROADMA"
 135                .format(self.restconf_baseurl))
 136         headers = {'content-type': 'application/json'}
 137         response = requests.request(
 138              "GET", url, headers=headers, auth=('admin', 'admin'))
 139         self.assertEqual(response.status_code, requests.codes.ok)
 140         res = response.json()
 141         self.assertIn(
 142              {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0',
 143               'logical-connection-point': 'DEG1-TTP-TXRX', 'port-direction': 'bidirectional'},
 144              res['nodes'][0]['mapping'])
 145         self.assertIn(
 146              {'supporting-port': 'C7', 'supporting-circuit-pack-name': '4/0',
 147               'logical-connection-point': 'SRG1-PP7-TXRX', 'port-direction': 'bidirectional'},
 148              res['nodes'][0]['mapping'])
 149
 150     def test_04_xpdr_portmapping(self):
 151         url = ("{}/config/transportpce-portmapping:network/"
 152                "nodes/XPDRA"
 153                .format(self.restconf_baseurl))
 154         headers = {'content-type': 'application/json'}
 155         response = requests.request(
 156              "GET", url, headers=headers, auth=('admin', 'admin'))
 157         self.assertEqual(response.status_code, requests.codes.ok)
 158         res = response.json()
 159         self.assertIn(
 160              {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 161               'logical-connection-point': 'XPDR1-NETWORK1', 'port-direction': 'bidirectional',
 162               'associated-lcp': 'XPDR1-CLIENT1', 'port-qual': 'xpdr-network'},
 163              res['nodes'][0]['mapping'])
 164         self.assertIn(
 165              {'supporting-port': 'C1',
 166               'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
 167               'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
 168               'associated-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client'},
 169              res['nodes'][0]['mapping'])
 170
 171     def test_05_service_path_create(self):
 172         url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl)
 173         data = {"renderer:input": {
 174              "renderer:service-name": "service_test",
 175              "renderer:wave-number": "7",
 176              "renderer:modulation-format": "qpsk",
 177              "renderer:operation": "create",
 178              "renderer:nodes": [
 179                  {"renderer:node-id": "ROADMA",
 180                   "renderer:src-tp": "SRG1-PP7-TXRX",
 181                   "renderer:dest-tp": "DEG1-TTP-TXRX"},
 182                  {"renderer:node-id": "XPDRA",
 183                   "renderer:src-tp": "XPDR1-CLIENT1",
 184                   "renderer:dest-tp": "XPDR1-NETWORK1"}]}}
 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('Roadm-connection successfully created for nodes: ROADMA', res["output"]["result"])
 192
 193     def test_06_service_path_create_rdm_check(self):
 194         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 195                 "node/ROADMA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 196                 "interface/DEG1-TTP-TXRX-7"
 197                 .format(self.restconf_baseurl))
 198         headers = {'content-type': 'application/json'}
 199         response = requests.request(
 200              "GET", url, headers=headers, auth=('admin', 'admin'))
 201         self.assertEqual(response.status_code, requests.codes.ok)
 202         res = response.json()
 203         self.assertDictContainsSubset({'name': 'DEG1-TTP-TXRX-7', 'administrative-state': 'inService',
 204               'supporting-circuit-pack-name': '2/0',
 205               'type': 'org-openroadm-interfaces:opticalChannel',
 206               'supporting-port': 'L1'}, res['interface'][0])
 207         self.assertDictEqual(
 208              {'wavelength-number': 7},
 209              res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
 210
 211     def test_07_service_path_create_rdm_check(self):
 212         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 213                 "node/ROADMA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 214                 "interface/SRG1-PP7-TXRX-7"
 215                 .format(self.restconf_baseurl))
 216         headers = {'content-type': 'application/json'}
 217         response = requests.request(
 218              "GET", url, headers=headers, auth=('admin', 'admin'))
 219         self.assertEqual(response.status_code, requests.codes.ok)
 220         res = response.json()
 221         self.assertDictContainsSubset(
 222              {'name': 'SRG1-PP7-TXRX-7', 'administrative-state': 'inService',
 223               'supporting-circuit-pack-name': '4/0',
 224               'type': 'org-openroadm-interfaces:opticalChannel',
 225               'supporting-port': 'C7'},
 226              res['interface'][0])
 227         self.assertDictEqual(
 228              {'wavelength-number': 7},
 229              res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
 230
 231     def test_08_service_path_create_rdm_check(self):
 232         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 233                "node/ROADMA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 234                "roadm-connections/SRG1-PP7-TXRX-DEG1-TTP-TXRX-7"
 235                .format(self.restconf_baseurl))
 236         headers = {'content-type': 'application/json'}
 237         response = requests.request(
 238              "GET", url, headers=headers, auth=('admin', 'admin'))
 239         self.assertEqual(response.status_code, requests.codes.ok)
 240         res = response.json()
 241         self.assertDictContainsSubset(
 242              {'connection-number': 'SRG1-PP7-TXRX-DEG1-TTP-TXRX-7',
 243               'wavelength-number': 7,
 244               'opticalControlMode': 'off'},
 245              res['roadm-connections'][0])
 246         self.assertDictEqual(
 247              {'src-if': 'SRG1-PP7-TXRX-7'},
 248              res['roadm-connections'][0]['source'])
 249         self.assertDictEqual(
 250              {'dst-if': 'DEG1-TTP-TXRX-7'},
 251              res['roadm-connections'][0]['destination'])
 252
 253     def test_09_service_path_create_xpdr_check(self):
 254         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 255                 "node/XPDRA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 256                 "interface/XPDR1-NETWORK1-7"
 257                 .format(self.restconf_baseurl))
 258         headers = {'content-type': 'application/json'}
 259         response = requests.request(
 260              "GET", url, headers=headers, auth=('admin', 'admin'))
 261         self.assertEqual(response.status_code, requests.codes.ok)
 262         res = response.json()
 263         self.assertDictContainsSubset(
 264              {'name': 'XPDR1-NETWORK1-7', 'administrative-state': 'inService',
 265               'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 266               'type': 'org-openroadm-interfaces:opticalChannel',
 267               'supporting-port': '1'},
 268              res['interface'][0])
 269         self.assertDictEqual(
 270              {u'rate': u'org-openroadm-optical-channel-interfaces:R100G',
 271               u'transmit-power':-5,
 272               u'wavelength-number': 7,
 273               u'modulation-format': u'dp-qpsk'},
 274              res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
 275
 276     def test_10_service_path_create_xpdr_check(self):
 277         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 278                 "node/XPDRA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 279                 "interface/XPDR1-NETWORK1-OTU"
 280                 .format(self.restconf_baseurl))
 281         headers = {'content-type': 'application/json'}
 282         response = requests.request(
 283              "GET", url, headers=headers, auth=('admin', 'admin'))
 284         self.assertEqual(response.status_code, requests.codes.ok)
 285         res = response.json()
 286         self.assertDictContainsSubset(
 287              {'name': 'XPDR1-NETWORK1-OTU', 'administrative-state': 'inService',
 288               'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 289               'type': 'org-openroadm-interfaces:otnOtu',
 290               'supporting-port': '1',
 291               'supporting-interface': 'XPDR1-NETWORK1-7'},
 292              res['interface'][0])
 293         self.assertDictEqual(
 294              {u'rate': u'org-openroadm-otn-otu-interfaces:OTU4',
 295               u'fec': u'scfec'},
 296              res['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
 297
 298     def test_11_service_path_create_xpdr_check(self):
 299         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 300                 "node/XPDRA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 301                 "interface/XPDR1-NETWORK1-ODU"
 302                 .format(self.restconf_baseurl))
 303         headers = {'content-type': 'application/json'}
 304         response = requests.request(
 305              "GET", url, headers=headers, auth=('admin', 'admin'))
 306         self.assertEqual(response.status_code, requests.codes.ok)
 307         res = response.json()
 308         self.assertDictContainsSubset(
 309              {'name': 'XPDR1-NETWORK1-ODU', 'administrative-state': 'inService',
 310               'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 311               'type': 'org-openroadm-interfaces:otnOdu',
 312               'supporting-port': '1',
 313               'supporting-interface': 'XPDR1-NETWORK1-OTU'},
 314              res['interface'][0])
 315         self.assertDictContainsSubset(
 316              {'rate': 'org-openroadm-otn-odu-interfaces:ODU4',
 317               u'monitoring-mode': u'terminated'},
 318              res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'])
 319         self.assertDictEqual({u'exp-payload-type': u'07', u'payload-type': u'07'},
 320                               res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
 321
 322     def test_12_service_path_create_xpdr_check(self):
 323         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 324                "node/XPDRA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 325                "interface/XPDR1-CLIENT1-ETHERNET"
 326                .format(self.restconf_baseurl))
 327         headers = {'content-type': 'application/json'}
 328         response = requests.request(
 329              "GET", url, headers=headers, auth=('admin', 'admin'))
 330         self.assertEqual(response.status_code, requests.codes.ok)
 331         res = response.json()
 332         self.assertDictContainsSubset(
 333              {'name': 'XPDR1-CLIENT1-ETHERNET', 'administrative-state': 'inService',
 334               'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
 335               'type': 'org-openroadm-interfaces:ethernetCsmacd',
 336               'supporting-port': 'C1'},
 337              res['interface'][0])
 338         self.assertDictEqual(
 339              {'speed': 100000,
 340               'mtu': 9000,
 341               'auto-negotiation': 'enabled',
 342               'duplex': 'full',
 343               'fec': 'off'},
 344              res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
 345
 346     def test_13_service_path_create_xpdr_check(self):
 347         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 348                 "node/XPDRA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 349                 "circuit-packs/1%2F0%2F1-PLUG-NET"
 350                 .format(self.restconf_baseurl))
 351         headers = {'content-type': 'application/json'}
 352         response = requests.request(
 353              "GET", url, headers=headers, auth=('admin', 'admin'))
 354         self.assertEqual(response.status_code, requests.codes.ok)
 355         res = response.json()
 356         self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
 357
 358     def test_14_service_path_delete(self):
 359         url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl)
 360         data = {"renderer:input": {
 361              "renderer:service-name": "service_test",
 362              "renderer:wave-number": "7",
 363              "renderer:operation": "delete",
 364              "renderer:nodes": [
 365                  {"renderer:node-id": "ROADMA",
 366                   "renderer:src-tp": "SRG1-PP7-TXRX",
 367                   "renderer:dest-tp": "DEG1-TTP-TXRX"},
 368                  {"renderer:node-id": "XPDRA",
 369                   "renderer:src-tp": "XPDR1-CLIENT1",
 370                   "renderer:dest-tp": "XPDR1-NETWORK1"}]}}
 371         headers = {'content-type': 'application/json'}
 372         response = requests.request(
 373              "POST", url, data=json.dumps(data),
 374              headers=headers, auth=('admin', 'admin'))
 375         self.assertEqual(response.status_code, requests.codes.ok)
 376         self.assertEqual(response.json(), {
 377              'output': {'result': 'Request processed', 'success': True}})
 378
 379     def test_15_service_path_delete_rdm_check(self):
 380         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 381                 "node/ROADMA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 382                 "interface/DEG1-TTP-TXRX-7"
 383                 .format(self.restconf_baseurl))
 384         headers = {'content-type': 'application/json'}
 385         response = requests.request(
 386              "GET", url, headers=headers, auth=('admin', 'admin'))
 387         self.assertEqual(response.status_code, requests.codes.not_found)
 388         res = response.json()
 389         self.assertIn(
 390              {"error-type":"application", "error-tag":"data-missing",
 391               "error-message":"Request could not be completed because the relevant data model content does not exist"},
 392              res['errors']['error'])
 393
 394     def test_16_service_path_delete_rdm_check(self):
 395         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 396                "node/ROADMA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 397                "interface/SRG1-PP7-TXRX-7"
 398                .format(self.restconf_baseurl))
 399         headers = {'content-type': 'application/json'}
 400         response = requests.request(
 401              "GET", url, headers=headers, auth=('admin', 'admin'))
 402         self.assertEqual(response.status_code, requests.codes.not_found)
 403         res = response.json()
 404         self.assertIn(
 405              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 406              res['errors']['error'])
 407
 408     def test_17_service_path_delete_rdm_check(self):
 409         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 410                "node/ROADMA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 411                "roadm-connections/SRG1-PP7-TXRX-DEG1-TTP-TXRX-7"
 412                .format(self.restconf_baseurl))
 413         headers = {'content-type': 'application/json'}
 414         response = requests.request(
 415              "GET", url, headers=headers, auth=('admin', 'admin'))
 416         self.assertEqual(response.status_code, requests.codes.not_found)
 417         res = response.json()
 418         self.assertIn(
 419              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 420              res['errors']['error'])
 421
 422     def test_18_service_path_delete_xpdr_check(self):
 423         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 424                "node/XPDRA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 425                "interface/XPDR1-NETWORK1-7"
 426                .format(self.restconf_baseurl))
 427         headers = {'content-type': 'application/json'}
 428         response = requests.request(
 429              "GET", url, headers=headers, auth=('admin', 'admin'))
 430         self.assertEqual(response.status_code, requests.codes.not_found)
 431         res = response.json()
 432         self.assertIn(
 433              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 434              res['errors']['error'])
 435
 436     def test_19_service_path_delete_xpdr_check(self):
 437         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 438                "node/XPDRA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 439                "interface/XPDR1-NETWORK1-OTU"
 440                .format(self.restconf_baseurl))
 441         headers = {'content-type': 'application/json'}
 442         response = requests.request(
 443              "GET", url, headers=headers, auth=('admin', 'admin'))
 444         self.assertEqual(response.status_code, requests.codes.not_found)
 445         res = response.json()
 446         self.assertIn(
 447              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 448              res['errors']['error'])
 449
 450     def test_20_service_path_delete_xpdr_check(self):
 451         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 452                "node/XPDRA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 453                "interface/XPDR1-NETWORK1-ODU"
 454                .format(self.restconf_baseurl))
 455         headers = {'content-type': 'application/json'}
 456         response = requests.request(
 457              "GET", url, headers=headers, auth=('admin', 'admin'))
 458         self.assertEqual(response.status_code, requests.codes.not_found)
 459         res = response.json()
 460         self.assertIn(
 461              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 462              res['errors']['error'])
 463
 464     def test_21_service_path_delete_xpdr_check(self):
 465         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 466                "node/XPDRA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 467                "interface/XPDR1-CLIENT1-ETHERNET"
 468                .format(self.restconf_baseurl))
 469         headers = {'content-type': 'application/json'}
 470         response = requests.request(
 471              "GET", url, headers=headers, auth=('admin', 'admin'))
 472         self.assertEqual(response.status_code, requests.codes.not_found)
 473         res = response.json()
 474         self.assertIn(
 475              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 476              res['errors']['error'])
 477
 478     def test_22_service_path_delete_xpdr_check(self):
 479         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 480                "node/XPDRA/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 481                "circuit-packs/1%2F0%2F1-PLUG-NET"
 482                .format(self.restconf_baseurl))
 483         headers = {'content-type': 'application/json'}
 484         response = requests.request(
 485             "GET", url, headers=headers, auth=('admin', 'admin'))
 486         self.assertEqual(response.status_code, requests.codes.ok)
 487         res = response.json()
 488         self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
 489
 490     def test_23_rdm_device_disconnected(self):
 491         url = ("{}/config/network-topology:"
 492                "network-topology/topology/topology-netconf/node/ROADMA"
 493                .format(self.restconf_baseurl))
 494         headers = {'content-type': 'application/json'}
 495         response = requests.request(
 496               "DELETE", url, headers=headers,
 497               auth=('admin', 'admin'))
 498         self.assertEqual(response.status_code, requests.codes.ok)
 499         time.sleep(20)
 500
 501     def test_24_xpdr_device_disconnected(self):
 502         url = ("{}/config/network-topology:"
 503                 "network-topology/topology/topology-netconf/node/XPDRA"
 504                .format(self.restconf_baseurl))
 505         headers = {'content-type': 'application/json'}
 506         response = requests.request(
 507              "DELETE", url, headers=headers,
 508              auth=('admin', 'admin'))
 509         self.assertEqual(response.status_code, requests.codes.ok)
 510         time.sleep(20)
 511
 512
 513 if __name__ == "__main__":
 514     unittest.main(verbosity=2, failfast=True)