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