tests/transportpce_tests/2.2.1/test_renderer_service_path_nominal.py

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