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     uuidTpAiOTSI = "f4c370be-e307-380d-a31b-7edc2b431e5d"
  43     uuidTpZiOTSI = "febb4502-6b27-3701-990b-3aa4941f48c4"
  44     uuidTpADSR = "3aca9b37-bc46-335d-b147-2423690dee18"
  45     uuidTpZDSR = "709d3595-6d56-3ad6-a3cd-5a4a09ce6f9d"
  46 #   SIP uuids
  47     # SIP+SPDR-SA1-XPDR1+PHOTONIC_MEDIA_OTS+XPDR1-NETWORK1 UUID IS
  48     sAOTS = "38d81f55-1798-3520-ba16-08efa56630c4"
  49     # SIP+SPDR-SC1-XPDR1+PHOTONIC_MEDIA_OTS+XPDR1-NETWORK1 UUID IS
  50     sZOTS = "97d9ba27-0efa-3010-8b98-b4d73240120c"
  51     # SIP+SPDR-SA1-XPDR1+eOTSi+XPDR1-NETWORK1 UUID IS
  52     sAeOTS = "f4dbce65-6191-3c84-b351-29ccb0629221"
  53     # SIP+SPDR-SC1-XPDR1+eOTSi+XPDR1-NETWORK1 UUID IS
  54     sZeOTS = "faabebd3-d7af-3389-96a7-261672744591"
  55     # SIP+SPDR-SA1-XPDR1+DSR+XPDR1-CLIENT1 UUID IS
  56     sADSR = "c14797a0-adcc-3875-a1fe-df8949d1a2d7"
  57     # SIP+SPDR-SC1-XPDR1+DSR+XPDR1-CLIENT1 UUID IS
  58     sZDSR = "25812ef2-625d-3bf8-af55-5e93946d1c22"
  59     # SIP+SPDR-SA1-XPDR1+eODU+XPDR1-CLIENT1 UUID IS
  60     sAeODU = "b6421484-531f-3444-adfc-e11c503f1fab"
  61     # SIP+SPDR-SC1-XPDR1+eODU+XPDR1-CLIENT1 UUID IS
  62     sZeODU = "4511fca7-0cf7-3b27-a128-3b372d5e1fa8"
  63     # uuid_A = uuid.UUID(bytes("SPDR-SA1-XPDR1+DSR+eOTSI+XPDR1-NETWORK1", 'utf-8'))
  64     # uuid_C = uuid.UUID(bytes("SPDR-SC1-XPDR1+DSR+eOTSI+XPDR1-NETWORK1", 'utf-8'))
  65
  66     cr_serv_input_data = {
  67         "end-point": [
  68             {
  69                 "layer-protocol-name": "PHOTONIC_MEDIA",
  70                 "service-interface-point": {
  71                     "service-interface-point-uuid": "c14797a0-adcc-3875-a1fe-df8949d1a2d7"
  72                 },
  73                 "administrative-state": "UNLOCKED",
  74                 "operational-state": "ENABLED",
  75                 "direction": "BIDIRECTIONAL",
  76                 "role": "SYMMETRIC",
  77                 "protection-role": "WORK",
  78                 "local-id": "SPDR-SA1-XPDR1",
  79                 "name": [
  80                         {
  81                             "value-name": "OpenROADM node id",
  82                             "value": "SPDR-SA1-XPDR1"
  83                         }
  84                 ]
  85             },
  86             {
  87                 "layer-protocol-name": "PHOTONIC_MEDIA",
  88                 "service-interface-point": {
  89                     "service-interface-point-uuid": "25812ef2-625d-3bf8-af55-5e93946d1c22"
  90                 },
  91                 "administrative-state": "UNLOCKED",
  92                 "operational-state": "ENABLED",
  93                 "direction": "BIDIRECTIONAL",
  94                 "role": "SYMMETRIC",
  95                 "protection-role": "WORK",
  96                 "local-id": "SPDR-SC1-XPDR1",
  97                 "name": [
  98                         {
  99                             "value-name": "OpenROADM node id",
 100                             "value": "SPDR-SC1-XPDR1"
 101                         }
 102                 ]
 103             }
 104         ],
 105         "connectivity-constraint": {
 106             "service-type": "POINT_TO_POINT_CONNECTIVITY",
 107             "service-level": "Some service-level",
 108             "requested-capacity": {
 109                 "total-size": {
 110                     "value": "100",
 111                     "unit": "tapi-common:CAPACITY_UNIT_GBPS"
 112                 }
 113             }
 114         },
 115         "topology-constraint": [
 116             {
 117                 "local-id": "localIdTopoConstraint",
 118                 "name": [
 119                     {
 120                             "value-name": "Dumb constraint",
 121                             "value": "for debug1"
 122                     }
 123                 ]
 124             }
 125         ],
 126         "state": "LOCKED",
 127         "layer-protocol-name": "PHOTONIC_MEDIA"}
 128
 129     del_serv_input_data = {"uuid": "TBD"}
 130
 131     tapi_topo = {"topology-id": "TBD"}
 132
 133     @classmethod
 134     def setUpClass(cls):
 135         # pylint: disable=unsubscriptable-object
 136         cls.init_failed = False
 137         os.environ['JAVA_MIN_MEM'] = '1024M'
 138         os.environ['JAVA_MAX_MEM'] = '4096M'
 139         cls.processes = test_utils.start_tpce()
 140         # TAPI feature is not installed by default in Karaf
 141         if "NO_ODL_STARTUP" not in os.environ or "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
 142             print("installing tapi feature...")
 143             result = test_utils.install_karaf_feature("odl-transportpce-tapi")
 144             if result.returncode != 0:
 145                 cls.init_failed = True
 146             print("Restarting OpenDaylight...")
 147             test_utils.shutdown_process(cls.processes[0])
 148             cls.processes[0] = test_utils.start_karaf()
 149             test_utils.process_list[0] = cls.processes[0]
 150             cls.init_failed = not test_utils.wait_until_log_contains(
 151                 test_utils.KARAF_LOG, test_utils.KARAF_OK_START_MSG, time_to_wait=60)
 152         if cls.init_failed:
 153             print("tapi installation feature failed...")
 154             test_utils.shutdown_process(cls.processes[0])
 155             sys.exit(2)
 156         cls.processes = test_utils.start_sims([('spdra', cls.NODE_VERSION),
 157                                                ('roadma', cls.NODE_VERSION),
 158                                                ('roadmc', cls.NODE_VERSION),
 159                                                ('spdrc', cls.NODE_VERSION)])
 160
 161     @classmethod
 162     def tearDownClass(cls):
 163         # pylint: disable=not-an-iterable
 164         for process in cls.processes:
 165             test_utils.shutdown_process(process)
 166         print("all processes killed")
 167
 168     def setUp(self):
 169         time.sleep(2)
 170
 171     def test_01_connect_spdrA(self):
 172         print("Connecting SPDRA")
 173         response = test_utils.mount_device("SPDR-SA1", ('spdra', self.NODE_VERSION))
 174         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 175         time.sleep(2)
 176
 177     def test_02_connect_spdrC(self):
 178         print("Connecting SPDRC")
 179         response = test_utils.mount_device("SPDR-SC1", ('spdrc', self.NODE_VERSION))
 180         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 181         time.sleep(2)
 182
 183     def test_03_connect_rdmA(self):
 184         print("Connecting ROADMA")
 185         response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION))
 186         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 187         time.sleep(2)
 188
 189     def test_04_connect_rdmC(self):
 190         print("Connecting ROADMC")
 191         response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION))
 192         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 193         time.sleep(2)
 194
 195     def test_05_connect_sprdA_1_N1_to_roadmA_PP1(self):
 196         response = test_utils.transportpce_api_rpc_request(
 197             'transportpce-networkutils', 'init-xpdr-rdm-links',
 198             {'links-input': {'xpdr-node': 'SPDR-SA1', 'xpdr-num': '1', 'network-num': '1',
 199                              'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 200         self.assertEqual(response['status_code'], requests.codes.ok)
 201         self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
 202         time.sleep(2)
 203
 204     def test_06_connect_roadmA_PP1_to_spdrA_1_N1(self):
 205         response = test_utils.transportpce_api_rpc_request(
 206             'transportpce-networkutils', 'init-rdm-xpdr-links',
 207             {'links-input': {'xpdr-node': 'SPDR-SA1', 'xpdr-num': '1', 'network-num': '1',
 208                              'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 209         self.assertEqual(response['status_code'], requests.codes.ok)
 210         self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
 211         time.sleep(2)
 212
 213     def test_07_connect_sprdC_1_N1_to_roadmC_PP1(self):
 214         response = test_utils.transportpce_api_rpc_request(
 215             'transportpce-networkutils', 'init-xpdr-rdm-links',
 216             {'links-input': {'xpdr-node': 'SPDR-SC1', 'xpdr-num': '1', 'network-num': '1',
 217                              'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 218         self.assertEqual(response['status_code'], requests.codes.ok)
 219         self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
 220         time.sleep(2)
 221
 222     def test_08_connect_roadmC_PP1_to_spdrC_1_N1(self):
 223         response = test_utils.transportpce_api_rpc_request(
 224             'transportpce-networkutils', 'init-rdm-xpdr-links',
 225             {'links-input': {'xpdr-node': 'SPDR-SC1', 'xpdr-num': '1', 'network-num': '1',
 226                              'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 227         self.assertEqual(response['status_code'], requests.codes.ok)
 228         self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
 229         time.sleep(2)
 230
 231     def test_09_add_omsAttributes_ROADMA_ROADMC(self):
 232         # Config ROADMA-ROADMC oms-attributes
 233         data = {"span": {
 234             "auto-spanloss": "true",
 235             "spanloss-base": 11.4,
 236             "spanloss-current": 12,
 237             "engineered-spanloss": 12.2,
 238             "link-concatenation": [{
 239                 "SRLG-Id": 0,
 240                 "fiber-type": "smf",
 241                 "SRLG-length": 100000,
 242                 "pmd": 0.5}]}}
 243         response = test_utils.add_oms_attr_request(
 244             "ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", data)
 245         self.assertEqual(response.status_code, requests.codes.created)
 246
 247     def test_10_add_omsAttributes_ROADMC_ROADMA(self):
 248         # Config ROADMC-ROADMA oms-attributes
 249         data = {"span": {
 250             "auto-spanloss": "true",
 251             "spanloss-base": 11.4,
 252             "spanloss-current": 12,
 253             "engineered-spanloss": 12.2,
 254             "link-concatenation": [{
 255                 "SRLG-Id": 0,
 256                 "fiber-type": "smf",
 257                 "SRLG-length": 100000,
 258                 "pmd": 0.5}]}}
 259         response = test_utils.add_oms_attr_request(
 260             "ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX", data)
 261         self.assertEqual(response.status_code, requests.codes.created)
 262
 263     def test_11_check_otn_topology(self):
 264         response = test_utils.get_ietf_network_request('otn-topology', 'config')
 265         self.assertEqual(response['status_code'], requests.codes.ok)
 266         self.assertEqual(len(response['network'][0]['node']), 6, 'There should be 6 otn nodes')
 267         self.assertNotIn('ietf-network-topology:link', response['network'][0])
 268
 269     def test_12_check_openroadm_topology(self):
 270         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
 271         self.assertEqual(response['status_code'], requests.codes.ok)
 272         self.assertEqual(len(response['network'][0]['node']), 13, 'There should be 13 openroadm nodes')
 273         self.assertEqual(len(response['network'][0]['ietf-network-topology:link']), 22,
 274                          'There should be 22 openroadm links')
 275
 276     def test_13_get_tapi_topology_details(self):
 277         self.tapi_topo["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
 278         response = test_utils.transportpce_api_rpc_request(
 279             'tapi-topology', 'get-topology-details', self.tapi_topo)
 280         time.sleep(2)
 281         self.assertEqual(response['status_code'], requests.codes.ok)
 282         self.assertEqual(len(response['output']['topology']['node']), 8, 'There should be 8 TAPI nodes')
 283         self.assertEqual(len(response['output']['topology']['link']), 3, 'There should be 3 TAPI links')
 284         print(response['output']['topology']['node'][0])
 285         print(response['output']['topology']['node'][1])
 286         print(response['output']['topology']['node'][2])
 287
 288     def test_14_check_sip_details(self):
 289         response = test_utils.transportpce_api_rpc_request(
 290             'tapi-common', 'get-service-interface-point-list', None)
 291         self.assertEqual(len(response['output']['sip']), 72, 'There should be 72 service interface point')
 292
 293 # test create connectivity service from spdrA to spdrC for Photonic_media
 294     def test_15_create_connectivity_service_PhotonicMedia(self):
 295         self.cr_serv_input_data["end-point"][0]["service-interface-point"]["service-interface-point-uuid"] = self.sAOTS
 296         self.cr_serv_input_data["end-point"][1]["service-interface-point"]["service-interface-point-uuid"] = self.sZOTS
 297         response = test_utils.transportpce_api_rpc_request(
 298             'tapi-connectivity', 'create-connectivity-service', self.cr_serv_input_data)
 299         time.sleep(self.WAITING)
 300         self.assertEqual(response['status_code'], requests.codes.ok)
 301         self.uuid_services.pm = response['output']['service']['uuid']
 302         # pylint: disable=consider-using-f-string
 303         print("photonic media service uuid : {}".format(self.uuid_services.pm))
 304
 305         input_dict_1 = {'administrative-state': 'LOCKED',
 306                         'lifecycle-state': 'PLANNED',
 307                         'operational-state': 'DISABLED',
 308                         # 'service-type': 'POINT_TO_POINT_CONNECTIVITY',
 309                         # 'service-layer': 'PHOTONIC_MEDIA',
 310                         'layer-protocol-name': 'PHOTONIC_MEDIA',
 311                         # 'connectivity-direction': 'BIDIRECTIONAL'
 312                         'direction': 'BIDIRECTIONAL'
 313                         }
 314         input_dict_2 = {'value-name': 'OpenROADM node id',
 315                         'value': 'SPDR-SC1-XPDR1'}
 316         input_dict_3 = {'value-name': 'OpenROADM node id',
 317                         'value': 'SPDR-SA1-XPDR1'}
 318
 319         self.assertDictEqual(dict(input_dict_1, **response['output']['service']),
 320                              response['output']['service'])
 321         self.assertDictEqual(dict(input_dict_2, **response['output']['service']['end-point'][0]['name'][0]),
 322                              response['output']['service']['end-point'][0]['name'][0])
 323         self.assertDictEqual(dict(input_dict_3, **response['output']['service']['end-point'][1]['name'][0]),
 324                              response['output']['service']['end-point'][1]['name'][0])
 325         # If the gate fails is because of the waiting time not being enough
 326 #        time.sleep(self.WAITING)
 327
 328     def test_16_get_service_PhotonicMedia(self):
 329         response = test_utils.get_ordm_serv_list_attr_request("services", str(self.uuid_services.pm))
 330         self.assertEqual(response['status_code'], requests.codes.ok)
 331         self.assertEqual(response['services'][0]['administrative-state'], 'inService')
 332         self.assertEqual(response['services'][0]['service-name'], str(self.uuid_services.pm))
 333         self.assertEqual(response['services'][0]['connection-type'], 'infrastructure')
 334         self.assertEqual(response['services'][0]['lifecycle-state'], 'planned')
 335         time.sleep(1)
 336
 337 # test create connectivity service from spdrA to spdrC for odu
 338     def test_17_create_connectivity_service_ODU(self):
 339         # pylint: disable=line-too-long
 340         self.cr_serv_input_data["layer-protocol-name"] = "ODU"
 341         self.cr_serv_input_data["end-point"][0]["layer-protocol-name"] = "ODU"
 342         self.cr_serv_input_data["end-point"][0]["service-interface-point"]["service-interface-point-uuid"] = self.sAeODU
 343         self.cr_serv_input_data["end-point"][1]["layer-protocol-name"] = "ODU"
 344         self.cr_serv_input_data["end-point"][1]["service-interface-point"]["service-interface-point-uuid"] = self.sZeODU
 345 #        self.cr_serv_input_data["connectivity-constraint"]["service-layer"] = "ODU"
 346         self.cr_serv_input_data["connectivity-constraint"]["service-level"] = self.uuid_services.pm
 347
 348         response = test_utils.transportpce_api_rpc_request(
 349             'tapi-connectivity', 'create-connectivity-service', self.cr_serv_input_data)
 350         time.sleep(self.WAITING)
 351         self.assertEqual(response['status_code'], requests.codes.ok)
 352         self.uuid_services.odu = response['output']['service']['uuid']
 353         # pylint: disable=consider-using-f-string
 354         print("odu service uuid : {}".format(self.uuid_services.odu))
 355
 356         input_dict_1 = {'administrative-state': 'LOCKED',
 357                         'lifecycle-state': 'PLANNED',
 358                         'operational-state': 'DISABLED',
 359                         # 'service-type': 'POINT_TO_POINT_CONNECTIVITY',
 360                         'layer-protocol-name': 'ODU',
 361                         'direction': 'BIDIRECTIONAL'
 362                         }
 363         input_dict_2 = {'value-name': 'OpenROADM node id',
 364                         'value': 'SPDR-SC1-XPDR1'}
 365         input_dict_3 = {'value-name': 'OpenROADM node id',
 366                         'value': 'SPDR-SA1-XPDR1'}
 367
 368         self.assertDictEqual(dict(input_dict_1, **response['output']['service']),
 369                              response['output']['service'])
 370         self.assertDictEqual(dict(input_dict_2, **response['output']['service']['end-point'][0]['name'][0]),
 371                              response['output']['service']['end-point'][0]['name'][0])
 372         self.assertDictEqual(dict(input_dict_3, **response['output']['service']['end-point'][1]['name'][0]),
 373                              response['output']['service']['end-point'][1]['name'][0])
 374         # If the gate fails is because of the waiting time not being enough
 375 #        time.sleep(self.WAITING)
 376
 377     def test_18_get_service_ODU(self):
 378         response = test_utils.get_ordm_serv_list_attr_request("services", str(self.uuid_services.odu))
 379         self.assertEqual(response['status_code'], requests.codes.ok)
 380         self.assertEqual(response['services'][0]['administrative-state'], 'inService')
 381         self.assertEqual(response['services'][0]['service-name'], str(self.uuid_services.odu))
 382         self.assertEqual(response['services'][0]['connection-type'], 'infrastructure')
 383         self.assertEqual(response['services'][0]['lifecycle-state'], 'planned')
 384         time.sleep(1)
 385
 386 # test create connectivity service from spdrA to spdrC for dsr
 387     def test_19_create_connectivity_service_DSR(self):
 388         # pylint: disable=line-too-long
 389         self.cr_serv_input_data["layer-protocol-name"] = "DSR"
 390         self.cr_serv_input_data["end-point"][0]["layer-protocol-name"] = "DSR"
 391         self.cr_serv_input_data["end-point"][0]["service-interface-point"]["service-interface-point-uuid"] = self.sADSR
 392         self.cr_serv_input_data["end-point"][1]["layer-protocol-name"] = "DSR"
 393         self.cr_serv_input_data["end-point"][1]["service-interface-point"]["service-interface-point-uuid"] = self.sZDSR
 394 #        self.cr_serv_input_data["connectivity-constraint"]["service-layer"] = "DSR"
 395         self.cr_serv_input_data["connectivity-constraint"]["requested-capacity"]["total-size"]["value"] = "10"
 396         self.cr_serv_input_data["connectivity-constraint"]["service-level"] = self.uuid_services.odu
 397
 398         response = test_utils.transportpce_api_rpc_request(
 399             'tapi-connectivity', 'create-connectivity-service', self.cr_serv_input_data)
 400         time.sleep(self.WAITING)
 401         self.assertEqual(response['status_code'], requests.codes.ok)
 402         self.uuid_services.dsr = response['output']['service']['uuid']
 403         # pylint: disable=consider-using-f-string
 404         print("dsr service uuid : {}".format(self.uuid_services.dsr))
 405
 406         input_dict_1 = {'administrative-state': 'LOCKED',
 407                         'lifecycle-state': 'PLANNED',
 408                         'operational-state': 'DISABLED',
 409                         # 'service-type': 'POINT_TO_POINT_CONNECTIVITY',
 410                         'layer-protocol-name': 'DSR',
 411                         'direction': 'BIDIRECTIONAL'
 412                         }
 413         input_dict_2 = {'value-name': 'OpenROADM node id',
 414                         'value': 'SPDR-SC1-XPDR1'}
 415         input_dict_3 = {'value-name': 'OpenROADM node id',
 416                         'value': 'SPDR-SA1-XPDR1'}
 417
 418         self.assertDictEqual(dict(input_dict_1,
 419                                   **response['output']['service']),
 420                              response['output']['service'])
 421         self.assertDictEqual(dict(input_dict_2,
 422                                   **response['output']['service']['end-point'][0]['name'][0]),
 423                              response['output']['service']['end-point'][0]['name'][0])
 424         self.assertDictEqual(dict(input_dict_3,
 425                                   **response['output']['service']['end-point'][1]['name'][0]),
 426                              response['output']['service']['end-point'][1]['name'][0])
 427         # The sleep here is okey as the DSR service creation is very fast
 428 #        time.sleep(self.WAITING)
 429
 430     def test_20_get_service_DSR(self):
 431         response = test_utils.get_ordm_serv_list_attr_request("services", str(self.uuid_services.dsr))
 432         self.assertEqual(response['status_code'], requests.codes.ok)
 433         self.assertEqual(response['services'][0]['administrative-state'], 'inService')
 434         self.assertEqual(response['services'][0]['service-name'], str(self.uuid_services.dsr))
 435         self.assertEqual(response['services'][0]['connection-type'], 'service')
 436         self.assertEqual(response['services'][0]['lifecycle-state'], 'planned')
 437         time.sleep(1)
 438
 439     def test_21_get_connectivity_service_list(self):
 440         response = test_utils.transportpce_api_rpc_request(
 441             'tapi-connectivity', 'get-connectivity-service-list', None)
 442         self.assertEqual(response['status_code'], requests.codes.ok)
 443         liste_service = response['output']['service']
 444         for ele in liste_service:
 445             if ele['uuid'] == self.uuid_services.pm:
 446                 self.assertEqual(ele['operational-state'], 'ENABLED')
 447 #                self.assertEqual(ele['service-layer'], 'PHOTONIC_MEDIA')
 448                 self.assertEqual(ele['layer-protocol-name'], 'PHOTONIC_MEDIA')
 449                 nbconnection = len(ele['connection'])
 450                 self.assertEqual(nbconnection, 3, 'There should be 3 connections')
 451             elif ele['uuid'] == self.uuid_services.odu:
 452                 self.assertEqual(ele['operational-state'], 'ENABLED')
 453 #                self.assertEqual(ele['service-layer'], 'ODU')
 454                 self.assertEqual(ele['layer-protocol-name'], 'ODU')
 455                 nbconnection = len(ele['connection'])
 456                 self.assertEqual(nbconnection, 1, 'There should be 1 connections')
 457             elif ele['uuid'] == self.uuid_services.dsr:
 458                 self.assertEqual(ele['operational-state'], 'ENABLED')
 459 #                self.assertEqual(ele['service-layer'], 'DSR')
 460                 self.assertEqual(ele['layer-protocol-name'], 'DSR')
 461                 nbconnection = len(ele['connection'])
 462                 self.assertEqual(nbconnection, 2, 'There should be 2 connections')
 463             else:
 464                 self.fail("get connectivity service failed")
 465         time.sleep(2)
 466
 467     def test_22_delete_connectivity_service_DSR(self):
 468         self.del_serv_input_data["uuid"] = str(self.uuid_services.dsr)
 469         response = test_utils.transportpce_api_rpc_request(
 470             'tapi-connectivity', 'delete-connectivity-service', self.del_serv_input_data)
 471         self.assertIn(response["status_code"], (requests.codes.ok, requests.codes.no_content))
 472         time.sleep(self.WAITING)
 473
 474     def test_23_delete_connectivity_service_ODU(self):
 475         self.del_serv_input_data["uuid"] = str(self.uuid_services.odu)
 476         response = test_utils.transportpce_api_rpc_request(
 477             'tapi-connectivity', 'delete-connectivity-service', self.del_serv_input_data)
 478         self.assertIn(response["status_code"], (requests.codes.ok, requests.codes.no_content))
 479         time.sleep(self.WAITING)
 480
 481     def test_24_delete_connectivity_service_PhotonicMedia(self):
 482         self.del_serv_input_data["uuid"] = str(self.uuid_services.pm)
 483         response = test_utils.transportpce_api_rpc_request(
 484             'tapi-connectivity', 'delete-connectivity-service', self.del_serv_input_data)
 485         self.assertIn(response["status_code"], (requests.codes.ok, requests.codes.no_content))
 486         time.sleep(self.WAITING)
 487
 488     def test_25_get_no_tapi_services(self):
 489         response = test_utils.transportpce_api_rpc_request(
 490             'tapi-connectivity', 'get-connectivity-service-list', None)
 491         self.assertEqual(response['status_code'], requests.codes.internal_server_error)
 492         self.assertIn(
 493             {"error-type": "rpc", "error-tag": "operation-failed",
 494              "error-message": "No services exist in datastore",
 495              "error-info": "<severity>error</severity>"},
 496             response['output']['errors']['error'])
 497
 498     def test_26_get_no_openroadm_services(self):
 499         response = test_utils.get_ordm_serv_list_request()
 500         self.assertEqual(response['status_code'], requests.codes.conflict)
 501
 502     def test_27_disconnect_spdrA(self):
 503         response = test_utils.unmount_device("SPDR-SA1")
 504         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 505
 506     def test_28_disconnect_spdrC(self):
 507         response = test_utils.unmount_device("SPDR-SC1")
 508         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 509
 510     def test_29_disconnect_roadmA(self):
 511         response = test_utils.unmount_device("ROADM-A1")
 512         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 513
 514     def test_30_disconnect_roadmC(self):
 515         response = test_utils.unmount_device("ROADM-C1")
 516         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 517
 518
 519 if __name__ == "__main__":
 520     unittest.main(verbosity=2)