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         self.assertDictContainsSubset({'name': 'DEG1-TTP-TXRX-nmc-7', 'administrative-state': 'inService',
 192                                        'supporting-circuit-pack-name': '1/0',
 193                                        'type': 'org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint',
 194                                        'supporting-port': 'L1'}, res['interface'][0])
 195         self.assertDictEqual(
 196             {u'frequency': 195.8, u'width': 40},
 197             res['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp'])
 198
 199     def test_07_service_path_create_rdm_check(self):
 200         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 201                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 202                "interface/DEG1-TTP-TXRX-mc-7"
 203                .format(self.restconf_baseurl))
 204         headers = {'content-type': 'application/json'}
 205         response = requests.request(
 206             "GET", url, headers=headers, auth=('admin', 'admin'))
 207         self.assertEqual(response.status_code, requests.codes.ok)
 208         res = response.json()
 209         self.assertDictContainsSubset({'name': 'DEG1-TTP-TXRX-mc-7', 'administrative-state': 'inService',
 210                                        'supporting-circuit-pack-name': '1/0',
 211                                        'type': 'org-openroadm-interfaces:mediaChannelTrailTerminationPoint',
 212                                        'supporting-port': 'L1'}, res['interface'][0])
 213         self.assertDictEqual(
 214             {u'min-freq': 195.775, u'max-freq': 195.825},
 215             res['interface'][0]['org-openroadm-media-channel-interfaces:mc-ttp'])
 216
 217     def test_08_service_path_create_rdm_check(self):
 218         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 219                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 220                "interface/SRG1-PP3-TXRX-nmc-7"
 221                .format(self.restconf_baseurl))
 222         headers = {'content-type': 'application/json'}
 223         response = requests.request(
 224             "GET", url, headers=headers, auth=('admin', 'admin'))
 225         self.assertEqual(response.status_code, requests.codes.ok)
 226         res = response.json()
 227         self.assertDictContainsSubset({'name': 'SRG1-PP3-TXRX-nmc-7', 'administrative-state': 'inService',
 228                                        'supporting-circuit-pack-name': '3/0',
 229                                        'type': 'org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint',
 230                                        'supporting-port': 'C3'}, res['interface'][0])
 231         self.assertDictEqual(
 232             {u'frequency': 195.8, u'width': 40},
 233             res['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp'])
 234
 235     # -mc supporting interfaces must not be created for SRG, only degrees
 236     def test_09_service_path_create_rdm_check(self):
 237         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 238                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 239                "interface/SRG1-PP3-TXRX-mc-7"
 240                .format(self.restconf_baseurl))
 241         headers = {'content-type': 'application/json'}
 242         response = requests.request(
 243             "GET", url, headers=headers, auth=('admin', 'admin'))
 244         self.assertEqual(response.status_code, requests.codes.not_found)
 245         res = response.json()
 246         self.assertIn(
 247             {"error-type": "application", "error-tag": "data-missing",
 248              "error-message": "Request could not be completed because the relevant data model content does not exist"},
 249             res['errors']['error'])
 250
 251     def test_10_service_path_create_rdm_check(self):
 252         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 253                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 254                "roadm-connections/SRG1-PP3-TXRX-DEG1-TTP-TXRX-7"
 255                .format(self.restconf_baseurl))
 256         headers = {'content-type': 'application/json'}
 257         response = requests.request(
 258             "GET", url, headers=headers, auth=('admin', 'admin'))
 259         self.assertEqual(response.status_code, requests.codes.ok)
 260         res = response.json()
 261         self.assertDictContainsSubset(
 262             {'connection-name': 'SRG1-PP3-TXRX-DEG1-TTP-TXRX-7',
 263              'opticalControlMode': 'off'},
 264             res['roadm-connections'][0])
 265         self.assertDictEqual(
 266             {'src-if': 'SRG1-PP3-TXRX-nmc-7'},
 267             res['roadm-connections'][0]['source'])
 268         self.assertDictEqual(
 269             {'dst-if': 'DEG1-TTP-TXRX-nmc-7'},
 270             res['roadm-connections'][0]['destination'])
 271
 272     def test_11_service_path_create_xpdr_check(self):
 273         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 274                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 275                "interface/XPDR1-NETWORK1-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         self.assertDictContainsSubset(
 283             {'name': 'XPDR1-NETWORK1-7', 'administrative-state': 'inService',
 284              'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 285              'type': 'org-openroadm-interfaces:opticalChannel',
 286              'supporting-port': '1'},
 287             res['interface'][0])
 288         self.assertDictEqual(
 289             {u'rate': u'org-openroadm-common-types:R100G',
 290              u'transmit-power': -5,
 291              u'frequency': 195.8},
 292             res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
 293
 294     def test_12_service_path_create_xpdr_check(self):
 295         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 296                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 297                "interface/XPDR1-NETWORK1-OTU"
 298                .format(self.restconf_baseurl))
 299         headers = {'content-type': 'application/json'}
 300         response = requests.request(
 301             "GET", url, headers=headers, auth=('admin', 'admin'))
 302         self.assertEqual(response.status_code, requests.codes.ok)
 303         res = response.json()
 304         self.assertDictContainsSubset(
 305             {'name': 'XPDR1-NETWORK1-OTU', '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         self.assertDictEqual(
 312             {u'rate': u'org-openroadm-otn-common-types:OTU4',
 313              u'fec': u'scfec'},
 314             res['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
 315
 316     def test_13_service_path_create_xpdr_check(self):
 317         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 318                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 319                "interface/XPDR1-NETWORK1-ODU"
 320                .format(self.restconf_baseurl))
 321         headers = {'content-type': 'application/json'}
 322         response = requests.request(
 323             "GET", url, headers=headers, auth=('admin', 'admin'))
 324         self.assertEqual(response.status_code, requests.codes.ok)
 325         res = response.json()
 326         self.assertDictContainsSubset(
 327             {'name': 'XPDR1-NETWORK1-ODU', 'administrative-state': 'inService',
 328              'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 329              'type': 'org-openroadm-interfaces:otnOdu',
 330              'supporting-port': '1',
 331              'supporting-interface': 'XPDR1-NETWORK1-OTU'},
 332             res['interface'][0])
 333         self.assertDictContainsSubset(
 334             {'rate': 'org-openroadm-otn-common-types:ODU4',
 335              u'monitoring-mode': u'terminated'},
 336             res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'])
 337         self.assertDictEqual({u'exp-payload-type': u'07', u'payload-type': u'07'},
 338                              res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
 339
 340     def test_14_service_path_create_xpdr_check(self):
 341         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 342                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 343                "interface/XPDR1-CLIENT1-ETHERNET"
 344                .format(self.restconf_baseurl))
 345         headers = {'content-type': 'application/json'}
 346         response = requests.request(
 347             "GET", url, headers=headers, auth=('admin', 'admin'))
 348         self.assertEqual(response.status_code, requests.codes.ok)
 349         res = response.json()
 350         self.assertDictContainsSubset(
 351             {'name': 'XPDR1-CLIENT1-ETHERNET', 'administrative-state': 'inService',
 352              'supporting-circuit-pack-name': '1/0/1-PLUG-CLIENT',
 353              'type': 'org-openroadm-interfaces:ethernetCsmacd',
 354              'supporting-port': 'C1'},
 355             res['interface'][0])
 356         self.assertDictEqual(
 357             {u'fec': u'off', u'speed': 100000},
 358             res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
 359
 360     def test_15_service_path_create_xpdr_check(self):
 361         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 362                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 363                "circuit-packs/1%2F0%2F1-PLUG-NET"
 364                .format(self.restconf_baseurl))
 365         headers = {'content-type': 'application/json'}
 366         response = requests.request(
 367             "GET", url, headers=headers, auth=('admin', 'admin'))
 368         self.assertEqual(response.status_code, requests.codes.ok)
 369         res = response.json()
 370         self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
 371
 372     def test_16_service_path_create_xpdr_check(self):
 373         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 374                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 375                "circuit-packs/1%2F0%2F1-PLUG-CLIENT"
 376                .format(self.restconf_baseurl))
 377         headers = {'content-type': 'application/json'}
 378         response = requests.request(
 379             "GET", url, headers=headers, auth=('admin', 'admin'))
 380         self.assertEqual(response.status_code, requests.codes.ok)
 381         res = response.json()
 382         self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
 383
 384     def test_17_service_path_delete(self):
 385         url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl)
 386         data = {"renderer:input": {
 387             "renderer:service-name": "service_test",
 388             "renderer:wave-number": "7",
 389             "renderer:operation": "delete",
 390             "renderer:nodes": [
 391                 {"renderer:node-id": "ROADM-A1",
 392                  "renderer:src-tp": "SRG1-PP3-TXRX",
 393                  "renderer:dest-tp": "DEG1-TTP-TXRX"},
 394                 {"renderer:node-id": "XPDR-A1",
 395                  "renderer:src-tp": "XPDR1-CLIENT1",
 396                  "renderer:dest-tp": "XPDR1-NETWORK1"}]}}
 397         headers = {'content-type': 'application/json'}
 398         response = requests.request(
 399             "POST", url, data=json.dumps(data),
 400             headers=headers, auth=('admin', 'admin'))
 401         self.assertEqual(response.status_code, requests.codes.ok)
 402         self.assertEqual(response.json(), {
 403             'output': {'result': 'Request processed', 'success': True}})
 404
 405     def test_18_service_path_delete_rdm_check(self):
 406         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 407                "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 408                "interface/DEG1-TTP-TXRX-mc-7"
 409                .format(self.restconf_baseurl))
 410         headers = {'content-type': 'application/json'}
 411         response = requests.request(
 412             "GET", url, headers=headers, auth=('admin', 'admin'))
 413         self.assertEqual(response.status_code, requests.codes.not_found)
 414         res = response.json()
 415         self.assertIn(
 416             {"error-type": "application", "error-tag": "data-missing",
 417              "error-message": "Request could not be completed because the relevant data model content does not exist"},
 418             res['errors']['error'])
 419
 420     def test_19_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-nmc-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_20_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/SRG1-PP3-TXRX-mc-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_21_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-nmc-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_22_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                "roadm-connections/SRG1-PP3-TXRX-DEG1-TTP-TXRX-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_23_service_path_delete_xpdr_check(self):
 481         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 482                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 483                "interface/XPDR1-NETWORK1-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_24_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-OTU"
 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_25_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-ODU"
 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_26_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-CLIENT1-ETHERNET"
 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_27_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                "circuit-packs/1%2F0%2F1-PLUG-NET"
 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.ok)
 549         res = response.json()
 550         self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
 551
 552     def test_28_service_path_delete_xpdr_check(self):
 553         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 554                "node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 555                "circuit-packs/1%2F0%2F1-PLUG-CLIENT"
 556                .format(self.restconf_baseurl))
 557         headers = {'content-type': 'application/json'}
 558         response = requests.request(
 559             "GET", url, headers=headers, auth=('admin', 'admin'))
 560         self.assertEqual(response.status_code, requests.codes.ok)
 561         res = response.json()
 562         self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
 563
 564     def test_29_rdm_device_disconnected(self):
 565         url = ("{}/config/network-topology:"
 566                "network-topology/topology/topology-netconf/node/ROADM-A1"
 567                .format(self.restconf_baseurl))
 568         headers = {'content-type': 'application/json'}
 569         response = requests.request(
 570             "DELETE", url, headers=headers,
 571             auth=('admin', 'admin'))
 572         self.assertEqual(response.status_code, requests.codes.ok)
 573         time.sleep(20)
 574
 575     def test_30_xpdr_device_disconnected(self):
 576         url = ("{}/config/network-topology:"
 577                "network-topology/topology/topology-netconf/node/XPDR-A1"
 578                .format(self.restconf_baseurl))
 579         headers = {'content-type': 'application/json'}
 580         response = requests.request(
 581             "DELETE", url, headers=headers,
 582             auth=('admin', 'admin'))
 583         self.assertEqual(response.status_code, requests.codes.ok)
 584         time.sleep(20)
 585
 586
 587 if __name__ == "__main__":
 588     unittest.main(verbosity=2, failfast=True)