tests/transportpce_tests/2.2.1/test_otn_renderer.py

   1 #!/usr/bin/env python
   2
   3 ##############################################################################
   4 # Copyright (c) 2020 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 json
  13 import os
  14 import psutil
  15 import requests
  16 import signal
  17 import shutil
  18 import subprocess
  19 import time
  20 import unittest
  21 import logging
  22 import test_utils
  23
  24
  25 class TransportPCEtesting(unittest.TestCase):
  26
  27     honeynode_process1 = None
  28     odl_process = None
  29     restconf_baseurl = "http://localhost:8181/restconf"
  30
  31     @classmethod
  32     def setUpClass(cls):
  33         print("starting honeynode1...")
  34         cls.honeynode_process1 = test_utils.start_spdra_honeynode()
  35         time.sleep(30)
  36
  37         print("starting opendaylight...")
  38         cls.odl_process = test_utils.start_tpce()
  39         time.sleep(60)
  40         print("opendaylight started")
  41
  42     @classmethod
  43     def tearDownClass(cls):
  44         for child in psutil.Process(cls.odl_process.pid).children():
  45             child.send_signal(signal.SIGINT)
  46             child.wait()
  47         cls.odl_process.send_signal(signal.SIGINT)
  48         cls.odl_process.wait()
  49         for child in psutil.Process(cls.honeynode_process1.pid).children():
  50             child.send_signal(signal.SIGINT)
  51             child.wait()
  52         cls.honeynode_process1.send_signal(signal.SIGINT)
  53         cls.honeynode_process1.wait()
  54
  55     def setUp(self):
  56         time.sleep(5)
  57
  58     def test_01_connect_SPDR_SA1(self):
  59         url = ("{}/config/network-topology:"
  60                "network-topology/topology/topology-netconf/node/SPDR-SA1"
  61                .format(self.restconf_baseurl))
  62         data = {"node": [{
  63             "node-id": "SPDR-SA1",
  64             "netconf-node-topology:username": "admin",
  65             "netconf-node-topology:password": "admin",
  66             "netconf-node-topology:host": "127.0.0.1",
  67             "netconf-node-topology:port": "17845",
  68             "netconf-node-topology:tcp-only": "false",
  69             "netconf-node-topology:pass-through": {}}]}
  70         headers = {'content-type': 'application/json'}
  71         response = requests.request(
  72             "PUT", url, data=json.dumps(data), headers=headers,
  73             auth=('admin', 'admin'))
  74         self.assertEqual(response.status_code, requests.codes.created)
  75         time.sleep(10)
  76         url = ("{}/operational/network-topology:"
  77                "network-topology/topology/topology-netconf/node/SPDR-SA1"
  78                .format(self.restconf_baseurl))
  79         response = requests.request(
  80             "GET", url, headers=headers, auth=('admin', 'admin'))
  81         self.assertEqual(response.status_code, requests.codes.ok)
  82         res = response.json()
  83         self.assertEqual(
  84             res['node'][0]['netconf-node-topology:connection-status'],
  85             'connected')
  86
  87     def test_02_get_portmapping_CLIENT1(self):
  88         url = ("{}/config/transportpce-portmapping:network/"
  89                "nodes/SPDR-SA1/mapping/XPDR1-CLIENT1"
  90                .format(self.restconf_baseurl))
  91         headers = {'content-type': 'application/json'}
  92         response = requests.request(
  93             "GET", url, headers=headers, auth=('admin', 'admin'))
  94         self.assertEqual(response.status_code, requests.codes.ok)
  95         res = response.json()
  96         self.assertIn(
  97             {'supported-interface-capability': [
  98                 'org-openroadm-port-types:if-10GE-ODU2e',
  99                 'org-openroadm-port-types:if-10GE-ODU2',
 100                 'org-openroadm-port-types:if-10GE'],
 101              'supporting-port': 'CP1-SFP4-P1',
 102              'supporting-circuit-pack-name': 'CP1-SFP4',
 103              'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
 104              'port-qual': 'xpdr-client'},
 105             res['mapping'])
 106
 107     def test_03_get_portmapping_NETWORK1(self):
 108         url = ("{}/config/transportpce-portmapping:network/"
 109                "nodes/SPDR-SA1/mapping/XPDR1-NETWORK1"
 110                .format(self.restconf_baseurl))
 111         headers = {'content-type': 'application/json'}
 112         response = requests.request(
 113             "GET", url, headers=headers, auth=('admin', 'admin'))
 114         self.assertEqual(response.status_code, requests.codes.ok)
 115         res = response.json()
 116         self.assertIn(
 117             {"logical-connection-point": "XPDR1-NETWORK1",
 118              "supporting-port": "CP1-CFP0-P1",
 119              "supported-interface-capability": [
 120                  "org-openroadm-port-types:if-OCH-OTU4-ODU4"
 121              ],
 122                 "port-direction": "bidirectional",
 123                 "port-qual": "xpdr-network",
 124                 "supporting-circuit-pack-name": "CP1-CFP0",
 125                 "xponder-type": "mpdr"},
 126             res['mapping'])
 127
 128     def test_04_service_path_create_OCH_OTU4(self):
 129         url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl)
 130         data = {"renderer:input": {
 131             "service-name": "service_ODU4",
 132             "wave-number": "1",
 133             "modulation-format": "qpsk",
 134             "operation": "create",
 135             "nodes": [
 136                 {"node-id": "SPDR-SA1",
 137                  "dest-tp": "XPDR1-NETWORK1"}]}}
 138         headers = {'content-type': 'application/json'}
 139         response = requests.request(
 140             "POST", url, data=json.dumps(data),
 141             headers=headers, auth=('admin', 'admin'))
 142         time.sleep(3)
 143         self.assertEqual(response.status_code, requests.codes.ok)
 144         res = response.json()
 145         self.assertIn('Roadm-connection successfully created for nodes: ', res["output"]["result"])
 146         self.assertTrue(res["output"]["success"])
 147         self.assertIn(
 148             {'node-id': 'SPDR-SA1',
 149              'otu-interface-id': ['XPDR1-NETWORK1-OTU'],
 150              'och-interface-id': ['XPDR1-NETWORK1-1']}, res["output"]['node-interface'])
 151
 152     def test_05_get_portmapping_NETWORK1(self):
 153         url = ("{}/config/transportpce-portmapping:network/"
 154                "nodes/SPDR-SA1/mapping/XPDR1-NETWORK1"
 155                .format(self.restconf_baseurl))
 156         headers = {'content-type': 'application/json'}
 157         response = requests.request(
 158             "GET", url, headers=headers, auth=('admin', 'admin'))
 159         self.assertEqual(response.status_code, requests.codes.ok)
 160         res = response.json()
 161         self.assertIn(
 162             {"logical-connection-point": "XPDR1-NETWORK1",
 163              "supporting-port": "CP1-CFP0-P1",
 164              "supported-interface-capability": [
 165                  "org-openroadm-port-types:if-OCH-OTU4-ODU4"
 166              ],
 167                 "port-direction": "bidirectional",
 168                 "port-qual": "xpdr-network",
 169                 "supporting-circuit-pack-name": "CP1-CFP0",
 170                 "xponder-type": "mpdr"},
 171             res['mapping'])
 172
 173     def test_06_check_interface_och(self):
 174         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 175                "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 176                "interface/XPDR1-NETWORK1-1"
 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         self.assertDictContainsSubset({'name': 'XPDR1-NETWORK1-1', 'administrative-state': 'inService',
 184                                        'supporting-circuit-pack-name': 'CP1-CFP0',
 185                                        'type': 'org-openroadm-interfaces:opticalChannel',
 186                                        'supporting-port': 'CP1-CFP0-P1'}, res['interface'][0])
 187         self.assertDictEqual(
 188             {u'frequency': 196.1, u'rate': u'org-openroadm-common-types:R100G',
 189              u'transmit-power': -5},
 190             res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
 191
 192     def test_07_check_interface_OTU(self):
 193         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 194                "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 195                "interface/XPDR1-NETWORK1-OTU"
 196                .format(self.restconf_baseurl))
 197         headers = {'content-type': 'application/json'}
 198         response = requests.request(
 199             "GET", url, headers=headers, auth=('admin', 'admin'))
 200         self.assertEqual(response.status_code, requests.codes.ok)
 201         res = response.json()
 202         self.assertDictContainsSubset({'name': 'XPDR1-NETWORK1-OTU', 'administrative-state': 'inService',
 203                                        'supporting-circuit-pack-name': 'CP1-CFP0', 'supporting-interface': 'XPDR1-NETWORK1-1',
 204                                        'type': 'org-openroadm-interfaces:otnOtu',
 205                                        'supporting-port': 'CP1-CFP0-P1'}, res['interface'][0])
 206         self.assertDictEqual(
 207             {u'rate': u'org-openroadm-otn-common-types:OTU4',
 208              u'fec': u'scfec'},
 209             res['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
 210
 211     def test_08_otn_service_path_create_ODU4(self):
 212         url = "{}/operations/transportpce-device-renderer:otn-service-path".format(self.restconf_baseurl)
 213         data = {"renderer:input": {
 214             "service-name": "service_ODU4",
 215             "operation": "create",
 216             "service-rate": "100G",
 217             "service-type": "ODU",
 218             "nodes": [
 219                 {"node-id": "SPDR-SA1",
 220                  "network-tp": "XPDR1-NETWORK1"}]}}
 221         headers = {'content-type': 'application/json'}
 222         response = requests.request(
 223             "POST", url, data=json.dumps(data),
 224             headers=headers, auth=('admin', 'admin'))
 225         time.sleep(3)
 226         self.assertEqual(response.status_code, requests.codes.ok)
 227         res = response.json()
 228         self.assertIn('Otn Service path was set up successfully for node :SPDR-SA1', res["output"]["result"])
 229         self.assertTrue(res["output"]["success"])
 230         self.assertIn(
 231             {'node-id': 'SPDR-SA1',
 232              'odu-interface-id': ['XPDR1-NETWORK1-ODU4']}, res["output"]['node-interface'])
 233
 234     def test_09_get_portmapping_NETWORK1(self):
 235         url = ("{}/config/transportpce-portmapping:network/"
 236                "nodes/SPDR-SA1/mapping/XPDR1-NETWORK1"
 237                .format(self.restconf_baseurl))
 238         headers = {'content-type': 'application/json'}
 239         response = requests.request(
 240             "GET", url, headers=headers, auth=('admin', 'admin'))
 241         self.assertEqual(response.status_code, requests.codes.ok)
 242         res = response.json()
 243         self.assertIn(
 244             {"logical-connection-point": "XPDR1-NETWORK1",
 245             "supporting-port": "CP1-CFP0-P1",
 246             "supported-interface-capability": [
 247                 "org-openroadm-port-types:if-OCH-OTU4-ODU4"
 248             ],
 249             "port-direction": "bidirectional",
 250             "port-qual": "xpdr-network",
 251             "supporting-circuit-pack-name": "CP1-CFP0",
 252             "xponder-type": "mpdr",
 253             "supporting-odu4": "XPDR1-NETWORK1-ODU4"},
 254             res['mapping'])
 255
 256     def test_10_check_interface_ODU4(self):
 257         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 258                "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 259                "interface/XPDR1-NETWORK1-ODU4"
 260                .format(self.restconf_baseurl))
 261         headers = {'content-type': 'application/json'}
 262         response = requests.request(
 263             "GET", url, headers=headers, auth=('admin', 'admin'))
 264         self.assertEqual(response.status_code, requests.codes.ok)
 265         res = response.json()
 266         self.assertDictContainsSubset({'name': 'XPDR1-NETWORK1-ODU4', 'administrative-state': 'inService',
 267                                        'supporting-circuit-pack-name': 'CP1-CFP0', 'supporting-interface': 'XPDR1-NETWORK1-OTU',
 268                                        'type': 'org-openroadm-interfaces:otnOdu',
 269                                        'supporting-port': 'CP1-CFP0-P1'}, res['interface'][0])
 270         self.assertDictContainsSubset(
 271             {'odu-function': 'org-openroadm-otn-common-types:ODU-TTP',
 272              'rate': 'org-openroadm-otn-common-types:ODU4'},
 273             res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'])
 274         self.assertDictEqual(
 275             {u'payload-type': u'21', u'exp-payload-type': u'21'},
 276             res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
 277
 278     def test_11_otn_service_path_create_10GE(self):
 279         url = "{}/operations/transportpce-device-renderer:otn-service-path".format(self.restconf_baseurl)
 280         data = {"renderer:input": {
 281             "service-name": "service1",
 282             "operation": "create",
 283             "service-rate": "10G",
 284             "service-type": "Ethernet",
 285             "ethernet-encoding": "eth encode",
 286             "trib-slot": ["1"],
 287             "trib-port-number": "1",
 288             "nodes": [
 289                 {"node-id": "SPDR-SA1",
 290                  "client-tp": "XPDR1-CLIENT1",
 291                  "network-tp": "XPDR1-NETWORK1"}]}}
 292         headers = {'content-type': 'application/json'}
 293         response = requests.request(
 294             "POST", url, data=json.dumps(data),
 295             headers=headers, auth=('admin', 'admin'))
 296         time.sleep(3)
 297         self.assertEqual(response.status_code, requests.codes.ok)
 298         res = response.json()
 299         self.assertIn('Otn Service path was set up successfully for node :SPDR-SA1', res["output"]["result"])
 300         self.assertTrue(res["output"]["success"])
 301         self.assertIn(
 302             {'node-id': 'SPDR-SA1',
 303              'connection-id': ['XPDR1-CLIENT1-ODU2e-service1-x-XPDR1-NETWORK1-ODU2e-service1'],
 304              'odu-interface-id': ['XPDR1-NETWORK1-ODU2e-service1', 'XPDR1-CLIENT1-ODU2e-service1'],
 305              'eth-interface-id': ['XPDR1-CLIENT1-ETHERNET10G']}, res["output"]['node-interface'])
 306
 307     def test_12_check_interface_10GE_CLIENT(self):
 308         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 309                "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 310                "interface/XPDR1-CLIENT1-ETHERNET10G"
 311                .format(self.restconf_baseurl))
 312         headers = {'content-type': 'application/json'}
 313         response = requests.request(
 314             "GET", url, headers=headers, auth=('admin', 'admin'))
 315         self.assertEqual(response.status_code, requests.codes.ok)
 316         res = response.json()
 317         self.assertDictContainsSubset({'name': 'XPDR1-CLIENT1-ETHERNET10G', 'administrative-state': 'inService',
 318                                        'supporting-circuit-pack-name': 'CP1-SFP4',
 319                                        'type': 'org-openroadm-interfaces:ethernetCsmacd',
 320                                        'supporting-port': 'CP1-SFP4-P1'}, res['interface'][0])
 321         self.assertDictEqual(
 322             {u'speed': 10000},
 323             res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
 324
 325     def test_13_check_interface_ODU2E_CLIENT(self):
 326         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 327                "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 328                "interface/XPDR1-CLIENT1-ODU2e-service1"
 329                .format(self.restconf_baseurl))
 330         headers = {'content-type': 'application/json'}
 331         response = requests.request(
 332             "GET", url, headers=headers, auth=('admin', 'admin'))
 333         self.assertEqual(response.status_code, requests.codes.ok)
 334         res = response.json()
 335         self.assertDictContainsSubset({'name': 'XPDR1-CLIENT1-ODU2e-service1', 'administrative-state': 'inService',
 336                                        'supporting-circuit-pack-name': 'CP1-SFP4',
 337                                        'supporting-interface': 'XPDR1-CLIENT1-ETHERNET10G',
 338                                        'type': 'org-openroadm-interfaces:otnOdu',
 339                                        'supporting-port': 'CP1-SFP4-P1'}, res['interface'][0])
 340         self.assertDictContainsSubset({
 341             'odu-function': 'org-openroadm-otn-common-types:ODU-TTP-CTP',
 342             'rate': 'org-openroadm-otn-common-types:ODU2e',
 343             'monitoring-mode': 'terminated'}, res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'])
 344         self.assertDictEqual(
 345             {u'payload-type': u'03', u'exp-payload-type': u'03'},
 346             res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
 347
 348     def test_14_check_interface_ODU2E_NETWORK(self):
 349         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 350                "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 351                "interface/XPDR1-NETWORK1-ODU2e-service1"
 352                .format(self.restconf_baseurl))
 353         headers = {'content-type': 'application/json'}
 354         response = requests.request(
 355             "GET", url, headers=headers, auth=('admin', 'admin'))
 356         self.assertEqual(response.status_code, requests.codes.ok)
 357         res = response.json()
 358         self.assertDictContainsSubset({'name': 'XPDR1-NETWORK1-ODU2e-service1', 'administrative-state': 'inService',
 359                                        'supporting-circuit-pack-name': 'CP1-CFP0',
 360                                        'supporting-interface': 'XPDR1-NETWORK1-ODU4',
 361                                        'type': 'org-openroadm-interfaces:otnOdu',
 362                                        'supporting-port': 'CP1-CFP0-P1'}, res['interface'][0])
 363         self.assertDictContainsSubset({
 364             'odu-function': 'org-openroadm-otn-common-types:ODU-CTP',
 365             'rate': 'org-openroadm-otn-common-types:ODU2e',
 366             'monitoring-mode': 'monitored'}, res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'])
 367         self.assertDictContainsSubset(
 368             {'trib-port-number': 1},
 369             res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['parent-odu-allocation'])
 370         self.assertIn(1,
 371                       res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['parent-odu-allocation']['trib-slots'])
 372
 373     def test_15_check_ODU2E_connection(self):
 374         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 375                "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 376                "odu-connection/XPDR1-CLIENT1-ODU2e-service1-x-XPDR1-NETWORK1-ODU2e-service1"
 377                .format(self.restconf_baseurl))
 378         headers = {'content-type': 'application/json'}
 379         response = requests.request(
 380             "GET", url, headers=headers, auth=('admin', 'admin'))
 381         self.assertEqual(response.status_code, requests.codes.ok)
 382         res = response.json()
 383         self.assertDictContainsSubset({
 384             'connection-name': 'XPDR1-CLIENT1-ODU2e-service1-x-XPDR1-NETWORK1-ODU2e-service1',
 385             'direction': 'bidirectional'},
 386             res['odu-connection'][0])
 387         self.assertDictEqual({u'dst-if': u'XPDR1-NETWORK1-ODU2e-service1'},
 388                              res['odu-connection'][0]['destination'])
 389         self.assertDictEqual({u'src-if': u'XPDR1-CLIENT1-ODU2e-service1'},
 390                              res['odu-connection'][0]['source'])
 391
 392     def test_16_otn_service_path_delete_10GE(self):
 393         url = "{}/operations/transportpce-device-renderer:otn-service-path".format(self.restconf_baseurl)
 394         data = {"renderer:input": {
 395             "service-name": "service1",
 396             "operation": "delete",
 397             "service-rate": "10G",
 398             "service-type": "Ethernet",
 399             "ethernet-encoding": "eth encode",
 400             "trib-slot": ["1"],
 401             "trib-port-number": "1",
 402             "nodes": [
 403                 {"node-id": "SPDR-SA1",
 404                  "client-tp": "XPDR1-CLIENT1",
 405                  "network-tp": "XPDR1-NETWORK1"}]}}
 406         headers = {'content-type': 'application/json'}
 407         response = requests.request(
 408             "POST", url, data=json.dumps(data),
 409             headers=headers, auth=('admin', 'admin'))
 410         time.sleep(3)
 411         self.assertEqual(response.status_code, requests.codes.ok)
 412         res = response.json()
 413         self.assertIn('Request processed', res["output"]["result"])
 414         self.assertTrue(res["output"]["success"])
 415
 416     def test_17_check_no_ODU2E_connection(self):
 417         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 418                "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 419                "odu-connection/XPDR1-CLIENT1-ODU2e-service1-x-XPDR1-NETWORK1-ODU2e-service1"
 420                .format(self.restconf_baseurl))
 421         headers = {'content-type': 'application/json'}
 422         response = requests.request(
 423             "GET", url, headers=headers, auth=('admin', 'admin'))
 424         self.assertEqual(response.status_code, requests.codes.not_found)
 425
 426     def test_18_check_no_interface_ODU2E_NETWORK(self):
 427         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 428                "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 429                "interface/XPDR1-NETWORK1-ODU2e-service1"
 430                .format(self.restconf_baseurl))
 431         headers = {'content-type': 'application/json'}
 432         response = requests.request(
 433             "GET", url, headers=headers, auth=('admin', 'admin'))
 434         self.assertEqual(response.status_code, requests.codes.not_found)
 435
 436     def test_19_check_no_interface_ODU2E_CLIENT(self):
 437         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 438                "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 439                "interface/XPDR1-CLIENT1-ODU2e-service1"
 440                .format(self.restconf_baseurl))
 441         headers = {'content-type': 'application/json'}
 442         response = requests.request(
 443             "GET", url, headers=headers, auth=('admin', 'admin'))
 444         self.assertEqual(response.status_code, requests.codes.not_found)
 445
 446     def test_20_check_no_interface_10GE_CLIENT(self):
 447         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 448                "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 449                "interface/XPDR1-CLIENT1-ETHERNET10G"
 450                .format(self.restconf_baseurl))
 451         headers = {'content-type': 'application/json'}
 452         response = requests.request(
 453             "GET", url, headers=headers, auth=('admin', 'admin'))
 454         self.assertEqual(response.status_code, requests.codes.not_found)
 455
 456     def test_21_otn_service_path_delete_ODU4(self):
 457         url = "{}/operations/transportpce-device-renderer:otn-service-path".format(self.restconf_baseurl)
 458         data = {"renderer:input": {
 459             "service-name": "service_ODU4",
 460             "operation": "delete",
 461             "service-rate": "100G",
 462             "service-type": "ODU",
 463             "nodes": [
 464                 {"node-id": "SPDR-SA1",
 465                  "network-tp": "XPDR1-NETWORK1"}]}}
 466         headers = {'content-type': 'application/json'}
 467         response = requests.request(
 468             "POST", url, data=json.dumps(data),
 469             headers=headers, auth=('admin', 'admin'))
 470         time.sleep(3)
 471         self.assertEqual(response.status_code, requests.codes.ok)
 472         res = response.json()
 473         self.assertIn('Request processed', res["output"]["result"])
 474         self.assertTrue(res["output"]["success"])
 475
 476     def test_22_check_no_interface_ODU4(self):
 477         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 478                "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 479                "interface/XPDR1-NETWORK1-ODU4"
 480                .format(self.restconf_baseurl))
 481         headers = {'content-type': 'application/json'}
 482         response = requests.request(
 483             "GET", url, headers=headers, auth=('admin', 'admin'))
 484         self.assertEqual(response.status_code, requests.codes.not_found)
 485
 486     def test_23_service_path_delete_OCH_OTU4(self):
 487         url = "{}/operations/transportpce-device-renderer:service-path".format(self.restconf_baseurl)
 488         data = {"renderer:input": {
 489             "service-name": "service_OTU4",
 490             "wave-number": "1",
 491             "modulation-format": "qpsk",
 492             "operation": "delete",
 493             "nodes": [
 494                 {"node-id": "SPDR-SA1",
 495                  "dest-tp": "XPDR1-NETWORK1"}]}}
 496         headers = {'content-type': 'application/json'}
 497         response = requests.request(
 498             "POST", url, data=json.dumps(data),
 499             headers=headers, auth=('admin', 'admin'))
 500         time.sleep(3)
 501         self.assertEqual(response.status_code, requests.codes.ok)
 502         res = response.json()
 503         self.assertIn('Request processed', res["output"]["result"])
 504         self.assertTrue(res["output"]["success"])
 505
 506     def test_24_check_no_interface_OTU4(self):
 507         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 508                "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 509                "interface/XPDR1-NETWORK1-OTU"
 510                .format(self.restconf_baseurl))
 511         headers = {'content-type': 'application/json'}
 512         response = requests.request(
 513             "GET", url, headers=headers, auth=('admin', 'admin'))
 514         self.assertEqual(response.status_code, requests.codes.not_found)
 515
 516     def test_25_check_no_interface_OCH(self):
 517         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 518                "node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 519                "interface/XPDR1-NETWORK1-1"
 520                .format(self.restconf_baseurl))
 521         headers = {'content-type': 'application/json'}
 522         response = requests.request(
 523             "GET", url, headers=headers, auth=('admin', 'admin'))
 524         self.assertEqual(response.status_code, requests.codes.not_found)
 525
 526     def test_26_disconnect_SPDR_SA1(self):
 527         url = ("{}/config/network-topology:"
 528                "network-topology/topology/topology-netconf/node/SPDR-SA1"
 529                .format(self.restconf_baseurl))
 530         data = {}
 531         headers = {'content-type': 'application/json'}
 532         response = requests.request(
 533             "DELETE", url, data=json.dumps(data), headers=headers,
 534             auth=('admin', 'admin'))
 535         self.assertEqual(response.status_code, requests.codes.ok)
 536
 537
 538 if __name__ == "__main__":
 539     unittest.main(verbosity=2)