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
  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.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, "17841", "sample_configs/openroadm/2.2.1/oper-ROADMA.xml"],
  41                     stdout=outfile)
  42
  43     @classmethod
  44     def __start_honeynode2(cls):
  45         executable = ("./honeynode/2.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, "17840", "sample_configs/openroadm/2.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(20)
  65         cls.__start_honeynode2()
  66         time.sleep(20)
  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/ROADM-A1"
  97                .format(self.restconf_baseurl))
  98         data = {"node": [{
  99              "node-id": "ROADM-A1",
 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": "17841",
 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/XPDR-A1"
 116               .format(self.restconf_baseurl))
 117         data = {"node": [{
 118             "node-id": "XPDR-A1",
 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": "17840",
 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/ROADM-A1"
 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': '1/0',
 143               'logical-connection-point': 'DEG1-TTP-TXRX', 'port-direction': 'bidirectional'},
 144              res['nodes'][0]['mapping'])
 145         self.assertIn(
 146              {'supporting-port': 'C3', 'supporting-circuit-pack-name': '3/0',
 147               'logical-connection-point': 'SRG1-PP3-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/XPDR-A1"
 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-qual': 'xpdr-network',
 162               'port-direction': 'bidirectional', 'associated-lcp': 'XPDR1-CLIENT1'},
 163              res['nodes'][0]['mapping'])
 164         self.assertIn(
 165              {'supporting-port': 'C1',
 166               'supporting-circuit-pack-name': '1/0/1-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": "ROADM-A1",
 180                   "renderer:src-tp": "SRG1-PP3-TXRX",
 181                   "renderer:dest-tp": "DEG1-TTP-TXRX"},
 182                  {"renderer:node-id": "XPDR-A1",
 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: ROADM-A1', 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/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 196                 "interface/DEG1-TTP-TXRX-nmc-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-nmc-7', 'administrative-state': 'inService',
 204               'supporting-circuit-pack-name': '1/0',
 205               'type': 'org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint',
 206               'supporting-port': 'L1'}, res['interface'][0])
 207         self.assertDictEqual(
 208              {u'frequency': 191.65, u'width': 40},
 209              res['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp'])
 210
 211     def test_07_service_path_create_rdm_check(self):
 212         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 213                 "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 214                 "interface/DEG1-TTP-TXRX-mc-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({'name': 'DEG1-TTP-TXRX-mc-7', 'administrative-state': 'inService',
 222               'supporting-circuit-pack-name': '1/0',
 223               'type': 'org-openroadm-interfaces:mediaChannelTrailTerminationPoint',
 224               'supporting-port': 'L1'}, res['interface'][0])
 225         self.assertDictEqual(
 226              {u'min-freq': 191.625, u'max-freq': 191.67499999999998},
 227              res['interface'][0]['org-openroadm-media-channel-interfaces:mc-ttp'])
 228
 229
 230     def test_08_service_path_create_rdm_check(self):
 231         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 232                 "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 233                 "interface/SRG1-PP3-TXRX-nmc-7"
 234                 .format(self.restconf_baseurl))
 235         headers = {'content-type': 'application/json'}
 236         response = requests.request(
 237              "GET", url, headers=headers, auth=('admin', 'admin'))
 238         self.assertEqual(response.status_code, requests.codes.ok)
 239         res = response.json()
 240         self.assertDictContainsSubset({'name': 'SRG1-PP3-TXRX-nmc-7', 'administrative-state': 'inService',
 241               'supporting-circuit-pack-name': '3/0',
 242               'type': 'org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint',
 243               'supporting-port': 'C3'}, res['interface'][0])
 244         self.assertDictEqual(
 245              {u'frequency': 191.65, u'width': 40},
 246              res['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp'])
 247
 248     # -mc supporting interfaces must not be created for SRG, only degrees
 249     def test_09_service_path_create_rdm_check(self):
 250         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 251                 "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 252                 "interface/SRG1-PP3-TXRX-mc-7"
 253                 .format(self.restconf_baseurl))
 254         headers = {'content-type': 'application/json'}
 255         response = requests.request(
 256              "GET", url, headers=headers, auth=('admin', 'admin'))
 257         self.assertEqual(response.status_code, requests.codes.not_found)
 258         res = response.json()
 259         self.assertIn(
 260              {"error-type":"application", "error-tag":"data-missing",
 261               "error-message":"Request could not be completed because the relevant data model content does not exist"},
 262              res['errors']['error'])
 263
 264     def test_10_service_path_create_rdm_check(self):
 265         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 266                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 267                "roadm-connections/SRG1-PP3-TXRX-DEG1-TTP-TXRX-7"
 268                .format(self.restconf_baseurl))
 269         headers = {'content-type': 'application/json'}
 270         response = requests.request(
 271              "GET", url, headers=headers, auth=('admin', 'admin'))
 272         self.assertEqual(response.status_code, requests.codes.ok)
 273         res = response.json()
 274         self.assertDictContainsSubset(
 275              {'connection-name': 'SRG1-PP3-TXRX-DEG1-TTP-TXRX-7',
 276               'opticalControlMode': 'off'},
 277              res['roadm-connections'][0])
 278         self.assertDictEqual(
 279              {'src-if': 'SRG1-PP3-TXRX-nmc-7'},
 280              res['roadm-connections'][0]['source'])
 281         self.assertDictEqual(
 282              {'dst-if': 'DEG1-TTP-TXRX-nmc-7'},
 283              res['roadm-connections'][0]['destination'])
 284
 285     def test_11_service_path_create_xpdr_check(self):
 286         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 287                 "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 288                 "interface/XPDR1-NETWORK1-7"
 289                 .format(self.restconf_baseurl))
 290         headers = {'content-type': 'application/json'}
 291         response = requests.request(
 292              "GET", url, headers=headers, auth=('admin', 'admin'))
 293         self.assertEqual(response.status_code, requests.codes.ok)
 294         res = response.json()
 295         self.assertDictContainsSubset(
 296              {'name': 'XPDR1-NETWORK1-7', 'administrative-state': 'inService',
 297               'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 298               'type': 'org-openroadm-interfaces:opticalChannel',
 299               'supporting-port': '1'},
 300              res['interface'][0])
 301         self.assertDictEqual(
 302              {u'rate': u'org-openroadm-common-types:R100G',
 303               u'transmit-power':-5,
 304               u'frequency': 191.65},
 305              res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
 306
 307     def test_12_service_path_create_xpdr_check(self):
 308         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 309                 "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 310                 "interface/XPDR1-NETWORK1-OTU"
 311                 .format(self.restconf_baseurl))
 312         headers = {'content-type': 'application/json'}
 313         response = requests.request(
 314              "GET", url, headers=headers, auth=('admin', 'admin'))
 315         self.assertEqual(response.status_code, requests.codes.ok)
 316         res = response.json()
 317         self.assertDictContainsSubset(
 318              {'name': 'XPDR1-NETWORK1-OTU', 'administrative-state': 'inService',
 319               'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 320               'type': 'org-openroadm-interfaces:otnOtu',
 321               'supporting-port': '1',
 322               'supporting-interface': 'XPDR1-NETWORK1-7'},
 323              res['interface'][0])
 324         self.assertDictEqual(
 325              {u'rate': u'org-openroadm-otn-common-types:OTU4',
 326               u'fec': u'scfec'},
 327              res['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
 328
 329     def test_13_service_path_create_xpdr_check(self):
 330         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 331                 "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 332                 "interface/XPDR1-NETWORK1-ODU"
 333                 .format(self.restconf_baseurl))
 334         headers = {'content-type': 'application/json'}
 335         response = requests.request(
 336              "GET", url, headers=headers, auth=('admin', 'admin'))
 337         self.assertEqual(response.status_code, requests.codes.ok)
 338         res = response.json()
 339         self.assertDictContainsSubset(
 340              {'name': 'XPDR1-NETWORK1-ODU', 'administrative-state': 'inService',
 341               'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 342               'type': 'org-openroadm-interfaces:otnOdu',
 343               'supporting-port': '1',
 344               'supporting-interface': 'XPDR1-NETWORK1-OTU'},
 345              res['interface'][0])
 346         self.assertDictContainsSubset(
 347              {'rate': 'org-openroadm-otn-common-types:ODU4',
 348               u'monitoring-mode': u'terminated'},
 349              res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'])
 350         self.assertDictEqual({u'exp-payload-type': u'07', u'payload-type': u'07'},
 351                              res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
 352
 353     def test_14_service_path_create_xpdr_check(self):
 354         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 355                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 356                "interface/XPDR1-CLIENT1-ETHERNET"
 357                .format(self.restconf_baseurl))
 358         headers = {'content-type': 'application/json'}
 359         response = requests.request(
 360              "GET", url, headers=headers, auth=('admin', 'admin'))
 361         self.assertEqual(response.status_code, requests.codes.ok)
 362         res = response.json()
 363         self.assertDictContainsSubset(
 364              {'name': 'XPDR1-CLIENT1-ETHERNET', 'administrative-state': 'inService',
 365               'supporting-circuit-pack-name': '1/0/1-PLUG-CLIENT',
 366               'type': 'org-openroadm-interfaces:ethernetCsmacd',
 367               'supporting-port': 'C1'},
 368              res['interface'][0])
 369         self.assertDictEqual(
 370              {u'fec': u'off', u'speed': 100000},
 371              res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
 372
 373     def test_15_service_path_create_xpdr_check(self):
 374         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 375                 "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 376                 "circuit-packs/1%2F0%2F1-PLUG-NET"
 377                 .format(self.restconf_baseurl))
 378         headers = {'content-type': 'application/json'}
 379         response = requests.request(
 380              "GET", url, headers=headers, auth=('admin', 'admin'))
 381         self.assertEqual(response.status_code, requests.codes.ok)
 382         res = response.json()
 383         self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
 384
 385     def test_16_service_path_create_xpdr_check(self):
 386         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 387                 "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 388                 "circuit-packs/1%2F0%2F1-PLUG-CLIENT"
 389                 .format(self.restconf_baseurl))
 390         headers = {'content-type': 'application/json'}
 391         response = requests.request(
 392              "GET", url, headers=headers, auth=('admin', 'admin'))
 393         self.assertEqual(response.status_code, requests.codes.ok)
 394         res = response.json()
 395         self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
 396
 397     def test_17_service_path_delete(self):
 398         url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl)
 399         data = {"renderer:input": {
 400              "renderer:service-name": "service_test",
 401              "renderer:wave-number": "7",
 402              "renderer:operation": "delete",
 403              "renderer:nodes": [
 404                  {"renderer:node-id": "ROADM-A1",
 405                   "renderer:src-tp": "SRG1-PP3-TXRX",
 406                   "renderer:dest-tp": "DEG1-TTP-TXRX"},
 407                  {"renderer:node-id": "XPDR-A1",
 408                   "renderer:src-tp": "XPDR1-CLIENT1",
 409                   "renderer:dest-tp": "XPDR1-NETWORK1"}]}}
 410         headers = {'content-type': 'application/json'}
 411         response = requests.request(
 412              "POST", url, data=json.dumps(data),
 413              headers=headers, auth=('admin', 'admin'))
 414         self.assertEqual(response.status_code, requests.codes.ok)
 415         self.assertEqual(response.json(), {
 416              'output': {'result': 'Request processed', 'success': True}})
 417
 418     def test_18_service_path_delete_rdm_check(self):
 419         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 420                 "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 421                 "interface/DEG1-TTP-TXRX-mc-7"
 422                 .format(self.restconf_baseurl))
 423         headers = {'content-type': 'application/json'}
 424         response = requests.request(
 425              "GET", url, headers=headers, auth=('admin', 'admin'))
 426         self.assertEqual(response.status_code, requests.codes.not_found)
 427         res = response.json()
 428         self.assertIn(
 429              {"error-type":"application", "error-tag":"data-missing",
 430               "error-message":"Request could not be completed because the relevant data model content does not exist"},
 431              res['errors']['error'])
 432
 433     def test_19_service_path_delete_rdm_check(self):
 434         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 435                 "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 436                 "interface/DEG1-TTP-TXRX-nmc-7"
 437                 .format(self.restconf_baseurl))
 438         headers = {'content-type': 'application/json'}
 439         response = requests.request(
 440              "GET", url, headers=headers, auth=('admin', 'admin'))
 441         self.assertEqual(response.status_code, requests.codes.not_found)
 442         res = response.json()
 443         self.assertIn(
 444              {"error-type":"application", "error-tag":"data-missing",
 445               "error-message":"Request could not be completed because the relevant data model content does not exist"},
 446              res['errors']['error'])
 447
 448     def test_20_service_path_delete_rdm_check(self):
 449         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 450                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 451                "interface/SRG1-PP3-TXRX-mc-7"
 452                .format(self.restconf_baseurl))
 453         headers = {'content-type': 'application/json'}
 454         response = requests.request(
 455              "GET", url, headers=headers, auth=('admin', 'admin'))
 456         self.assertEqual(response.status_code, requests.codes.not_found)
 457         res = response.json()
 458         self.assertIn(
 459              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 460              res['errors']['error'])
 461
 462     def test_21_service_path_delete_rdm_check(self):
 463         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 464                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 465                "interface/SRG1-PP3-TXRX-nmc-7"
 466                .format(self.restconf_baseurl))
 467         headers = {'content-type': 'application/json'}
 468         response = requests.request(
 469              "GET", url, headers=headers, auth=('admin', 'admin'))
 470         self.assertEqual(response.status_code, requests.codes.not_found)
 471         res = response.json()
 472         self.assertIn(
 473              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 474              res['errors']['error'])
 475
 476     def test_22_service_path_delete_rdm_check(self):
 477         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 478                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 479                "roadm-connections/SRG1-PP3-TXRX-DEG1-TTP-TXRX-7"
 480                .format(self.restconf_baseurl))
 481         headers = {'content-type': 'application/json'}
 482         response = requests.request(
 483              "GET", url, headers=headers, auth=('admin', 'admin'))
 484         self.assertEqual(response.status_code, requests.codes.not_found)
 485         res = response.json()
 486         self.assertIn(
 487              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 488              res['errors']['error'])
 489
 490     def test_23_service_path_delete_xpdr_check(self):
 491         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 492                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 493                "interface/XPDR1-NETWORK1-7"
 494                .format(self.restconf_baseurl))
 495         headers = {'content-type': 'application/json'}
 496         response = requests.request(
 497              "GET", url, headers=headers, auth=('admin', 'admin'))
 498         self.assertEqual(response.status_code, requests.codes.not_found)
 499         res = response.json()
 500         self.assertIn(
 501              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 502              res['errors']['error'])
 503
 504     def test_24_service_path_delete_xpdr_check(self):
 505         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 506                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 507                "interface/XPDR1-NETWORK1-OTU"
 508                .format(self.restconf_baseurl))
 509         headers = {'content-type': 'application/json'}
 510         response = requests.request(
 511              "GET", url, headers=headers, auth=('admin', 'admin'))
 512         self.assertEqual(response.status_code, requests.codes.not_found)
 513         res = response.json()
 514         self.assertIn(
 515              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 516              res['errors']['error'])
 517
 518     def test_25_service_path_delete_xpdr_check(self):
 519         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 520                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 521                "interface/XPDR1-NETWORK1-ODU"
 522                .format(self.restconf_baseurl))
 523         headers = {'content-type': 'application/json'}
 524         response = requests.request(
 525              "GET", url, headers=headers, auth=('admin', 'admin'))
 526         self.assertEqual(response.status_code, requests.codes.not_found)
 527         res = response.json()
 528         self.assertIn(
 529              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 530              res['errors']['error'])
 531
 532     def test_26_service_path_delete_xpdr_check(self):
 533         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 534                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 535                "interface/XPDR1-CLIENT1-ETHERNET"
 536                .format(self.restconf_baseurl))
 537         headers = {'content-type': 'application/json'}
 538         response = requests.request(
 539              "GET", url, headers=headers, auth=('admin', 'admin'))
 540         self.assertEqual(response.status_code, requests.codes.not_found)
 541         res = response.json()
 542         self.assertIn(
 543              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 544              res['errors']['error'])
 545
 546     def test_27_service_path_delete_xpdr_check(self):
 547         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 548                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 549                "circuit-packs/1%2F0%2F1-PLUG-NET"
 550                .format(self.restconf_baseurl))
 551         headers = {'content-type': 'application/json'}
 552         response = requests.request(
 553             "GET", url, headers=headers, auth=('admin', 'admin'))
 554         self.assertEqual(response.status_code, requests.codes.ok)
 555         res = response.json()
 556         self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
 557
 558     def test_28_service_path_delete_xpdr_check(self):
 559         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 560                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 561                "circuit-packs/1%2F0%2F1-PLUG-CLIENT"
 562                .format(self.restconf_baseurl))
 563         headers = {'content-type': 'application/json'}
 564         response = requests.request(
 565             "GET", url, headers=headers, auth=('admin', 'admin'))
 566         self.assertEqual(response.status_code, requests.codes.ok)
 567         res = response.json()
 568         self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
 569
 570     def test_29_rdm_device_disconnected(self):
 571         url = ("{}/config/network-topology:"
 572                "network-topology/topology/topology-netconf/node/ROADM-A1"
 573                .format(self.restconf_baseurl))
 574         headers = {'content-type': 'application/json'}
 575         response = requests.request(
 576               "DELETE", url, headers=headers,
 577               auth=('admin', 'admin'))
 578         self.assertEqual(response.status_code, requests.codes.ok)
 579         time.sleep(20)
 580
 581     def test_30_xpdr_device_disconnected(self):
 582         url = ("{}/config/network-topology:"
 583                 "network-topology/topology/topology-netconf/node/XPDR-A1"
 584                .format(self.restconf_baseurl))
 585         headers = {'content-type': 'application/json'}
 586         response = requests.request(
 587              "DELETE", url, headers=headers,
 588              auth=('admin', 'admin'))
 589         self.assertEqual(response.status_code, requests.codes.ok)
 590         time.sleep(20)
 591
 592
 593 if __name__ == "__main__":
 594     unittest.main(verbosity=2, failfast=True)