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