tests/transportpce_tests/2.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 import test_utils
  23
  24
  25 class TransportPCERendererTesting(unittest.TestCase):
  26
  27     honeynode_process1 = None
  28     honeynode_process2 = None
  29     odl_process = None
  30     restconf_baseurl = "http://localhost:8181/restconf"
  31
  32 #START_IGNORE_XTESTING
  33
  34     @classmethod
  35     def setUpClass(cls):
  36         print ("starting honeynode1...")
  37         cls.honeynode_process1 = test_utils.start_xpdra_honeynode()
  38         time.sleep(20)
  39
  40         print ("starting honeynode2...")
  41         cls.honeynode_process2 = test_utils.start_roadma_honeynode()
  42         time.sleep(20)
  43         print ("all honeynodes started")
  44
  45         print ("starting opendaylight...")
  46         cls.odl_process = test_utils.start_tpce()
  47         time.sleep(60)
  48         print ("opendaylight started")
  49
  50     @classmethod
  51     def tearDownClass(cls):
  52         for child in psutil.Process(cls.odl_process.pid).children():
  53             child.send_signal(signal.SIGINT)
  54             child.wait()
  55         cls.odl_process.send_signal(signal.SIGINT)
  56         cls.odl_process.wait()
  57         for child in psutil.Process(cls.honeynode_process1.pid).children():
  58             child.send_signal(signal.SIGINT)
  59             child.wait()
  60         cls.honeynode_process1.send_signal(signal.SIGINT)
  61         cls.honeynode_process1.wait()
  62         for child in psutil.Process(cls.honeynode_process2.pid).children():
  63             child.send_signal(signal.SIGINT)
  64             child.wait()
  65         cls.honeynode_process2.send_signal(signal.SIGINT)
  66         cls.honeynode_process2.wait()
  67
  68     def setUp(self):
  69         print ("execution of {}".format(self.id().split(".")[-1]))
  70         time.sleep(10)
  71
  72 #END_IGNORE_XTESTING
  73
  74     def test_01_rdm_device_connected(self):
  75         url = ("{}/config/network-topology:"
  76                "network-topology/topology/topology-netconf/node/ROADM-A1"
  77                .format(self.restconf_baseurl))
  78         data = {"node": [{
  79              "node-id": "ROADM-A1",
  80              "netconf-node-topology:username": "admin",
  81              "netconf-node-topology:password": "admin",
  82              "netconf-node-topology:host": "127.0.0.1",
  83              "netconf-node-topology:port": "17841",
  84              "netconf-node-topology:tcp-only": "false",
  85              "netconf-node-topology:pass-through": {}}]}
  86         headers = {'content-type': 'application/json'}
  87         response = requests.request(
  88              "PUT", url, data=json.dumps(data), headers=headers,
  89               auth=('admin', 'admin'))
  90         self.assertEqual(response.status_code, requests.codes.created)
  91         time.sleep(20)
  92
  93     def test_02_xpdr_device_connected(self):
  94         url = ("{}/config/network-topology:"
  95                "network-topology/topology/topology-netconf/node/XPDR-A1"
  96               .format(self.restconf_baseurl))
  97         data = {"node": [{
  98             "node-id": "XPDR-A1",
  99             "netconf-node-topology:username": "admin",
 100             "netconf-node-topology:password": "admin",
 101             "netconf-node-topology:host": "127.0.0.1",
 102             "netconf-node-topology:port": "17840",
 103             "netconf-node-topology:tcp-only": "false",
 104             "netconf-node-topology:pass-through": {}}]}
 105         headers = {'content-type': 'application/json'}
 106         response = requests.request(
 107             "PUT", url, data=json.dumps(data), headers=headers,
 108             auth=('admin', 'admin'))
 109         self.assertEqual(response.status_code, requests.codes.created)
 110         time.sleep(20)
 111
 112     def test_03_rdm_portmapping(self):
 113         url = ("{}/config/transportpce-portmapping:network/"
 114                "nodes/ROADM-A1"
 115                .format(self.restconf_baseurl))
 116         headers = {'content-type': 'application/json'}
 117         response = requests.request(
 118              "GET", url, headers=headers, auth=('admin', 'admin'))
 119         self.assertEqual(response.status_code, requests.codes.ok)
 120         res = response.json()
 121         self.assertIn(
 122              {'supporting-port': 'L1', 'supporting-circuit-pack-name': '1/0',
 123               'logical-connection-point': 'DEG1-TTP-TXRX', 'port-direction': 'bidirectional'},
 124              res['nodes'][0]['mapping'])
 125         self.assertIn(
 126              {'supporting-port': 'C3', 'supporting-circuit-pack-name': '3/0',
 127               'logical-connection-point': 'SRG1-PP3-TXRX', 'port-direction': 'bidirectional'},
 128              res['nodes'][0]['mapping'])
 129
 130     def test_04_xpdr_portmapping(self):
 131         url = ("{}/config/transportpce-portmapping:network/"
 132                "nodes/XPDR-A1"
 133                .format(self.restconf_baseurl))
 134         headers = {'content-type': 'application/json'}
 135         response = requests.request(
 136              "GET", url, headers=headers, auth=('admin', 'admin'))
 137         self.assertEqual(response.status_code, requests.codes.ok)
 138         res = response.json()
 139         self.assertIn(
 140              {'supported-interface-capability': ['org-openroadm-port-types:if-OCH'],
 141               'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 142               'logical-connection-point': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-network',
 143               'port-direction': 'bidirectional', 'connection-map-lcp': 'XPDR1-CLIENT1'},
 144              res['nodes'][0]['mapping'])
 145         self.assertIn(
 146              {'supported-interface-capability': ['org-openroadm-port-types:if-100GE'],
 147               'supporting-port': 'C1',
 148               'supporting-circuit-pack-name': '1/0/1-PLUG-CLIENT',
 149               'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
 150               'connection-map-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client'},
 151              res['nodes'][0]['mapping'])
 152
 153     def test_05_service_path_create(self):
 154         url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl)
 155         data = {"renderer:input": {
 156              "renderer:service-name": "service_test",
 157              "renderer:wave-number": "7",
 158              "renderer:modulation-format": "qpsk",
 159              "renderer:operation": "create",
 160              "renderer:nodes": [
 161                  {"renderer:node-id": "ROADM-A1",
 162                   "renderer:src-tp": "SRG1-PP3-TXRX",
 163                   "renderer:dest-tp": "DEG1-TTP-TXRX"},
 164                  {"renderer:node-id": "XPDR-A1",
 165                   "renderer:src-tp": "XPDR1-CLIENT1",
 166                   "renderer:dest-tp": "XPDR1-NETWORK1"}]}}
 167         headers = {'content-type': 'application/json'}
 168         response = requests.request(
 169              "POST", url, data=json.dumps(data),
 170              headers=headers, auth=('admin', 'admin'))
 171         self.assertEqual(response.status_code, requests.codes.ok)
 172         res = response.json()
 173         self.assertIn('Roadm-connection successfully created for nodes: ROADM-A1', res["output"]["result"])
 174
 175     def test_06_service_path_create_rdm_check(self):
 176         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 177                 "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 178                 "interface/DEG1-TTP-TXRX-nmc-7"
 179                 .format(self.restconf_baseurl))
 180         headers = {'content-type': 'application/json'}
 181         response = requests.request(
 182              "GET", url, headers=headers, auth=('admin', 'admin'))
 183         self.assertEqual(response.status_code, requests.codes.ok)
 184         res = response.json()
 185         self.assertDictContainsSubset({'name': 'DEG1-TTP-TXRX-nmc-7', 'administrative-state': 'inService',
 186               'supporting-circuit-pack-name': '1/0',
 187               'type': 'org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint',
 188               'supporting-port': 'L1'}, res['interface'][0])
 189         self.assertDictEqual(
 190              {u'frequency': 195.8, u'width': 40},
 191              res['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp'])
 192
 193     def test_07_service_path_create_rdm_check(self):
 194         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 195                 "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 196                 "interface/DEG1-TTP-TXRX-mc-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-mc-7', 'administrative-state': 'inService',
 204               'supporting-circuit-pack-name': '1/0',
 205               'type': 'org-openroadm-interfaces:mediaChannelTrailTerminationPoint',
 206               'supporting-port': 'L1'}, res['interface'][0])
 207         self.assertDictEqual(
 208              {u'min-freq': 195.775, u'max-freq': 195.825},
 209              res['interface'][0]['org-openroadm-media-channel-interfaces:mc-ttp'])
 210
 211
 212     def test_08_service_path_create_rdm_check(self):
 213         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 214                 "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 215                 "interface/SRG1-PP3-TXRX-nmc-7"
 216                 .format(self.restconf_baseurl))
 217         headers = {'content-type': 'application/json'}
 218         response = requests.request(
 219              "GET", url, headers=headers, auth=('admin', 'admin'))
 220         self.assertEqual(response.status_code, requests.codes.ok)
 221         res = response.json()
 222         self.assertDictContainsSubset({'name': 'SRG1-PP3-TXRX-nmc-7', 'administrative-state': 'inService',
 223               'supporting-circuit-pack-name': '3/0',
 224               'type': 'org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint',
 225               'supporting-port': 'C3'}, res['interface'][0])
 226         self.assertDictEqual(
 227              {u'frequency': 195.8, u'width': 40},
 228              res['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp'])
 229
 230     # -mc supporting interfaces must not be created for SRG, only degrees
 231     def test_09_service_path_create_rdm_check(self):
 232         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 233                 "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 234                 "interface/SRG1-PP3-TXRX-mc-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.not_found)
 240         res = response.json()
 241         self.assertIn(
 242              {"error-type":"application", "error-tag":"data-missing",
 243               "error-message":"Request could not be completed because the relevant data model content does not exist"},
 244              res['errors']['error'])
 245
 246     def test_10_service_path_create_rdm_check(self):
 247         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 248                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 249                "roadm-connections/SRG1-PP3-TXRX-DEG1-TTP-TXRX-7"
 250                .format(self.restconf_baseurl))
 251         headers = {'content-type': 'application/json'}
 252         response = requests.request(
 253              "GET", url, headers=headers, auth=('admin', 'admin'))
 254         self.assertEqual(response.status_code, requests.codes.ok)
 255         res = response.json()
 256         self.assertDictContainsSubset(
 257              {'connection-name': 'SRG1-PP3-TXRX-DEG1-TTP-TXRX-7',
 258               'opticalControlMode': 'off'},
 259              res['roadm-connections'][0])
 260         self.assertDictEqual(
 261              {'src-if': 'SRG1-PP3-TXRX-nmc-7'},
 262              res['roadm-connections'][0]['source'])
 263         self.assertDictEqual(
 264              {'dst-if': 'DEG1-TTP-TXRX-nmc-7'},
 265              res['roadm-connections'][0]['destination'])
 266
 267     def test_11_service_path_create_xpdr_check(self):
 268         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 269                 "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 270                 "interface/XPDR1-NETWORK1-7"
 271                 .format(self.restconf_baseurl))
 272         headers = {'content-type': 'application/json'}
 273         response = requests.request(
 274              "GET", url, headers=headers, auth=('admin', 'admin'))
 275         self.assertEqual(response.status_code, requests.codes.ok)
 276         res = response.json()
 277         self.assertDictContainsSubset(
 278              {'name': 'XPDR1-NETWORK1-7', 'administrative-state': 'inService',
 279               'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 280               'type': 'org-openroadm-interfaces:opticalChannel',
 281               'supporting-port': '1'},
 282              res['interface'][0])
 283         self.assertDictEqual(
 284              {u'rate': u'org-openroadm-common-types:R100G',
 285               u'transmit-power':-5,
 286               u'frequency': 195.8},
 287              res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
 288
 289     def test_12_service_path_create_xpdr_check(self):
 290         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 291                 "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 292                 "interface/XPDR1-NETWORK1-OTU"
 293                 .format(self.restconf_baseurl))
 294         headers = {'content-type': 'application/json'}
 295         response = requests.request(
 296              "GET", url, headers=headers, auth=('admin', 'admin'))
 297         self.assertEqual(response.status_code, requests.codes.ok)
 298         res = response.json()
 299         self.assertDictContainsSubset(
 300              {'name': 'XPDR1-NETWORK1-OTU', 'administrative-state': 'inService',
 301               'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 302               'type': 'org-openroadm-interfaces:otnOtu',
 303               'supporting-port': '1',
 304               'supporting-interface': 'XPDR1-NETWORK1-7'},
 305              res['interface'][0])
 306         self.assertDictEqual(
 307              {u'rate': u'org-openroadm-otn-common-types:OTU4',
 308               u'fec': u'scfec'},
 309              res['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
 310
 311     def test_13_service_path_create_xpdr_check(self):
 312         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 313                 "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 314                 "interface/XPDR1-NETWORK1-ODU"
 315                 .format(self.restconf_baseurl))
 316         headers = {'content-type': 'application/json'}
 317         response = requests.request(
 318              "GET", url, headers=headers, auth=('admin', 'admin'))
 319         self.assertEqual(response.status_code, requests.codes.ok)
 320         res = response.json()
 321         self.assertDictContainsSubset(
 322              {'name': 'XPDR1-NETWORK1-ODU', 'administrative-state': 'inService',
 323               'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 324               'type': 'org-openroadm-interfaces:otnOdu',
 325               'supporting-port': '1',
 326               'supporting-interface': 'XPDR1-NETWORK1-OTU'},
 327              res['interface'][0])
 328         self.assertDictContainsSubset(
 329              {'rate': 'org-openroadm-otn-common-types:ODU4',
 330               u'monitoring-mode': u'terminated'},
 331              res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'])
 332         self.assertDictEqual({u'exp-payload-type': u'07', u'payload-type': u'07'},
 333                              res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
 334
 335     def test_14_service_path_create_xpdr_check(self):
 336         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 337                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 338                "interface/XPDR1-CLIENT1-ETHERNET"
 339                .format(self.restconf_baseurl))
 340         headers = {'content-type': 'application/json'}
 341         response = requests.request(
 342              "GET", url, headers=headers, auth=('admin', 'admin'))
 343         self.assertEqual(response.status_code, requests.codes.ok)
 344         res = response.json()
 345         self.assertDictContainsSubset(
 346              {'name': 'XPDR1-CLIENT1-ETHERNET', 'administrative-state': 'inService',
 347               'supporting-circuit-pack-name': '1/0/1-PLUG-CLIENT',
 348               'type': 'org-openroadm-interfaces:ethernetCsmacd',
 349               'supporting-port': 'C1'},
 350              res['interface'][0])
 351         self.assertDictEqual(
 352              {u'fec': u'off', u'speed': 100000},
 353              res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
 354
 355     def test_15_service_path_create_xpdr_check(self):
 356         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 357                 "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 358                 "circuit-packs/1%2F0%2F1-PLUG-NET"
 359                 .format(self.restconf_baseurl))
 360         headers = {'content-type': 'application/json'}
 361         response = requests.request(
 362              "GET", url, headers=headers, auth=('admin', 'admin'))
 363         self.assertEqual(response.status_code, requests.codes.ok)
 364         res = response.json()
 365         self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
 366
 367     def test_16_service_path_create_xpdr_check(self):
 368         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 369                 "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 370                 "circuit-packs/1%2F0%2F1-PLUG-CLIENT"
 371                 .format(self.restconf_baseurl))
 372         headers = {'content-type': 'application/json'}
 373         response = requests.request(
 374              "GET", url, headers=headers, auth=('admin', 'admin'))
 375         self.assertEqual(response.status_code, requests.codes.ok)
 376         res = response.json()
 377         self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
 378
 379     def test_17_service_path_delete(self):
 380         url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl)
 381         data = {"renderer:input": {
 382              "renderer:service-name": "service_test",
 383              "renderer:wave-number": "7",
 384              "renderer:operation": "delete",
 385              "renderer:nodes": [
 386                  {"renderer:node-id": "ROADM-A1",
 387                   "renderer:src-tp": "SRG1-PP3-TXRX",
 388                   "renderer:dest-tp": "DEG1-TTP-TXRX"},
 389                  {"renderer:node-id": "XPDR-A1",
 390                   "renderer:src-tp": "XPDR1-CLIENT1",
 391                   "renderer:dest-tp": "XPDR1-NETWORK1"}]}}
 392         headers = {'content-type': 'application/json'}
 393         response = requests.request(
 394              "POST", url, data=json.dumps(data),
 395              headers=headers, auth=('admin', 'admin'))
 396         self.assertEqual(response.status_code, requests.codes.ok)
 397         self.assertEqual(response.json(), {
 398              'output': {'result': 'Request processed', 'success': True}})
 399
 400     def test_18_service_path_delete_rdm_check(self):
 401         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 402                 "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 403                 "interface/DEG1-TTP-TXRX-mc-7"
 404                 .format(self.restconf_baseurl))
 405         headers = {'content-type': 'application/json'}
 406         response = requests.request(
 407              "GET", url, headers=headers, auth=('admin', 'admin'))
 408         self.assertEqual(response.status_code, requests.codes.not_found)
 409         res = response.json()
 410         self.assertIn(
 411              {"error-type":"application", "error-tag":"data-missing",
 412               "error-message":"Request could not be completed because the relevant data model content does not exist"},
 413              res['errors']['error'])
 414
 415     def test_19_service_path_delete_rdm_check(self):
 416         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 417                 "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 418                 "interface/DEG1-TTP-TXRX-nmc-7"
 419                 .format(self.restconf_baseurl))
 420         headers = {'content-type': 'application/json'}
 421         response = requests.request(
 422              "GET", url, headers=headers, auth=('admin', 'admin'))
 423         self.assertEqual(response.status_code, requests.codes.not_found)
 424         res = response.json()
 425         self.assertIn(
 426              {"error-type":"application", "error-tag":"data-missing",
 427               "error-message":"Request could not be completed because the relevant data model content does not exist"},
 428              res['errors']['error'])
 429
 430     def test_20_service_path_delete_rdm_check(self):
 431         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 432                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 433                "interface/SRG1-PP3-TXRX-mc-7"
 434                .format(self.restconf_baseurl))
 435         headers = {'content-type': 'application/json'}
 436         response = requests.request(
 437              "GET", url, headers=headers, auth=('admin', 'admin'))
 438         self.assertEqual(response.status_code, requests.codes.not_found)
 439         res = response.json()
 440         self.assertIn(
 441              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 442              res['errors']['error'])
 443
 444     def test_21_service_path_delete_rdm_check(self):
 445         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 446                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 447                "interface/SRG1-PP3-TXRX-nmc-7"
 448                .format(self.restconf_baseurl))
 449         headers = {'content-type': 'application/json'}
 450         response = requests.request(
 451              "GET", url, headers=headers, auth=('admin', 'admin'))
 452         self.assertEqual(response.status_code, requests.codes.not_found)
 453         res = response.json()
 454         self.assertIn(
 455              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 456              res['errors']['error'])
 457
 458     def test_22_service_path_delete_rdm_check(self):
 459         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 460                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 461                "roadm-connections/SRG1-PP3-TXRX-DEG1-TTP-TXRX-7"
 462                .format(self.restconf_baseurl))
 463         headers = {'content-type': 'application/json'}
 464         response = requests.request(
 465              "GET", url, headers=headers, auth=('admin', 'admin'))
 466         self.assertEqual(response.status_code, requests.codes.not_found)
 467         res = response.json()
 468         self.assertIn(
 469              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 470              res['errors']['error'])
 471
 472     def test_23_service_path_delete_xpdr_check(self):
 473         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 474                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 475                "interface/XPDR1-NETWORK1-7"
 476                .format(self.restconf_baseurl))
 477         headers = {'content-type': 'application/json'}
 478         response = requests.request(
 479              "GET", url, headers=headers, auth=('admin', 'admin'))
 480         self.assertEqual(response.status_code, requests.codes.not_found)
 481         res = response.json()
 482         self.assertIn(
 483              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 484              res['errors']['error'])
 485
 486     def test_24_service_path_delete_xpdr_check(self):
 487         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 488                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 489                "interface/XPDR1-NETWORK1-OTU"
 490                .format(self.restconf_baseurl))
 491         headers = {'content-type': 'application/json'}
 492         response = requests.request(
 493              "GET", url, headers=headers, auth=('admin', 'admin'))
 494         self.assertEqual(response.status_code, requests.codes.not_found)
 495         res = response.json()
 496         self.assertIn(
 497              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 498              res['errors']['error'])
 499
 500     def test_25_service_path_delete_xpdr_check(self):
 501         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 502                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 503                "interface/XPDR1-NETWORK1-ODU"
 504                .format(self.restconf_baseurl))
 505         headers = {'content-type': 'application/json'}
 506         response = requests.request(
 507              "GET", url, headers=headers, auth=('admin', 'admin'))
 508         self.assertEqual(response.status_code, requests.codes.not_found)
 509         res = response.json()
 510         self.assertIn(
 511              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 512              res['errors']['error'])
 513
 514     def test_26_service_path_delete_xpdr_check(self):
 515         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 516                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 517                "interface/XPDR1-CLIENT1-ETHERNET"
 518                .format(self.restconf_baseurl))
 519         headers = {'content-type': 'application/json'}
 520         response = requests.request(
 521              "GET", url, headers=headers, auth=('admin', 'admin'))
 522         self.assertEqual(response.status_code, requests.codes.not_found)
 523         res = response.json()
 524         self.assertIn(
 525              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 526              res['errors']['error'])
 527
 528     def test_27_service_path_delete_xpdr_check(self):
 529         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 530                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 531                "circuit-packs/1%2F0%2F1-PLUG-NET"
 532                .format(self.restconf_baseurl))
 533         headers = {'content-type': 'application/json'}
 534         response = requests.request(
 535             "GET", url, headers=headers, auth=('admin', 'admin'))
 536         self.assertEqual(response.status_code, requests.codes.ok)
 537         res = response.json()
 538         self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
 539
 540     def test_28_service_path_delete_xpdr_check(self):
 541         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 542                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 543                "circuit-packs/1%2F0%2F1-PLUG-CLIENT"
 544                .format(self.restconf_baseurl))
 545         headers = {'content-type': 'application/json'}
 546         response = requests.request(
 547             "GET", url, headers=headers, auth=('admin', 'admin'))
 548         self.assertEqual(response.status_code, requests.codes.ok)
 549         res = response.json()
 550         self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
 551
 552     def test_29_rdm_device_disconnected(self):
 553         url = ("{}/config/network-topology:"
 554                "network-topology/topology/topology-netconf/node/ROADM-A1"
 555                .format(self.restconf_baseurl))
 556         headers = {'content-type': 'application/json'}
 557         response = requests.request(
 558               "DELETE", url, headers=headers,
 559               auth=('admin', 'admin'))
 560         self.assertEqual(response.status_code, requests.codes.ok)
 561         time.sleep(20)
 562
 563     def test_30_xpdr_device_disconnected(self):
 564         url = ("{}/config/network-topology:"
 565                 "network-topology/topology/topology-netconf/node/XPDR-A1"
 566                .format(self.restconf_baseurl))
 567         headers = {'content-type': 'application/json'}
 568         response = requests.request(
 569              "DELETE", url, headers=headers,
 570              auth=('admin', 'admin'))
 571         self.assertEqual(response.status_code, requests.codes.ok)
 572         time.sleep(20)
 573
 574
 575 if __name__ == "__main__":
 576     unittest.main(verbosity=2, failfast=True)