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