tests/transportpce_tests/tapi/test03_tapi_device_change_notifications.py

   1 #!/usr/bin/env python
   2 ##############################################################################
   3 # Copyright (c) 2021 Orange, Inc. and others.  All rights reserved.
   4 #
   5 # All rights reserved. This program and the accompanying materials
   6 # are made available under the terms of the Apache License, Version 2.0
   7 # which accompanies this distribution, and is available at
   8 # http://www.apache.org/licenses/LICENSE-2.0
   9 ##############################################################################
  10
  11 # pylint: disable=no-member
  12 # pylint: disable=too-many-public-methods
  13 import os
  14 import unittest
  15 import time
  16 import requests
  17 # pylint: disable=wrong-import-order
  18 import sys
  19 sys.path.append('transportpce_tests/common/')
  20 # pylint: disable=wrong-import-position
  21 # pylint: disable=import-error
  22 import test_utils  # nopep8
  23
  24 # pylint: disable=too-few-public-methods
  25
  26
  27 class UuidServices:
  28     def __init__(self):
  29         # pylint: disable=invalid-name
  30         self.pm = None
  31         self.odu = None
  32         self.dsr = None
  33         self.eth = None
  34 # XPDR-A1-XPDR1+XPONDER Uuid is:
  35 # 4378fc29-6408-39ec-8737-5008c3dc49e5
  36 # XPDR-C1-XPDR1+XPONDER Uuid is:
  37 # 1770bea4-b1da-3b20-abce-7d182c0ec0df
  38 # ROADM-A1+PHOTONIC_MEDIA Uuid is:
  39 # 3b726367-6f2d-3e3f-9033-d99b61459075
  40
  41
  42 class TransportPCEFulltesting(unittest.TestCase):
  43
  44     cr_serv_input_data = {
  45         "end-point": [
  46             {
  47                 "layer-protocol-name": "DSR",
  48                 "service-interface-point": {
  49                     "service-interface-point-uuid": "b1f4bd3b-7fa9-367b-a8ab-6e80293238df"
  50                 },
  51                 "administrative-state": "UNLOCKED",
  52                 "operational-state": "ENABLED",
  53                 "direction": "BIDIRECTIONAL",
  54                 "role": "SYMMETRIC",
  55                 "protection-role": "WORK",
  56                 "local-id": "XPDR-C1-XPDR1",
  57                 "name": [
  58                         {
  59                             "value-name": "OpenROADM node id",
  60                             "value": "XPDR-C1-XPDR1"
  61                         }
  62                 ]
  63             },
  64             {
  65                 "layer-protocol-name": "DSR",
  66                 "service-interface-point": {
  67                     "service-interface-point-uuid": "b5964ce9-274c-3f68-b4d1-83c0b61bc74e"
  68                 },
  69                 "administrative-state": "UNLOCKED",
  70                 "operational-state": "ENABLED",
  71                 "direction": "BIDIRECTIONAL",
  72                 "role": "SYMMETRIC",
  73                 "protection-role": "WORK",
  74                 "local-id": "XPDR-A1-XPDR1",
  75                 "name": [
  76                         {
  77                             "value-name": "OpenROADM node id",
  78                             "value": "XPDR-A1-XPDR1"
  79                         }
  80                 ]
  81             }
  82         ],
  83         "connectivity-constraint": {
  84             #            "service-layer": "ETH",
  85             "service-type": "POINT_TO_POINT_CONNECTIVITY",
  86             "service-level": "Some service-level",
  87             "requested-capacity": {
  88                 "total-size": {
  89                     "value": "100",
  90                     "unit": "tapi-common:CAPACITY_UNIT_GBPS"
  91                 }
  92             }
  93         },
  94         "topology-constraint": [
  95             {
  96                 "local-id": "localIdTopoConstraint",
  97                 "name": [
  98                     {
  99                             "value-name": "Dumb constraint",
 100                             "value": "for debug1"
 101                     }
 102                 ]
 103             }
 104         ],
 105         "state": "LOCKED",
 106         "layer-protocol-name": "ETH"
 107     }
 108
 109     del_serv_input_data = {"uuid": "TBD"}
 110
 111     tapi_topo = {"topology-id": "TBD"}
 112
 113     node_details = {
 114         "topology-id": "TBD",
 115         "node-id": "TBD"
 116     }
 117
 118     tapi_serv_details = {"uuid": "TBD"}
 119
 120     processes = []
 121     uuid_services = UuidServices()
 122     WAITING = 25  # nominal value is 300
 123     NODE_VERSION_221 = '2.2.1'
 124
 125     @classmethod
 126     def setUpClass(cls):
 127         # pylint: disable=unsubscriptable-object
 128         cls.init_failed = False
 129         os.environ['JAVA_MIN_MEM'] = '1024M'
 130         os.environ['JAVA_MAX_MEM'] = '4096M'
 131         cls.processes = test_utils.start_tpce()
 132         # TAPI feature is not installed by default in Karaf
 133         if "NO_ODL_STARTUP" not in os.environ or "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
 134             print("installing tapi feature...")
 135             result = test_utils.install_karaf_feature("odl-transportpce-tapi")
 136             if result.returncode != 0:
 137                 cls.init_failed = True
 138             print("Restarting OpenDaylight...")
 139             test_utils.shutdown_process(cls.processes[0])
 140             cls.processes[0] = test_utils.start_karaf()
 141             test_utils.process_list[0] = cls.processes[0]
 142             cls.init_failed = not test_utils.wait_until_log_contains(
 143                 test_utils.KARAF_LOG, test_utils.KARAF_OK_START_MSG, time_to_wait=60)
 144         if cls.init_failed:
 145             print("tapi installation feature failed...")
 146             test_utils.shutdown_process(cls.processes[0])
 147             sys.exit(2)
 148         cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION_221),
 149                                                ('roadma', cls.NODE_VERSION_221),
 150                                                ('roadmc', cls.NODE_VERSION_221),
 151                                                ('xpdrc', cls.NODE_VERSION_221)])
 152
 153     @classmethod
 154     def tearDownClass(cls):
 155         # pylint: disable=not-an-iterable
 156         for process in cls.processes:
 157             test_utils.shutdown_process(process)
 158         print("all processes killed")
 159
 160     def setUp(self):  # instruction executed before each test method
 161         # pylint: disable=consider-using-f-string
 162         print("execution of {}".format(self.id().split(".")[-1]))
 163         time.sleep(1)
 164
 165     def test_01_connect_xpdrA(self):
 166         response = test_utils.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION_221))
 167         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 168
 169     def test_02_connect_xpdrC(self):
 170         response = test_utils.mount_device("XPDR-C1", ('xpdrc', self.NODE_VERSION_221))
 171         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 172
 173     def test_03_connect_rdmA(self):
 174         response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION_221))
 175         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 176
 177     def test_04_connect_rdmC(self):
 178         response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION_221))
 179         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 180
 181     def test_05_connect_xpdrA_N1_to_roadmA_PP1(self):
 182         response = test_utils.transportpce_api_rpc_request(
 183             'transportpce-networkutils', 'init-xpdr-rdm-links',
 184             {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1',
 185                              'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 186         self.assertEqual(response['status_code'], requests.codes.ok)
 187         self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
 188         time.sleep(2)
 189
 190     def test_06_connect_roadmA_PP1_to_xpdrA_N1(self):
 191         response = test_utils.transportpce_api_rpc_request(
 192             'transportpce-networkutils', 'init-rdm-xpdr-links',
 193             {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1',
 194                              'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 195         self.assertEqual(response['status_code'], requests.codes.ok)
 196         self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
 197         time.sleep(2)
 198
 199     def test_07_connect_xpdrC_N1_to_roadmC_PP1(self):
 200         response = test_utils.transportpce_api_rpc_request(
 201             'transportpce-networkutils', 'init-xpdr-rdm-links',
 202             {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1',
 203                              'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 204         self.assertEqual(response['status_code'], requests.codes.ok)
 205         self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
 206         time.sleep(2)
 207
 208     def test_08_connect_roadmC_PP1_to_xpdrC_N1(self):
 209         response = test_utils.transportpce_api_rpc_request(
 210             'transportpce-networkutils', 'init-rdm-xpdr-links',
 211             {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1',
 212                              'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 213         self.assertEqual(response['status_code'], requests.codes.ok)
 214         self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
 215         time.sleep(2)
 216
 217     def test_09_add_omsAttributes_ROADMA_ROADMC(self):
 218         # Config ROADMA-ROADMC oms-attributes
 219         data = {"span": {
 220             "auto-spanloss": "true",
 221             "spanloss-base": 11.4,
 222             "spanloss-current": 12,
 223             "engineered-spanloss": 12.2,
 224             "link-concatenation": [{
 225                 "SRLG-Id": 0,
 226                 "fiber-type": "smf",
 227                 "SRLG-length": 100000,
 228                 "pmd": 0.5}]}}
 229         response = test_utils.add_oms_attr_request(
 230             "ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", data)
 231         self.assertEqual(response.status_code, requests.codes.created)
 232
 233     def test_10_add_omsAttributes_ROADMC_ROADMA(self):
 234         # Config ROADMC-ROADMA oms-attributes
 235         data = {"span": {
 236             "auto-spanloss": "true",
 237             "spanloss-base": 11.4,
 238             "spanloss-current": 12,
 239             "engineered-spanloss": 12.2,
 240             "link-concatenation": [{
 241                 "SRLG-Id": 0,
 242                 "fiber-type": "smf",
 243                 "SRLG-length": 100000,
 244                 "pmd": 0.5}]}}
 245         response = test_utils.add_oms_attr_request(
 246             "ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX", data)
 247         self.assertEqual(response.status_code, requests.codes.created)
 248
 249 # test service-create for Eth service from xpdr to xpdr
 250     def test_11_create_connectivity_service_Ethernet(self):
 251         response = test_utils.transportpce_api_rpc_request(
 252             'tapi-connectivity', 'create-connectivity-service', self.cr_serv_input_data)
 253         time.sleep(self.WAITING)
 254         self.uuid_services.eth = response['output']['service']['uuid']
 255         # pylint: disable=consider-using-f-string
 256
 257         input_dict_1 = {'administrative-state': 'LOCKED',
 258                         'lifecycle-state': 'PLANNED',
 259                         'operational-state': 'DISABLED',
 260                         # 'service-type': 'POINT_TO_POINT_CONNECTIVITY',
 261                         # 'service-layer': 'ETH',
 262                         'layer-protocol-name': 'ETH',
 263                         # 'connectivity-direction': 'BIDIRECTIONAL'
 264                         'direction': 'BIDIRECTIONAL'
 265                         }
 266         input_dict_2 = {'value-name': 'OpenROADM node id',
 267                         'value': 'XPDR-C1-XPDR1'}
 268         input_dict_3 = {'value-name': 'OpenROADM node id',
 269                         'value': 'XPDR-A1-XPDR1'}
 270
 271         self.assertDictEqual(dict(input_dict_1, **response['output']['service']),
 272                              response['output']['service'])
 273         self.assertDictEqual(dict(input_dict_2, **response['output']['service']['end-point'][0]['name'][0]),
 274                              response['output']['service']['end-point'][0]['name'][0])
 275         self.assertDictEqual(dict(input_dict_3, **response['output']['service']['end-point'][1]['name'][0]),
 276                              response['output']['service']['end-point'][1]['name'][0])
 277         # If the gate fails is because of the waiting time not being enough
 278 #        time.sleep(self.WAITING)
 279
 280     def test_12_get_service_Ethernet(self):
 281         response = test_utils.get_ordm_serv_list_attr_request("services", str(self.uuid_services.eth))
 282         self.assertEqual(response['status_code'], requests.codes.ok)
 283         self.assertEqual(response['services'][0]['administrative-state'], 'inService')
 284         self.assertEqual(response['services'][0]['service-name'], str(self.uuid_services.eth))
 285         self.assertEqual(response['services'][0]['connection-type'], 'service')
 286         self.assertEqual(response['services'][0]['lifecycle-state'], 'planned')
 287         time.sleep(1)
 288
 289     def test_13_get_connectivity_service_Ethernet(self):
 290         self.tapi_serv_details["uuid"] = str(self.uuid_services.eth)
 291         response = test_utils.transportpce_api_rpc_request(
 292             'tapi-connectivity', 'get-connectivity-service-details', self.tapi_serv_details)
 293         self.assertEqual(response['status_code'], requests.codes.ok)
 294         self.assertEqual(response['output']['service']['operational-state'], 'ENABLED')
 295         self.assertEqual(response['output']['service']['name'][0]['value'], self.uuid_services.eth)
 296         self.assertEqual(response['output']['service']['administrative-state'], 'UNLOCKED')
 297         self.assertEqual(response['output']['service']['lifecycle-state'], 'INSTALLED')
 298
 299     def test_14_change_status_line_port_xpdrc(self):
 300         self.assertTrue(test_utils.sims_update_cp_port(('xpdrc', self.NODE_VERSION_221), '1/0/1-PLUG-NET', '1',
 301                                                        {
 302             "port-name": "1",
 303             "port-type": "CFP2",
 304             "administrative-state": "outOfService",
 305             "port-qual": "xpdr-network"
 306         }))
 307         time.sleep(2)
 308
 309     def test_15_check_update_portmapping(self):
 310         response = test_utils.get_portmapping_node_attr("XPDR-C1", None, None)
 311         self.assertEqual(response['status_code'], requests.codes.ok)
 312         mapping_list = response['nodes'][0]['mapping']
 313         for mapping in mapping_list:
 314             if mapping['logical-connection-point'] == 'XPDR1-NETWORK1':
 315                 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
 316                                  "Operational State should be 'OutOfService'")
 317                 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
 318                                  "Administrative State should be 'OutOfService'")
 319             else:
 320                 self.assertEqual(mapping['port-oper-state'], 'InService',
 321                                  "Operational State should be 'InService'")
 322                 self.assertEqual(mapping['port-admin-state'], 'InService',
 323                                  "Administrative State should be 'InService'")
 324         time.sleep(1)
 325
 326     def test_16_check_update_openroadm_topo(self):
 327         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
 328         self.assertEqual(response['status_code'], requests.codes.ok)
 329         node_list = response['network'][0]['node']
 330         nb_updated_tp = 0
 331         for node in node_list:
 332             self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
 333             self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
 334             tp_list = node['ietf-network-topology:termination-point']
 335             for tp in tp_list:
 336                 if node['node-id'] == 'XPDR-C1-XPDR1' and tp['tp-id'] == 'XPDR1-NETWORK1':
 337                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
 338                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
 339                     nb_updated_tp += 1
 340                 else:
 341                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
 342                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
 343         self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
 344
 345         link_list = response['network'][0]['ietf-network-topology:link']
 346         updated_links = ['XPDR-C1-XPDR1-XPDR1-NETWORK1toROADM-C1-SRG1-SRG1-PP1-TXRX',
 347                          'ROADM-C1-SRG1-SRG1-PP1-TXRXtoXPDR-C1-XPDR1-XPDR1-NETWORK1']
 348         nb_updated_link = 0
 349         for link in link_list:
 350             if link['link-id'] in updated_links:
 351                 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'outOfService')
 352                 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'outOfService')
 353                 nb_updated_link += 1
 354             else:
 355                 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
 356                 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
 357         self.assertEqual(nb_updated_link, 2, "Only two xponder-output/input links should have been modified")
 358         time.sleep(10)
 359
 360     def test_17_check_update_tapi_neps(self):
 361         self.node_details["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
 362 #        self.node_details["node-id"] = "XPDR-C1-XPDR1+OTSi"
 363         self.node_details["node-id"] = "1770bea4-b1da-3b20-abce-7d182c0ec0df"
 364         response = test_utils.transportpce_api_rpc_request(
 365             'tapi-topology', 'get-node-details', self.node_details)
 366         self.assertEqual(response['status_code'], requests.codes.ok)
 367         nep_list = response['output']['node']['owned-node-edge-point']
 368         nb_updated_neps = 0
 369         for nep in nep_list:
 370             if 'XPDR1-NETWORK1' in nep['name'][0]['value']:
 371                 self.assertEqual(nep['operational-state'], 'DISABLED',
 372                                  "Operational State should be 'DISABLED'")
 373                 self.assertEqual(nep['administrative-state'], 'LOCKED',
 374                                  "Administrative State should be 'LOCKED'")
 375                 nb_updated_neps += 1
 376             else:
 377                 self.assertEqual(nep['operational-state'], 'ENABLED',
 378                                  "Operational State should be 'ENABLED'")
 379                 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
 380                                  "Administrative State should be 'UNLOCKED'")
 381         self.node_details["node-id"] = "XPDR-C1-XPDR1+DSR"
 382 #        response = test_utils.transportpce_api_rpc_request(
 383 #            'tapi-topology', 'get-node-details', self.node_details)
 384 #        self.assertEqual(response['status_code'], requests.codes.ok)
 385 #        nep_list = response['output']['node']['owned-node-edge-point']
 386 #        for nep in nep_list:
 387 #            if 'XPDR1-NETWORK1' in nep['name'][0]['value']:
 388 #                self.assertEqual(nep['operational-state'], 'DISABLED',
 389 #                                 "Operational State should be 'DISABLED'")
 390 #                self.assertEqual(nep['administrative-state'], 'LOCKED',
 391 #                                 "Administrative State should be 'LOCKED'")
 392 #                nb_updated_neps += 1
 393 #            else:
 394 #                self.assertEqual(nep['operational-state'], 'ENABLED',
 395 #                                 "Operational State should be 'ENABLED'")
 396 #                self.assertEqual(nep['administrative-state'], 'UNLOCKED',
 397 #                                 "Administrative State should be 'UNLOCKED'")
 398         self.assertEqual(nb_updated_neps, 3, "3 xponder neps (OTS, OTSI_MC, eODU) should have been modified")
 399         time.sleep(1)
 400
 401     def test_18_check_update_tapi_links(self):
 402         self.tapi_topo["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
 403         response = test_utils.transportpce_api_rpc_request(
 404             'tapi-topology', 'get-topology-details', self.tapi_topo)
 405         time.sleep(2)
 406         self.assertEqual(response['status_code'], requests.codes.ok)
 407         link_list = response['output']['topology']['link']
 408         nb_updated_link = 0
 409         for link in link_list:
 410             if all(x in link['name'][0]['value'] for x in ['XPDR-C1-XPDR1', 'XPDR1-NETWORK1']):
 411                 self.assertEqual(link['operational-state'], 'DISABLED')
 412                 self.assertEqual(link['administrative-state'], 'LOCKED')
 413                 nb_updated_link += 1
 414             else:
 415                 self.assertEqual(link['operational-state'], 'ENABLED')
 416                 self.assertEqual(link['administrative-state'], 'UNLOCKED')
 417         self.assertEqual(nb_updated_link, 1,
 418                          "Only one xponder-output/input bidirectional link should have been modified")
 419         time.sleep(1)
 420
 421     def test_19_check_update_service_Ethernet(self):
 422         response = test_utils.get_ordm_serv_list_attr_request("services", str(self.uuid_services.eth))
 423         self.assertEqual(response['status_code'], requests.codes.ok)
 424         self.assertEqual(response['services'][0]['operational-state'], 'outOfService')
 425         self.assertEqual(response['services'][0]['administrative-state'], 'inService')
 426
 427     def test_20_check_update_connectivity_service_Ethernet(self):
 428         self.tapi_serv_details["uuid"] = str(self.uuid_services.eth)
 429         response = test_utils.transportpce_api_rpc_request(
 430             'tapi-connectivity', 'get-connectivity-service-details', self.tapi_serv_details)
 431         self.assertEqual(response['status_code'], requests.codes.ok)
 432         self.assertEqual(response['output']['service']['operational-state'], 'DISABLED')
 433         self.assertEqual(response['output']['service']['administrative-state'], 'LOCKED')
 434         time.sleep(1)
 435
 436     def test_21_restore_status_line_port_xpdrc(self):
 437         self.assertTrue(test_utils.sims_update_cp_port(('xpdrc', self.NODE_VERSION_221), '1/0/1-PLUG-NET', '1',
 438                                                        {
 439             "port-name": "1",
 440             "port-type": "CFP2",
 441             "administrative-state": "inService",
 442             "port-qual": "xpdr-network"
 443         }))
 444         time.sleep(2)
 445
 446     def test_22_check_update_portmapping_ok(self):
 447         response = test_utils.get_portmapping_node_attr("XPDR-C1", None, None)
 448         self.assertEqual(response['status_code'], requests.codes.ok)
 449         mapping_list = response['nodes'][0]['mapping']
 450         for mapping in mapping_list:
 451             self.assertEqual(mapping['port-oper-state'], 'InService',
 452                              "Operational State should be 'InService'")
 453             self.assertEqual(mapping['port-admin-state'], 'InService',
 454                              "Administrative State should be 'InService'")
 455         time.sleep(1)
 456
 457     def test_23_check_update_openroadm_topo_ok(self):
 458         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
 459         self.assertEqual(response['status_code'], requests.codes.ok)
 460         node_list = response['network'][0]['node']
 461         for node in node_list:
 462             self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
 463             self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
 464             tp_list = node['ietf-network-topology:termination-point']
 465             for tp in tp_list:
 466                 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
 467                 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
 468
 469         link_list = response['network'][0]['ietf-network-topology:link']
 470         for link in link_list:
 471             self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
 472             self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
 473         time.sleep(1)
 474
 475     def test_24_check_update_tapi_neps_ok(self):
 476         self.node_details["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
 477 #        self.node_details["node-id"] = "XPDR-C1-XPDR1+OTSi"
 478         self.node_details["node-id"] = "1770bea4-b1da-3b20-abce-7d182c0ec0df"
 479         response = test_utils.transportpce_api_rpc_request(
 480             'tapi-topology', 'get-node-details', self.node_details)
 481         self.assertEqual(response['status_code'], requests.codes.ok)
 482         nep_list = response['output']['node']['owned-node-edge-point']
 483         for nep in nep_list:
 484             self.assertEqual(nep['operational-state'], 'ENABLED',
 485                              "Operational State should be 'ENABLED'")
 486             self.assertEqual(nep['administrative-state'], 'UNLOCKED',
 487                              "Administrative State should be 'UNLOCKED'")
 488
 489 #        self.node_details["node-id"] = "XPDR-C1-XPDR1+DSR"
 490 #        response = test_utils.transportpce_api_rpc_request(
 491 #            'tapi-topology', 'get-node-details', self.node_details)
 492 #        self.assertEqual(response['status_code'], requests.codes.ok)
 493 #        nep_list = response['output']['node']['owned-node-edge-point']
 494 #        for nep in nep_list:
 495 #            self.assertEqual(nep['operational-state'], 'ENABLED',
 496 #                             "Operational State should be 'ENABLED'")
 497 #            self.assertEqual(nep['administrative-state'], 'UNLOCKED',
 498 #                             "Administrative State should be 'UNLOCKED'")
 499         time.sleep(1)
 500
 501     def test_25_check_update_tapi_links_ok(self):
 502         self.tapi_topo["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
 503         response = test_utils.transportpce_api_rpc_request(
 504             'tapi-topology', 'get-topology-details', self.tapi_topo)
 505         time.sleep(2)
 506         link_list = response['output']['topology']['link']
 507         for link in link_list:
 508             self.assertEqual(link['operational-state'], 'ENABLED')
 509             self.assertEqual(link['administrative-state'], 'UNLOCKED')
 510         time.sleep(1)
 511
 512     def test_26_check_update_service1_ok(self):
 513         self.test_12_get_service_Ethernet()
 514
 515     def test_27_check_update_connectivity_service_Ethernet_ok(self):
 516         self.test_13_get_connectivity_service_Ethernet()
 517
 518     def test_28_change_status_port_roadma_srg(self):
 519         self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3/0', 'C1',
 520                                                        {
 521             "port-name": "C1",
 522             "logical-connection-point": "SRG1-PP1",
 523             "port-type": "client",
 524             "circuit-id": "SRG1",
 525             "administrative-state": "outOfService",
 526             "port-qual": "roadm-external"
 527         }))
 528         time.sleep(2)
 529
 530     def test_29_check_update_portmapping(self):
 531         response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
 532         self.assertEqual(response['status_code'], requests.codes.ok)
 533         mapping_list = response['nodes'][0]['mapping']
 534         for mapping in mapping_list:
 535             if mapping['logical-connection-point'] == 'SRG1-PP1-TXRX':
 536                 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
 537                                  "Operational State should be 'OutOfService'")
 538                 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
 539                                  "Administrative State should be 'OutOfService'")
 540             else:
 541                 self.assertEqual(mapping['port-oper-state'], 'InService',
 542                                  "Operational State should be 'InService'")
 543                 self.assertEqual(mapping['port-admin-state'], 'InService',
 544                                  "Administrative State should be 'InService'")
 545         time.sleep(1)
 546
 547     def test_30_check_update_openroadm_topo(self):
 548         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
 549         self.assertEqual(response['status_code'], requests.codes.ok)
 550         node_list = response['network'][0]['node']
 551         nb_updated_tp = 0
 552         for node in node_list:
 553             self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
 554             self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
 555             tp_list = node['ietf-network-topology:termination-point']
 556             for tp in tp_list:
 557                 if node['node-id'] == 'ROADM-A1-SRG1' and tp['tp-id'] == 'SRG1-PP1-TXRX':
 558                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
 559                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
 560                     nb_updated_tp += 1
 561                 else:
 562                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
 563                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
 564         self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
 565
 566         link_list = response['network'][0]['ietf-network-topology:link']
 567         updated_links = ['XPDR-A1-XPDR1-XPDR1-NETWORK1toROADM-A1-SRG1-SRG1-PP1-TXRX',
 568                          'ROADM-A1-SRG1-SRG1-PP1-TXRXtoXPDR-A1-XPDR1-XPDR1-NETWORK1']
 569         nb_updated_link = 0
 570         for link in link_list:
 571             if link['link-id'] in updated_links:
 572                 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'outOfService')
 573                 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'outOfService')
 574                 nb_updated_link += 1
 575             else:
 576                 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
 577                 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
 578         self.assertEqual(nb_updated_link, 2, "Only two xponder-output/input links should have been modified")
 579         time.sleep(1)
 580
 581     def test_31_check_update_tapi_neps(self):
 582         self.node_details["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
 583 #        self.node_details["node-id"] = "ROADM-A1+PHOTONIC_MEDIA"
 584         self.node_details["node-id"] = "3b726367-6f2d-3e3f-9033-d99b61459075"
 585         response = test_utils.transportpce_api_rpc_request(
 586             'tapi-topology', 'get-node-details', self.node_details)
 587         self.assertEqual(response['status_code'], requests.codes.ok)
 588         nep_list = response['output']['node']['owned-node-edge-point']
 589         nb_updated_neps = 0
 590         for nep in nep_list:
 591             if 'SRG1-PP1-TXRX' in nep['name'][0]['value']:
 592                 self.assertEqual(nep['operational-state'], 'DISABLED',
 593                                  "Operational State should be 'DISABLED'")
 594                 self.assertEqual(nep['administrative-state'], 'LOCKED',
 595                                  "Administrative State should be 'LOCKED'")
 596                 nb_updated_neps += 1
 597             else:
 598                 self.assertEqual(nep['operational-state'], 'ENABLED',
 599                                  "Operational State should be 'ENABLED'")
 600                 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
 601                                  "Administrative State should be 'UNLOCKED'")
 602         self.assertEqual(nb_updated_neps, 2, "Only 2 roadm SRG-PP nep (OTS/MC)should have been modified")
 603         time.sleep(1)
 604
 605     def test_32_check_update_tapi_links(self):
 606         self.tapi_topo["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
 607         response = test_utils.transportpce_api_rpc_request(
 608             'tapi-topology', 'get-topology-details', self.tapi_topo)
 609         time.sleep(2)
 610         link_list = response['output']['topology']['link']
 611         nb_updated_link = 0
 612         for link in link_list:
 613             if all(x in link['name'][0]['value'] for x in ['ROADM-A1', 'SRG1-PP1-TXRX']):
 614                 self.assertEqual(link['operational-state'], 'DISABLED')
 615                 self.assertEqual(link['administrative-state'], 'LOCKED')
 616                 nb_updated_link += 1
 617             else:
 618                 self.assertEqual(link['operational-state'], 'ENABLED')
 619                 self.assertEqual(link['administrative-state'], 'UNLOCKED')
 620         self.assertEqual(nb_updated_link, 1,
 621                          "Only one xponder-output/input link should have been modified")
 622         time.sleep(1)
 623
 624     def test_33_check_update_service_Ethernet(self):
 625         self.test_19_check_update_service_Ethernet()
 626
 627     def test_34_check_update_connectivity_service_Ethernet(self):
 628         self.test_20_check_update_connectivity_service_Ethernet()
 629
 630     def test_35_restore_status_port_roadma_srg(self):
 631         self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3/0', 'C1',
 632                                                        {
 633             "port-name": "C1",
 634             "logical-connection-point": "SRG1-PP1",
 635             "port-type": "client",
 636             "circuit-id": "SRG1",
 637             "administrative-state": "inService",
 638             "port-qual": "roadm-external"
 639         }))
 640         time.sleep(2)
 641
 642     def test_36_check_update_portmapping_ok(self):
 643         response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
 644         self.assertEqual(response['status_code'], requests.codes.ok)
 645         mapping_list = response['nodes'][0]['mapping']
 646         for mapping in mapping_list:
 647             self.assertEqual(mapping['port-oper-state'], 'InService',
 648                              "Operational State should be 'InService'")
 649             self.assertEqual(mapping['port-admin-state'], 'InService',
 650                              "Administrative State should be 'InService'")
 651         time.sleep(1)
 652
 653     def test_37_check_update_openroadm_topo_ok(self):
 654         self.test_23_check_update_openroadm_topo_ok()
 655
 656     def test_38_check_update_tapi_neps_ok(self):
 657         self.node_details["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
 658 #        self.node_details["node-id"] = "ROADM-A1+PHOTONIC_MEDIA"
 659         self.node_details["node-id"] = "3b726367-6f2d-3e3f-9033-d99b61459075"
 660         response = test_utils.transportpce_api_rpc_request(
 661             'tapi-topology', 'get-node-details', self.node_details)
 662         self.assertEqual(response['status_code'], requests.codes.ok)
 663         nep_list = response['output']['node']['owned-node-edge-point']
 664         for nep in nep_list:
 665             self.assertEqual(nep['operational-state'], 'ENABLED',
 666                              "Operational State should be 'ENABLED'")
 667             self.assertEqual(nep['administrative-state'], 'UNLOCKED',
 668                              "Administrative State should be 'UNLOCKED'")
 669
 670         time.sleep(1)
 671
 672     def test_39_check_update_tapi_links_ok(self):
 673         self.test_25_check_update_tapi_links_ok()
 674
 675     def test_40_check_update_service1_ok(self):
 676         self.test_12_get_service_Ethernet()
 677
 678     def test_41_check_update_connectivity_service_Ethernet_ok(self):
 679         self.test_13_get_connectivity_service_Ethernet()
 680
 681     def test_42_change_status_line_port_roadma_deg(self):
 682         self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '2/0', 'L1',
 683                                                        {
 684             "port-name": "L1",
 685             "logical-connection-point": "DEG2-TTP-TXRX",
 686             "port-type": "LINE",
 687             "circuit-id": "1",
 688             "administrative-state": "outOfService",
 689             "port-qual": "roadm-external"
 690         }))
 691         time.sleep(2)
 692
 693     def test_43_check_update_portmapping(self):
 694         response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
 695         self.assertEqual(response['status_code'], requests.codes.ok)
 696         mapping_list = response['nodes'][0]['mapping']
 697         for mapping in mapping_list:
 698             if mapping['logical-connection-point'] == 'DEG2-TTP-TXRX':
 699                 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
 700                                  "Operational State should be 'OutOfService'")
 701                 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
 702                                  "Administrative State should be 'OutOfService'")
 703             else:
 704                 self.assertEqual(mapping['port-oper-state'], 'InService',
 705                                  "Operational State should be 'InService'")
 706                 self.assertEqual(mapping['port-admin-state'], 'InService',
 707                                  "Administrative State should be 'InService'")
 708         time.sleep(1)
 709
 710     def test_44_check_update_openroadm_topo(self):
 711         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
 712         self.assertEqual(response['status_code'], requests.codes.ok)
 713         node_list = response['network'][0]['node']
 714         nb_updated_tp = 0
 715         for node in node_list:
 716             self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
 717             self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
 718             tp_list = node['ietf-network-topology:termination-point']
 719             for tp in tp_list:
 720                 if node['node-id'] == 'ROADM-A1-DEG2' and tp['tp-id'] == 'DEG2-TTP-TXRX':
 721                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
 722                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
 723                     nb_updated_tp += 1
 724                 else:
 725                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
 726                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
 727         self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
 728
 729         link_list = response['network'][0]['ietf-network-topology:link']
 730         updated_links = ['ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX',
 731                          'ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX']
 732         nb_updated_link = 0
 733         for link in link_list:
 734             if link['link-id'] in updated_links:
 735                 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'outOfService')
 736                 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'outOfService')
 737                 nb_updated_link += 1
 738             else:
 739                 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
 740                 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
 741         self.assertEqual(nb_updated_link, 2, "Only two xponder-output/input links should have been modified")
 742         time.sleep(1)
 743
 744     def test_45_check_update_tapi_neps(self):
 745         self.node_details["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
 746 #        self.node_details["node-id"] = "ROADM-A1+PHOTONIC_MEDIA"
 747         self.node_details["node-id"] = "3b726367-6f2d-3e3f-9033-d99b61459075"
 748         response = test_utils.transportpce_api_rpc_request(
 749             'tapi-topology', 'get-node-details', self.node_details)
 750         self.assertEqual(response['status_code'], requests.codes.ok)
 751         nep_list = response['output']['node']['owned-node-edge-point']
 752         nb_updated_neps = 0
 753         for nep in nep_list:
 754             if 'DEG2-TTP-TXRX' in nep['name'][0]['value']:
 755                 self.assertEqual(nep['operational-state'], 'DISABLED',
 756                                  "Operational State should be 'DISABLED'")
 757                 self.assertEqual(nep['administrative-state'], 'LOCKED',
 758                                  "Administrative State should be 'LOCKED'")
 759                 nb_updated_neps += 1
 760             else:
 761                 self.assertEqual(nep['operational-state'], 'ENABLED',
 762                                  "Operational State should be 'ENABLED'")
 763                 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
 764                                  "Administrative State should be 'UNLOCKED'")
 765         self.assertEqual(nb_updated_neps, 4, "4 roadm NEPS should have been modified (OTS/OMS/MC/OTSI_MC")
 766         time.sleep(1)
 767
 768     def test_46_check_update_tapi_links(self):
 769         self.tapi_topo["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
 770         response = test_utils.transportpce_api_rpc_request(
 771             'tapi-topology', 'get-topology-details', self.tapi_topo)
 772         time.sleep(2)
 773         self.assertEqual(response['status_code'], requests.codes.ok)
 774         link_list = response['output']['topology']['link']
 775         nb_updated_link = 0
 776         for link in link_list:
 777             if all(x in link['name'][0]['value'] for x in ['ROADM-A1', 'DEG2-TTP-TXRX']):
 778                 self.assertEqual(link['operational-state'], 'DISABLED')
 779                 self.assertEqual(link['administrative-state'], 'LOCKED')
 780                 nb_updated_link += 1
 781             else:
 782                 self.assertEqual(link['operational-state'], 'ENABLED')
 783                 self.assertEqual(link['administrative-state'], 'UNLOCKED')
 784         self.assertEqual(nb_updated_link, 1,
 785                          "Only one rdm-rdm link should have been modified")
 786         time.sleep(1)
 787
 788     def test_47_check_update_service_Ethernet(self):
 789         self.test_19_check_update_service_Ethernet()
 790
 791     def test_48_check_update_connectivity_service_Ethernet(self):
 792         self.test_20_check_update_connectivity_service_Ethernet()
 793
 794     def test_49_restore_status_line_port_roadma_deg(self):
 795         self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '2/0', 'L1',
 796                                                        {
 797             "port-name": "L1",
 798             "logical-connection-point": "DEG2-TTP-TXRX",
 799             "port-type": "LINE",
 800             "circuit-id": "1",
 801             "administrative-state": "inService",
 802             "port-qual": "roadm-external"
 803         }))
 804         time.sleep(2)
 805
 806     def test_50_check_update_portmapping_ok(self):
 807         self.test_36_check_update_portmapping_ok()
 808
 809     def test_51_check_update_openroadm_topo_ok(self):
 810         self.test_23_check_update_openroadm_topo_ok()
 811
 812     def test_52_check_update_tapi_neps_ok(self):
 813         self.test_38_check_update_tapi_neps_ok()
 814
 815     def test_53_check_update_tapi_links_ok(self):
 816         self.test_25_check_update_tapi_links_ok()
 817
 818     def test_54_check_update_service1_ok(self):
 819         self.test_12_get_service_Ethernet()
 820
 821     def test_55_check_update_connectivity_service_Ethernet_ok(self):
 822         self.test_13_get_connectivity_service_Ethernet()
 823
 824     def test_56_change_status_port_roadma_srg(self):
 825         self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3/0', 'C2',
 826                                                        {
 827             "port-name": "C2",
 828             "logical-connection-point": "SRG1-PP2",
 829             "port-type": "client",
 830             "circuit-id": "SRG1",
 831             "administrative-state": "outOfService",
 832             "port-qual": "roadm-external"
 833         }))
 834         time.sleep(2)
 835
 836     def test_57_check_update_portmapping(self):
 837         response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
 838         self.assertEqual(response['status_code'], requests.codes.ok)
 839         mapping_list = response['nodes'][0]['mapping']
 840         for mapping in mapping_list:
 841             if mapping['logical-connection-point'] == 'SRG1-PP2-TXRX':
 842                 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
 843                                  "Operational State should be 'OutOfService'")
 844                 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
 845                                  "Administrative State should be 'OutOfService'")
 846             else:
 847                 self.assertEqual(mapping['port-oper-state'], 'InService',
 848                                  "Operational State should be 'InService'")
 849                 self.assertEqual(mapping['port-admin-state'], 'InService',
 850                                  "Administrative State should be 'InService'")
 851         time.sleep(1)
 852
 853     def test_58_check_update_openroadm_topo(self):
 854         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
 855         self.assertEqual(response['status_code'], requests.codes.ok)
 856         node_list = response['network'][0]['node']
 857         nb_updated_tp = 0
 858         for node in node_list:
 859             self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
 860             self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
 861             tp_list = node['ietf-network-topology:termination-point']
 862             for tp in tp_list:
 863                 if node['node-id'] == 'ROADM-A1-SRG1' and tp['tp-id'] == 'SRG1-PP2-TXRX':
 864                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
 865                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
 866                     nb_updated_tp += 1
 867                 else:
 868                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
 869                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
 870         self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
 871
 872         link_list = response['network'][0]['ietf-network-topology:link']
 873         nb_updated_link = 0
 874         for link in link_list:
 875             self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
 876             self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
 877         self.assertEqual(nb_updated_link, 0, "No link should have been modified")
 878         time.sleep(1)
 879
 880     def test_59_check_update_tapi_neps(self):
 881         self.node_details["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
 882 #        self.node_details["node-id"] = uuid.UUID(bytes="ROADM-A1+PHOTONIC_MEDIA".bytes)
 883         self.node_details["node-id"] = "3b726367-6f2d-3e3f-9033-d99b61459075"
 884         response = test_utils.transportpce_api_rpc_request(
 885             'tapi-topology', 'get-node-details', self.node_details)
 886         self.assertEqual(response['status_code'], requests.codes.ok)
 887         nep_list = response['output']['node']['owned-node-edge-point']
 888         nb_updated_neps = 0
 889         for nep in nep_list:
 890             if 'SRG1-PP2-TXRX' in nep['name'][0]['value']:
 891                 self.assertEqual(nep['operational-state'], 'DISABLED',
 892                                  "Operational State should be 'DISABLED'")
 893                 self.assertEqual(nep['administrative-state'], 'LOCKED',
 894                                  "Administrative State should be 'LOCKED'")
 895                 nb_updated_neps += 1
 896             else:
 897                 self.assertEqual(nep['operational-state'], 'ENABLED',
 898                                  "Operational State should be 'ENABLED'")
 899                 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
 900                                  "Administrative State should be 'UNLOCKED'")
 901         self.assertEqual(nb_updated_neps, 1, "Only 1 roadm neps OTS should have been modified for SRG2PP")
 902         time.sleep(1)
 903
 904     def test_60_check_update_tapi_links(self):
 905         self.tapi_topo["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
 906         response = test_utils.transportpce_api_rpc_request(
 907             'tapi-topology', 'get-topology-details', self.tapi_topo)
 908         time.sleep(2)
 909         self.assertEqual(response['status_code'], requests.codes.ok)
 910         link_list = response['output']['topology']['link']
 911         nb_updated_link = 0
 912         for link in link_list:
 913             if all(x in link['name'][0]['value'] for x in ['ROADM-A1', 'SRG1-PP2-TXRX']):
 914                 self.assertEqual(link['operational-state'], 'DISABLED')
 915                 self.assertEqual(link['administrative-state'], 'LOCKED')
 916                 nb_updated_link += 1
 917             else:
 918                 self.assertEqual(link['operational-state'], 'ENABLED')
 919                 self.assertEqual(link['administrative-state'], 'UNLOCKED')
 920         self.assertEqual(nb_updated_link, 0,
 921                          "No link should have been modified")
 922         time.sleep(1)
 923
 924     def test_61_check_update_service1_ok(self):
 925         self.test_12_get_service_Ethernet()
 926
 927     def test_62_check_update_connectivity_service_Ethernet_ok(self):
 928         self.test_13_get_connectivity_service_Ethernet()
 929
 930     def test_63_delete_connectivity_service_Ethernet(self):
 931         self.del_serv_input_data["uuid"] = str(self.uuid_services.eth)
 932         response = test_utils.transportpce_api_rpc_request(
 933             'tapi-connectivity', 'delete-connectivity-service', self.del_serv_input_data)
 934         self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content))
 935         time.sleep(self.WAITING)
 936
 937     def test_64_disconnect_xponders_from_roadm(self):
 938         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
 939         self.assertEqual(response['status_code'], requests.codes.ok)
 940         links = response['network'][0]['ietf-network-topology:link']
 941         for link in links:
 942             if link["org-openroadm-common-network:link-type"] in ('XPONDER-OUTPUT', 'XPONDER-INPUT'):
 943                 response = test_utils.del_ietf_network_link_request(
 944                     'openroadm-topology', link['link-id'], 'config')
 945                 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 946
 947     def test_65_disconnect_XPDRA(self):
 948         response = test_utils.unmount_device("XPDR-A1")
 949         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 950
 951     def test_66_disconnect_XPDRC(self):
 952         response = test_utils.unmount_device("XPDR-C1")
 953         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 954
 955     def test_67_disconnect_ROADMA(self):
 956         response = test_utils.unmount_device("ROADM-A1")
 957         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 958
 959     def test_68_disconnect_ROADMC(self):
 960         response = test_utils.unmount_device("ROADM-C1")
 961         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 962
 963     def test_69_restore_status_port_roadma_srg(self):
 964         self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3/0', 'C2',
 965                                                        {
 966             "port-name": "C2",
 967             "logical-connection-point": "SRG1-PP2",
 968             "port-type": "client",
 969             "circuit-id": "SRG1",
 970             "administrative-state": "inService",
 971             "port-qual": "roadm-external"
 972         }))
 973         time.sleep(2)
 974
 975     def test_70_clean_openroadm_topology(self):
 976         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
 977         self.assertEqual(response['status_code'], requests.codes.ok)
 978         links = response['network'][0]['ietf-network-topology:link']
 979         for link in links:
 980             if link["org-openroadm-common-network:link-type"] in ('XPONDER-OUTPUT', 'XPONDER-INPUT', 'ROADM-TO-ROADM'):
 981                 response = test_utils.del_ietf_network_link_request('openroadm-topology', link['link-id'], 'config')
 982                 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 983
 984
 985 if __name__ == "__main__":
 986     unittest.main(verbosity=2)