tests/transportpce_tests/tapi/test02_full_topology.py

   1 #!/usr/bin/env python
   2
   3 ##############################################################################
   4 # Copyright (c) 2021 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 # pylint: disable=no-member
  13 # pylint: disable=too-many-public-methods
  14
  15 import os
  16 import unittest
  17 import time
  18 import requests
  19 # pylint: disable=wrong-import-order
  20 import sys
  21 sys.path.append('transportpce_tests/common/')
  22 # pylint: disable=wrong-import-position
  23 # pylint: disable=import-error
  24 import test_utils  # nopep8
  25
  26
  27 # pylint: disable=too-few-public-methods
  28 class UuidServices:
  29     def __init__(self):
  30         # pylint: disable=invalid-name
  31         self.pm = None
  32         self.odu = None
  33         self.dsr = None
  34
  35
  36 class TransportPCEtesting(unittest.TestCase):
  37
  38     processes = []
  39     WAITING = 20  # nominal value is 300
  40     NODE_VERSION = '2.2.1'
  41     uuid_services = UuidServices()
  42
  43     cr_serv_input_data = {
  44         "end-point": [
  45             {
  46                 "layer-protocol-name": "PHOTONIC_MEDIA",
  47                 "service-interface-point": {
  48                     "service-interface-point-uuid": "b1a0d883-32b8-3b0b-93d6-7ed074f6f107"
  49                 },
  50                 "administrative-state": "UNLOCKED",
  51                 "operational-state": "ENABLED",
  52                 "direction": "BIDIRECTIONAL",
  53                 "role": "SYMMETRIC",
  54                 "protection-role": "WORK",
  55                 "local-id": "SPDR-SA1-XPDR1",
  56                 "name": [
  57                         {
  58                             "value-name": "OpenROADM node id",
  59                             "value": "SPDR-SA1-XPDR1"
  60                         }
  61                 ]
  62             },
  63             {
  64                 "layer-protocol-name": "PHOTONIC_MEDIA",
  65                 "service-interface-point": {
  66                     "service-interface-point-uuid": "d1d6305e-179b-346f-b02d-8260aebe1ce8"
  67                 },
  68                 "administrative-state": "UNLOCKED",
  69                 "operational-state": "ENABLED",
  70                 "direction": "BIDIRECTIONAL",
  71                 "role": "SYMMETRIC",
  72                 "protection-role": "WORK",
  73                 "local-id": "SPDR-SC1-XPDR1",
  74                 "name": [
  75                         {
  76                             "value-name": "OpenROADM node id",
  77                             "value": "SPDR-SC1-XPDR1"
  78                         }
  79                 ]
  80             }
  81         ],
  82         "connectivity-constraint": {
  83             "service-layer": "PHOTONIC_MEDIA",
  84             "service-type": "POINT_TO_POINT_CONNECTIVITY",
  85             "service-level": "Some service-level",
  86             "requested-capacity": {
  87                 "total-size": {
  88                     "value": "100",
  89                     "unit": "GB"
  90                 }
  91             }
  92         },
  93         "state": "Some state"}
  94
  95     del_serv_input_data = {"service-id-or-name": "TBD"}
  96
  97     tapi_topo = {"topology-id-or-name": "TBD"}
  98
  99     @classmethod
 100     def setUpClass(cls):
 101         # pylint: disable=unsubscriptable-object
 102         cls.init_failed = False
 103         os.environ['JAVA_MIN_MEM'] = '1024M'
 104         os.environ['JAVA_MAX_MEM'] = '4096M'
 105         cls.processes = test_utils.start_tpce()
 106         # TAPI feature is not installed by default in Karaf
 107         if "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
 108             print("installing tapi feature...")
 109             result = test_utils.install_karaf_feature("odl-transportpce-tapi")
 110             if result.returncode != 0:
 111                 cls.init_failed = True
 112         if cls.init_failed:
 113             print("tapi installation feature failed...")
 114             test_utils.shutdown_process(cls.processes[0])
 115             sys.exit(2)
 116         cls.processes = test_utils.start_sims([('spdra', cls.NODE_VERSION),
 117                                                ('roadma', cls.NODE_VERSION),
 118                                                ('roadmc', cls.NODE_VERSION),
 119                                                ('spdrc', cls.NODE_VERSION)])
 120
 121     @classmethod
 122     def tearDownClass(cls):
 123         # pylint: disable=not-an-iterable
 124         for process in cls.processes:
 125             test_utils.shutdown_process(process)
 126         print("all processes killed")
 127
 128     def setUp(self):
 129         time.sleep(5)
 130
 131     def test_01_connect_spdrA(self):
 132         print("Connecting SPDRA")
 133         response = test_utils.mount_device("SPDR-SA1", ('spdra', self.NODE_VERSION))
 134         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 135         time.sleep(2)
 136
 137     def test_02_connect_spdrC(self):
 138         print("Connecting SPDRC")
 139         response = test_utils.mount_device("SPDR-SC1", ('spdrc', self.NODE_VERSION))
 140         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 141         time.sleep(2)
 142
 143     def test_03_connect_rdmA(self):
 144         print("Connecting ROADMA")
 145         response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION))
 146         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 147         time.sleep(2)
 148
 149     def test_04_connect_rdmC(self):
 150         print("Connecting ROADMC")
 151         response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION))
 152         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 153         time.sleep(2)
 154
 155     def test_05_connect_sprdA_1_N1_to_roadmA_PP1(self):
 156         response = test_utils.transportpce_api_rpc_request(
 157             'transportpce-networkutils', 'init-xpdr-rdm-links',
 158             {'links-input': {'xpdr-node': 'SPDR-SA1', 'xpdr-num': '1', 'network-num': '1',
 159                              'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 160         self.assertEqual(response['status_code'], requests.codes.ok)
 161         self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
 162         time.sleep(2)
 163
 164     def test_06_connect_roadmA_PP1_to_spdrA_1_N1(self):
 165         response = test_utils.transportpce_api_rpc_request(
 166             'transportpce-networkutils', 'init-rdm-xpdr-links',
 167             {'links-input': {'xpdr-node': 'SPDR-SA1', 'xpdr-num': '1', 'network-num': '1',
 168                              'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 169         self.assertEqual(response['status_code'], requests.codes.ok)
 170         self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
 171         time.sleep(2)
 172
 173     def test_07_connect_sprdC_1_N1_to_roadmC_PP1(self):
 174         response = test_utils.transportpce_api_rpc_request(
 175             'transportpce-networkutils', 'init-xpdr-rdm-links',
 176             {'links-input': {'xpdr-node': 'SPDR-SC1', 'xpdr-num': '1', 'network-num': '1',
 177                              'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 178         self.assertEqual(response['status_code'], requests.codes.ok)
 179         self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
 180         time.sleep(2)
 181
 182     def test_08_connect_roadmC_PP1_to_spdrC_1_N1(self):
 183         response = test_utils.transportpce_api_rpc_request(
 184             'transportpce-networkutils', 'init-rdm-xpdr-links',
 185             {'links-input': {'xpdr-node': 'SPDR-SC1', 'xpdr-num': '1', 'network-num': '1',
 186                              'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 187         self.assertEqual(response['status_code'], requests.codes.ok)
 188         self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
 189         time.sleep(2)
 190
 191     def test_09_add_omsAttributes_ROADMA_ROADMC(self):
 192         # Config ROADMA-ROADMC oms-attributes
 193         data = {"span": {
 194             "auto-spanloss": "true",
 195             "spanloss-base": 11.4,
 196             "spanloss-current": 12,
 197             "engineered-spanloss": 12.2,
 198             "link-concatenation": [{
 199                 "SRLG-Id": 0,
 200                 "fiber-type": "smf",
 201                 "SRLG-length": 100000,
 202                 "pmd": 0.5}]}}
 203         response = test_utils.add_oms_attr_request(
 204             "ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", data)
 205         self.assertEqual(response.status_code, requests.codes.created)
 206
 207     def test_10_add_omsAttributes_ROADMC_ROADMA(self):
 208         # Config ROADMC-ROADMA oms-attributes
 209         data = {"span": {
 210             "auto-spanloss": "true",
 211             "spanloss-base": 11.4,
 212             "spanloss-current": 12,
 213             "engineered-spanloss": 12.2,
 214             "link-concatenation": [{
 215                 "SRLG-Id": 0,
 216                 "fiber-type": "smf",
 217                 "SRLG-length": 100000,
 218                 "pmd": 0.5}]}}
 219         response = test_utils.add_oms_attr_request(
 220             "ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX", data)
 221         self.assertEqual(response.status_code, requests.codes.created)
 222
 223     def test_11_check_otn_topology(self):
 224         response = test_utils.get_ietf_network_request('otn-topology', 'config')
 225         self.assertEqual(response['status_code'], requests.codes.ok)
 226         self.assertEqual(len(response['network'][0]['node']), 6, 'There should be 6 otn nodes')
 227         self.assertNotIn('ietf-network-topology:link', response['network'][0])
 228
 229     def test_12_check_openroadm_topology(self):
 230         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
 231         self.assertEqual(response['status_code'], requests.codes.ok)
 232         self.assertEqual(len(response['network'][0]['node']), 13, 'There should be 13 openroadm nodes')
 233         self.assertEqual(len(response['network'][0]['ietf-network-topology:link']), 22,
 234                          'There should be 22 openroadm links')
 235
 236     def test_13_get_tapi_topology_details(self):
 237         self.tapi_topo["topology-id-or-name"] = test_utils.T0_FULL_MULTILAYER_TOPO
 238         response = test_utils.transportpce_api_rpc_request(
 239             'tapi-topology', 'get-topology-details', self.tapi_topo)
 240         time.sleep(2)
 241         self.assertEqual(response['status_code'], requests.codes.ok)
 242         self.assertEqual(len(response['output']['topology']['node']), 14, 'There should be 14 TAPI nodes')
 243         self.assertEqual(len(response['output']['topology']['link']), 15, 'There should be 15 TAPI links')
 244
 245     def test_14_check_sip_details(self):
 246         response = test_utils.transportpce_api_rpc_request(
 247             'tapi-common', 'get-service-interface-point-list', None)
 248         self.assertEqual(len(response['output']['sip']), 60, 'There should be 60 service interface point')
 249
 250 # test create connectivity service from spdrA to spdrC for Photonic_media
 251     def test_15_create_connectivity_service_PhotonicMedia(self):
 252         response = test_utils.transportpce_api_rpc_request(
 253             'tapi-connectivity', 'create-connectivity-service', self.cr_serv_input_data)
 254         time.sleep(self.WAITING)
 255         self.assertEqual(response['status_code'], requests.codes.ok)
 256         self.uuid_services.pm = response['output']['service']['uuid']
 257         # pylint: disable=consider-using-f-string
 258         print("photonic media service uuid : {}".format(self.uuid_services.pm))
 259
 260         input_dict_1 = {'administrative-state': 'LOCKED',
 261                         'lifecycle-state': 'PLANNED',
 262                         'operational-state': 'DISABLED',
 263                         'service-type': 'POINT_TO_POINT_CONNECTIVITY',
 264                         'service-layer': 'PHOTONIC_MEDIA',
 265                         'connectivity-direction': 'BIDIRECTIONAL'
 266                         }
 267         input_dict_2 = {'value-name': 'OpenROADM node id',
 268                         'value': 'SPDR-SC1-XPDR1'}
 269         input_dict_3 = {'value-name': 'OpenROADM node id',
 270                         'value': 'SPDR-SA1-XPDR1'}
 271
 272         self.assertDictEqual(dict(input_dict_1, **response['output']['service']),
 273                              response['output']['service'])
 274         self.assertDictEqual(dict(input_dict_2, **response['output']['service']['end-point'][0]['name'][0]),
 275                              response['output']['service']['end-point'][0]['name'][0])
 276         self.assertDictEqual(dict(input_dict_3, **response['output']['service']['end-point'][1]['name'][0]),
 277                              response['output']['service']['end-point'][1]['name'][0])
 278         # If the gate fails is because of the waiting time not being enough
 279 #        time.sleep(self.WAITING)
 280
 281     def test_16_get_service_PhotonicMedia(self):
 282         response = test_utils.get_ordm_serv_list_attr_request("services", str(self.uuid_services.pm))
 283         self.assertEqual(response['status_code'], requests.codes.ok)
 284         self.assertEqual(response['services'][0]['administrative-state'], 'inService')
 285         self.assertEqual(response['services'][0]['service-name'], str(self.uuid_services.pm))
 286         self.assertEqual(response['services'][0]['connection-type'], 'infrastructure')
 287         self.assertEqual(response['services'][0]['lifecycle-state'], 'planned')
 288         time.sleep(1)
 289
 290 # test create connectivity service from spdrA to spdrC for odu
 291     def test_17_create_connectivity_service_ODU(self):
 292         # pylint: disable=line-too-long
 293         self.cr_serv_input_data["end-point"][0]["layer-protocol-name"] = "ODU"
 294         self.cr_serv_input_data["end-point"][0]["service-interface-point"]["service-interface-point-uuid"] = "5efda776-f8de-3e0b-9bbd-2c702e210946"
 295         self.cr_serv_input_data["end-point"][1]["layer-protocol-name"] = "ODU"
 296         self.cr_serv_input_data["end-point"][1]["service-interface-point"]["service-interface-point-uuid"] = "8116d0af-39fa-3df5-bed2-dd2cd5e8217d"
 297         self.cr_serv_input_data["connectivity-constraint"]["service-layer"] = "ODU"
 298         self.cr_serv_input_data["connectivity-constraint"]["service-level"] = self.uuid_services.pm
 299
 300         response = test_utils.transportpce_api_rpc_request(
 301             'tapi-connectivity', 'create-connectivity-service', self.cr_serv_input_data)
 302         time.sleep(self.WAITING)
 303         self.assertEqual(response['status_code'], requests.codes.ok)
 304         self.uuid_services.odu = response['output']['service']['uuid']
 305         # pylint: disable=consider-using-f-string
 306         print("odu service uuid : {}".format(self.uuid_services.odu))
 307
 308         input_dict_1 = {'administrative-state': 'LOCKED',
 309                         'lifecycle-state': 'PLANNED',
 310                         'operational-state': 'DISABLED',
 311                         'service-type': 'POINT_TO_POINT_CONNECTIVITY',
 312                         'service-layer': 'ODU',
 313                         'connectivity-direction': 'BIDIRECTIONAL'
 314                         }
 315         input_dict_2 = {'value-name': 'OpenROADM node id',
 316                         'value': 'SPDR-SC1-XPDR1'}
 317         input_dict_3 = {'value-name': 'OpenROADM node id',
 318                         'value': 'SPDR-SA1-XPDR1'}
 319
 320         self.assertDictEqual(dict(input_dict_1, **response['output']['service']),
 321                              response['output']['service'])
 322         self.assertDictEqual(dict(input_dict_2, **response['output']['service']['end-point'][0]['name'][0]),
 323                              response['output']['service']['end-point'][0]['name'][0])
 324         self.assertDictEqual(dict(input_dict_3, **response['output']['service']['end-point'][1]['name'][0]),
 325                              response['output']['service']['end-point'][1]['name'][0])
 326         # If the gate fails is because of the waiting time not being enough
 327 #        time.sleep(self.WAITING)
 328
 329     def test_18_get_service_ODU(self):
 330         response = test_utils.get_ordm_serv_list_attr_request("services", str(self.uuid_services.odu))
 331         self.assertEqual(response['status_code'], requests.codes.ok)
 332         self.assertEqual(response['services'][0]['administrative-state'], 'inService')
 333         self.assertEqual(response['services'][0]['service-name'], str(self.uuid_services.odu))
 334         self.assertEqual(response['services'][0]['connection-type'], 'infrastructure')
 335         self.assertEqual(response['services'][0]['lifecycle-state'], 'planned')
 336         time.sleep(1)
 337
 338 # test create connectivity service from spdrA to spdrC for dsr
 339     def test_19_create_connectivity_service_DSR(self):
 340         # pylint: disable=line-too-long
 341         self.cr_serv_input_data["end-point"][0]["layer-protocol-name"] = "DSR"
 342         self.cr_serv_input_data["end-point"][0]["service-interface-point"]["service-interface-point-uuid"] = "c14797a0-adcc-3875-a1fe-df8949d1a2d7"
 343         self.cr_serv_input_data["end-point"][1]["layer-protocol-name"] = "DSR"
 344         self.cr_serv_input_data["end-point"][1]["service-interface-point"]["service-interface-point-uuid"] = "25812ef2-625d-3bf8-af55-5e93946d1c22"
 345         self.cr_serv_input_data["connectivity-constraint"]["service-layer"] = "DSR"
 346         self.cr_serv_input_data["connectivity-constraint"]["requested-capacity"]["total-size"]["value"] = "10"
 347         self.cr_serv_input_data["connectivity-constraint"]["service-level"] = self.uuid_services.odu
 348
 349         response = test_utils.transportpce_api_rpc_request(
 350             'tapi-connectivity', 'create-connectivity-service', self.cr_serv_input_data)
 351         time.sleep(self.WAITING)
 352         self.assertEqual(response['status_code'], requests.codes.ok)
 353         self.uuid_services.dsr = response['output']['service']['uuid']
 354         # pylint: disable=consider-using-f-string
 355         print("dsr service uuid : {}".format(self.uuid_services.dsr))
 356
 357         input_dict_1 = {'administrative-state': 'LOCKED',
 358                         'lifecycle-state': 'PLANNED',
 359                         'operational-state': 'DISABLED',
 360                         'service-type': 'POINT_TO_POINT_CONNECTIVITY',
 361                         'service-layer': 'DSR',
 362                         'connectivity-direction': 'BIDIRECTIONAL'
 363                         }
 364         input_dict_2 = {'value-name': 'OpenROADM node id',
 365                         'value': 'SPDR-SC1-XPDR1'}
 366         input_dict_3 = {'value-name': 'OpenROADM node id',
 367                         'value': 'SPDR-SA1-XPDR1'}
 368
 369         self.assertDictEqual(dict(input_dict_1,
 370                                   **response['output']['service']),
 371                              response['output']['service'])
 372         self.assertDictEqual(dict(input_dict_2,
 373                                   **response['output']['service']['end-point'][0]['name'][0]),
 374                              response['output']['service']['end-point'][0]['name'][0])
 375         self.assertDictEqual(dict(input_dict_3,
 376                                   **response['output']['service']['end-point'][1]['name'][0]),
 377                              response['output']['service']['end-point'][1]['name'][0])
 378         # The sleep here is okey as the DSR service creation is very fast
 379 #        time.sleep(self.WAITING)
 380
 381     def test_20_get_service_DSR(self):
 382         response = test_utils.get_ordm_serv_list_attr_request("services", str(self.uuid_services.dsr))
 383         self.assertEqual(response['status_code'], requests.codes.ok)
 384         self.assertEqual(response['services'][0]['administrative-state'], 'inService')
 385         self.assertEqual(response['services'][0]['service-name'], str(self.uuid_services.dsr))
 386         self.assertEqual(response['services'][0]['connection-type'], 'service')
 387         self.assertEqual(response['services'][0]['lifecycle-state'], 'planned')
 388         time.sleep(1)
 389
 390     def test_21_get_connectivity_service_list(self):
 391         response = test_utils.transportpce_api_rpc_request(
 392             'tapi-connectivity', 'get-connectivity-service-list', None)
 393         self.assertEqual(response['status_code'], requests.codes.ok)
 394         liste_service = response['output']['service']
 395         for ele in liste_service:
 396             if ele['uuid'] == self.uuid_services.pm:
 397                 self.assertEqual(ele['operational-state'], 'ENABLED')
 398                 self.assertEqual(ele['service-layer'], 'PHOTONIC_MEDIA')
 399                 nbconnection = len(ele['connection'])
 400                 self.assertEqual(nbconnection, 3, 'There should be 3 connections')
 401             elif ele['uuid'] == self.uuid_services.odu:
 402                 self.assertEqual(ele['operational-state'], 'ENABLED')
 403                 self.assertEqual(ele['service-layer'], 'ODU')
 404                 nbconnection = len(ele['connection'])
 405                 self.assertEqual(nbconnection, 1, 'There should be 1 connections')
 406             elif ele['uuid'] == self.uuid_services.dsr:
 407                 self.assertEqual(ele['operational-state'], 'ENABLED')
 408                 self.assertEqual(ele['service-layer'], 'DSR')
 409                 nbconnection = len(ele['connection'])
 410                 self.assertEqual(nbconnection, 2, 'There should be 2 connections')
 411             else:
 412                 self.fail("get connectivity service failed")
 413         time.sleep(2)
 414
 415     def test_22_delete_connectivity_service_DSR(self):
 416         self.del_serv_input_data["service-id-or-name"] = str(self.uuid_services.dsr)
 417         response = test_utils.transportpce_api_rpc_request(
 418             'tapi-connectivity', 'delete-connectivity-service', self.del_serv_input_data)
 419         self.assertIn(response["status_code"], (requests.codes.ok, requests.codes.no_content))
 420         time.sleep(self.WAITING)
 421
 422     def test_23_delete_connectivity_service_ODU(self):
 423         self.del_serv_input_data["service-id-or-name"] = str(self.uuid_services.odu)
 424         response = test_utils.transportpce_api_rpc_request(
 425             'tapi-connectivity', 'delete-connectivity-service', self.del_serv_input_data)
 426         self.assertIn(response["status_code"], (requests.codes.ok, requests.codes.no_content))
 427         time.sleep(self.WAITING)
 428
 429     def test_24_delete_connectivity_service_PhotonicMedia(self):
 430         self.del_serv_input_data["service-id-or-name"] = str(self.uuid_services.pm)
 431         response = test_utils.transportpce_api_rpc_request(
 432             'tapi-connectivity', 'delete-connectivity-service', self.del_serv_input_data)
 433         self.assertIn(response["status_code"], (requests.codes.ok, requests.codes.no_content))
 434         time.sleep(self.WAITING)
 435
 436     def test_25_get_no_tapi_services(self):
 437         response = test_utils.transportpce_api_rpc_request(
 438             'tapi-connectivity', 'get-connectivity-service-list', None)
 439         self.assertEqual(response['status_code'], requests.codes.internal_server_error)
 440         self.assertIn(
 441             {"error-type": "rpc", "error-tag": "operation-failed",
 442              "error-message": "No services exist in datastore",
 443              "error-info": "<severity>error</severity>"},
 444             response['output']['errors']['error'])
 445
 446     def test_26_get_no_openroadm_services(self):
 447         response = test_utils.get_ordm_serv_list_request()
 448         self.assertEqual(response['status_code'], requests.codes.conflict)
 449
 450     def test_27_disconnect_spdrA(self):
 451         response = test_utils.unmount_device("SPDR-SA1")
 452         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 453
 454     def test_28_disconnect_spdrC(self):
 455         response = test_utils.unmount_device("SPDR-SC1")
 456         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 457
 458     def test_29_disconnect_roadmA(self):
 459         response = test_utils.unmount_device("ROADM-A1")
 460         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 461
 462     def test_30_disconnect_roadmC(self):
 463         response = test_utils.unmount_device("ROADM-C1")
 464         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 465
 466
 467 if __name__ == "__main__":
 468     unittest.main(verbosity=2)