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'},
 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'},
 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'},
 162              res['nodes'][0]['mapping'])
 163         self.assertIn(
 164              {'supporting-port': 'C1',
 165               'supporting-circuit-pack-name': '1/0/1-PLUG-CLIENT',
 166               'logical-connection-point': 'XPDR1-CLIENT1'},
 167              res['nodes'][0]['mapping'])
 168
 169     def test_05_service_path_create(self):
 170         url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl)
 171         data = {"renderer:input": {
 172              "renderer:service-name": "service_test",
 173              "renderer:wave-number": "7",
 174              "renderer:modulation-format": "qpsk",
 175              "renderer:operation": "create",
 176              "renderer:nodes": [
 177                  {"renderer:node-id": "ROADM-A1",
 178                   "renderer:src-tp": "SRG1-PP3-TXRX",
 179                   "renderer:dest-tp": "DEG1-TTP-TXRX"},
 180                  {"renderer:node-id": "XPDR-A1",
 181                   "renderer:src-tp": "XPDR1-CLIENT1",
 182                   "renderer:dest-tp": "XPDR1-NETWORK1"}]}}
 183         headers = {'content-type': 'application/json'}
 184         response = requests.request(
 185              "POST", url, data=json.dumps(data),
 186              headers=headers, auth=('admin', 'admin'))
 187         self.assertEqual(response.status_code, requests.codes.ok)
 188         res = response.json()
 189         self.assertIn('Roadm-connection successfully created for nodes: ROADM-A1', res["output"]["result"])
 190
 191     def test_06_service_path_create_rdm_check(self):
 192         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 193                 "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 194                 "interface/DEG1-TTP-TXRX-nmc"
 195                 .format(self.restconf_baseurl))
 196         headers = {'content-type': 'application/json'}
 197         response = requests.request(
 198              "GET", url, headers=headers, auth=('admin', 'admin'))
 199         self.assertEqual(response.status_code, requests.codes.ok)
 200         res = response.json()
 201         self.assertDictContainsSubset({'name': 'DEG1-TTP-TXRX-nmc', 'administrative-state': 'inService',
 202               'supporting-circuit-pack-name': '1/0',
 203               'type': 'org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint',
 204               'supporting-port': 'L1'}, res['interface'][0])
 205         self.assertDictEqual(
 206              {u'frequency': 195.79999999999998, u'width': 1531.11},
 207              res['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp'])
 208
 209     def test_07_service_path_create_rdm_check(self):
 210         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 211                 "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 212                 "interface/DEG1-TTP-TXRX-mc"
 213                 .format(self.restconf_baseurl))
 214         headers = {'content-type': 'application/json'}
 215         response = requests.request(
 216              "GET", url, headers=headers, auth=('admin', 'admin'))
 217         self.assertEqual(response.status_code, requests.codes.ok)
 218         res = response.json()
 219         self.assertDictContainsSubset({'name': 'DEG1-TTP-TXRX-mc', 'administrative-state': 'inService',
 220               'supporting-circuit-pack-name': '1/0',
 221               'type': 'org-openroadm-interfaces:mediaChannelTrailTerminationPoint',
 222               'supporting-port': 'L1'}, res['interface'][0])
 223         self.assertDictEqual(
 224              {u'min-freq': 195.77499999999998, u'max-freq': 195.825},
 225              res['interface'][0]['org-openroadm-media-channel-interfaces:mc-ttp'])
 226
 227
 228     def test_08_service_path_create_rdm_check(self):
 229         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 230                 "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 231                 "interface/SRG1-PP3-TXRX-nmc"
 232                 .format(self.restconf_baseurl))
 233         headers = {'content-type': 'application/json'}
 234         response = requests.request(
 235              "GET", url, headers=headers, auth=('admin', 'admin'))
 236         self.assertEqual(response.status_code, requests.codes.ok)
 237         res = response.json()
 238         self.assertDictContainsSubset({'name': 'SRG1-PP3-TXRX-nmc', 'administrative-state': 'inService',
 239               'supporting-circuit-pack-name': '3/0',
 240               'type': 'org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint',
 241               'supporting-port': 'C3'}, res['interface'][0])
 242         self.assertDictEqual(
 243              {u'frequency': 195.79999999999998, u'width': 1531.11},
 244              res['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp'])
 245
 246     def test_09_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                 "interface/SRG1-PP3-TXRX-mc"
 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({'name': 'SRG1-PP3-TXRX-mc', 'administrative-state': 'inService',
 257               'supporting-circuit-pack-name': '3/0',
 258               'type': 'org-openroadm-interfaces:mediaChannelTrailTerminationPoint',
 259               'supporting-port': 'C3'}, res['interface'][0])
 260         self.assertDictEqual(
 261              {u'min-freq': 195.77499999999998, u'max-freq': 195.825},
 262              res['interface'][0]['org-openroadm-media-channel-interfaces:mc-ttp'])
 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'},
 280              res['roadm-connections'][0]['source'])
 281         self.assertDictEqual(
 282              {'dst-if': 'DEG1-TTP-TXRX-nmc'},
 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': 195.79999999999998},
 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              {'speed': 100000,
 371               'mtu': 9000,
 372               'auto-negotiation': 'enabled',
 373               'duplex': 'full',
 374               'fec': 'off'},
 375              res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
 376
 377     def test_15_service_path_create_xpdr_check(self):
 378         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 379                 "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 380                 "circuit-packs/1%2F0%2F1-PLUG-NET"
 381                 .format(self.restconf_baseurl))
 382         headers = {'content-type': 'application/json'}
 383         response = requests.request(
 384              "GET", url, headers=headers, auth=('admin', 'admin'))
 385         self.assertEqual(response.status_code, requests.codes.ok)
 386         res = response.json()
 387         self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
 388
 389     def test_16_service_path_create_xpdr_check(self):
 390         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 391                 "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 392                 "circuit-packs/1%2F0%2F1-PLUG-CLIENT"
 393                 .format(self.restconf_baseurl))
 394         headers = {'content-type': 'application/json'}
 395         response = requests.request(
 396              "GET", url, headers=headers, auth=('admin', 'admin'))
 397         self.assertEqual(response.status_code, requests.codes.ok)
 398         res = response.json()
 399         self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
 400
 401     def test_17_service_path_delete(self):
 402         url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl)
 403         data = {"renderer:input": {
 404              "renderer:service-name": "service_test",
 405              "renderer:wave-number": "7",
 406              "renderer:operation": "delete",
 407              "renderer:nodes": [
 408                  {"renderer:node-id": "ROADM-A1",
 409                   "renderer:src-tp": "SRG1-PP3-TXRX",
 410                   "renderer:dest-tp": "DEG1-TTP-TXRX"},
 411                  {"renderer:node-id": "XPDR-A1",
 412                   "renderer:src-tp": "XPDR1-CLIENT1",
 413                   "renderer:dest-tp": "XPDR1-NETWORK1"}]}}
 414         headers = {'content-type': 'application/json'}
 415         response = requests.request(
 416              "POST", url, data=json.dumps(data),
 417              headers=headers, auth=('admin', 'admin'))
 418         self.assertEqual(response.status_code, requests.codes.ok)
 419         self.assertEqual(response.json(), {
 420              'output': {'result': 'Request processed', 'success': True}})
 421
 422     def test_18_service_path_delete_rdm_check(self):
 423         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 424                 "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 425                 "interface/DEG1-TTP-TXRX-mc"
 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",
 434               "error-message":"Request could not be completed because the relevant data model content does not exist"},
 435              res['errors']['error'])
 436
 437     def test_19_service_path_delete_rdm_check(self):
 438         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 439                 "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 440                 "interface/DEG1-TTP-TXRX-nmc"
 441                 .format(self.restconf_baseurl))
 442         headers = {'content-type': 'application/json'}
 443         response = requests.request(
 444              "GET", url, headers=headers, auth=('admin', 'admin'))
 445         self.assertEqual(response.status_code, requests.codes.not_found)
 446         res = response.json()
 447         self.assertIn(
 448              {"error-type":"application", "error-tag":"data-missing",
 449               "error-message":"Request could not be completed because the relevant data model content does not exist"},
 450              res['errors']['error'])
 451
 452     def test_20_service_path_delete_rdm_check(self):
 453         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 454                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 455                "interface/SRG1-PP3-TXRX-mc"
 456                .format(self.restconf_baseurl))
 457         headers = {'content-type': 'application/json'}
 458         response = requests.request(
 459              "GET", url, headers=headers, auth=('admin', 'admin'))
 460         self.assertEqual(response.status_code, requests.codes.not_found)
 461         res = response.json()
 462         self.assertIn(
 463              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 464              res['errors']['error'])
 465
 466     def test_21_service_path_delete_rdm_check(self):
 467         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 468                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 469                "interface/SRG1-PP3-TXRX-nmc"
 470                .format(self.restconf_baseurl))
 471         headers = {'content-type': 'application/json'}
 472         response = requests.request(
 473              "GET", url, headers=headers, auth=('admin', 'admin'))
 474         self.assertEqual(response.status_code, requests.codes.not_found)
 475         res = response.json()
 476         self.assertIn(
 477              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 478              res['errors']['error'])
 479
 480     def test_22_service_path_delete_rdm_check(self):
 481         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 482                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 483                "roadm-connections/SRG1-PP3-TXRX-DEG1-TTP-TXRX-7"
 484                .format(self.restconf_baseurl))
 485         headers = {'content-type': 'application/json'}
 486         response = requests.request(
 487              "GET", url, headers=headers, auth=('admin', 'admin'))
 488         self.assertEqual(response.status_code, requests.codes.not_found)
 489         res = response.json()
 490         self.assertIn(
 491              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 492              res['errors']['error'])
 493
 494     def test_23_service_path_delete_xpdr_check(self):
 495         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 496                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 497                "interface/XPDR1-NETWORK1-7"
 498                .format(self.restconf_baseurl))
 499         headers = {'content-type': 'application/json'}
 500         response = requests.request(
 501              "GET", url, headers=headers, auth=('admin', 'admin'))
 502         self.assertEqual(response.status_code, requests.codes.not_found)
 503         res = response.json()
 504         self.assertIn(
 505              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 506              res['errors']['error'])
 507
 508     def test_24_service_path_delete_xpdr_check(self):
 509         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 510                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 511                "interface/XPDR1-NETWORK1-OTU"
 512                .format(self.restconf_baseurl))
 513         headers = {'content-type': 'application/json'}
 514         response = requests.request(
 515              "GET", url, headers=headers, auth=('admin', 'admin'))
 516         self.assertEqual(response.status_code, requests.codes.not_found)
 517         res = response.json()
 518         self.assertIn(
 519              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 520              res['errors']['error'])
 521
 522     def test_25_service_path_delete_xpdr_check(self):
 523         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 524                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 525                "interface/XPDR1-NETWORK1-ODU"
 526                .format(self.restconf_baseurl))
 527         headers = {'content-type': 'application/json'}
 528         response = requests.request(
 529              "GET", url, headers=headers, auth=('admin', 'admin'))
 530         self.assertEqual(response.status_code, requests.codes.not_found)
 531         res = response.json()
 532         self.assertIn(
 533              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 534              res['errors']['error'])
 535
 536     def test_26_service_path_delete_xpdr_check(self):
 537         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 538                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 539                "interface/XPDR1-CLIENT1-ETHERNET"
 540                .format(self.restconf_baseurl))
 541         headers = {'content-type': 'application/json'}
 542         response = requests.request(
 543              "GET", url, headers=headers, auth=('admin', 'admin'))
 544         self.assertEqual(response.status_code, requests.codes.not_found)
 545         res = response.json()
 546         self.assertIn(
 547              {"error-type":"application", "error-tag":"data-missing", "error-message":"Request could not be completed because the relevant data model content does not exist"},
 548              res['errors']['error'])
 549
 550     def test_27_service_path_delete_xpdr_check(self):
 551         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 552                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 553                "circuit-packs/1%2F0%2F1-PLUG-NET"
 554                .format(self.restconf_baseurl))
 555         headers = {'content-type': 'application/json'}
 556         response = requests.request(
 557             "GET", url, headers=headers, auth=('admin', 'admin'))
 558         self.assertEqual(response.status_code, requests.codes.ok)
 559         res = response.json()
 560         self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
 561
 562     def test_28_service_path_delete_xpdr_check(self):
 563         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 564                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 565                "circuit-packs/1%2F0%2F1-PLUG-CLIENT"
 566                .format(self.restconf_baseurl))
 567         headers = {'content-type': 'application/json'}
 568         response = requests.request(
 569             "GET", url, headers=headers, auth=('admin', 'admin'))
 570         self.assertEqual(response.status_code, requests.codes.ok)
 571         res = response.json()
 572         self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
 573
 574     def test_29_rdm_device_disconnected(self):
 575         url = ("{}/config/network-topology:"
 576                "network-topology/topology/topology-netconf/node/ROADM-A1"
 577                .format(self.restconf_baseurl))
 578         headers = {'content-type': 'application/json'}
 579         response = requests.request(
 580               "DELETE", url, headers=headers,
 581               auth=('admin', 'admin'))
 582         self.assertEqual(response.status_code, requests.codes.ok)
 583         time.sleep(20)
 584
 585     def test_30_xpdr_device_disconnected(self):
 586         url = ("{}/config/network-topology:"
 587                 "network-topology/topology/topology-netconf/node/XPDR-A1"
 588                .format(self.restconf_baseurl))
 589         headers = {'content-type': 'application/json'}
 590         response = requests.request(
 591              "DELETE", url, headers=headers,
 592              auth=('admin', 'admin'))
 593         self.assertEqual(response.status_code, requests.codes.ok)
 594         time.sleep(20)
 595
 596
 597 if __name__ == "__main__":
 598     unittest.main(verbosity=2, failfast=True)