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         print(response['output']['topology']['link'])
 409         nb_updated_link = 0
 410         for link in link_list:
 411             if all(x in link['name'][0]['value'] for x in ['XPDR-C1-XPDR1', 'XPDR1-NETWORK1']):
 412                 self.assertEqual(link['operational-state'], 'DISABLED')
 413                 self.assertEqual(link['administrative-state'], 'LOCKED')
 414                 nb_updated_link += 1
 415             else:
 416                 self.assertEqual(link['operational-state'], 'ENABLED')
 417                 self.assertEqual(link['administrative-state'], 'UNLOCKED')
 418         self.assertEqual(nb_updated_link, 1,
 419                          "Only one xponder-output/input bidirectional link should have been modified")
 420         time.sleep(1)
 421
 422     def test_19_check_update_service_Ethernet(self):
 423         response = test_utils.get_ordm_serv_list_attr_request("services", str(self.uuid_services.eth))
 424         self.assertEqual(response['status_code'], requests.codes.ok)
 425         self.assertEqual(response['services'][0]['operational-state'], 'outOfService')
 426         self.assertEqual(response['services'][0]['administrative-state'], 'inService')
 427
 428     def test_20_check_update_connectivity_service_Ethernet(self):
 429         self.tapi_serv_details["uuid"] = str(self.uuid_services.eth)
 430         print(str(self.uuid_services.eth))
 431         response = test_utils.transportpce_api_rpc_request(
 432             'tapi-connectivity', 'get-connectivity-service-details', self.tapi_serv_details)
 433         self.assertEqual(response['status_code'], requests.codes.ok)
 434         self.assertEqual(response['output']['service']['operational-state'], 'DISABLED')
 435         self.assertEqual(response['output']['service']['administrative-state'], 'LOCKED')
 436         time.sleep(1)
 437
 438     def test_21_restore_status_line_port_xpdrc(self):
 439         self.assertTrue(test_utils.sims_update_cp_port(('xpdrc', self.NODE_VERSION_221), '1/0/1-PLUG-NET', '1',
 440                                                        {
 441             "port-name": "1",
 442             "port-type": "CFP2",
 443             "administrative-state": "inService",
 444             "port-qual": "xpdr-network"
 445         }))
 446         time.sleep(2)
 447
 448     def test_22_check_update_portmapping_ok(self):
 449         response = test_utils.get_portmapping_node_attr("XPDR-C1", None, None)
 450         self.assertEqual(response['status_code'], requests.codes.ok)
 451         mapping_list = response['nodes'][0]['mapping']
 452         for mapping in mapping_list:
 453             self.assertEqual(mapping['port-oper-state'], 'InService',
 454                              "Operational State should be 'InService'")
 455             self.assertEqual(mapping['port-admin-state'], 'InService',
 456                              "Administrative State should be 'InService'")
 457         time.sleep(1)
 458
 459     def test_23_check_update_openroadm_topo_ok(self):
 460         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
 461         self.assertEqual(response['status_code'], requests.codes.ok)
 462         node_list = response['network'][0]['node']
 463         for node in node_list:
 464             self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
 465             self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
 466             tp_list = node['ietf-network-topology:termination-point']
 467             for tp in tp_list:
 468                 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
 469                 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
 470
 471         link_list = response['network'][0]['ietf-network-topology:link']
 472         for link in link_list:
 473             self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
 474             self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
 475         time.sleep(1)
 476
 477     def test_24_check_update_tapi_neps_ok(self):
 478         self.node_details["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
 479 #        self.node_details["node-id"] = "XPDR-C1-XPDR1+OTSi"
 480         self.node_details["node-id"] = "1770bea4-b1da-3b20-abce-7d182c0ec0df"
 481         response = test_utils.transportpce_api_rpc_request(
 482             'tapi-topology', 'get-node-details', self.node_details)
 483         self.assertEqual(response['status_code'], requests.codes.ok)
 484         nep_list = response['output']['node']['owned-node-edge-point']
 485         for nep in nep_list:
 486             self.assertEqual(nep['operational-state'], 'ENABLED',
 487                              "Operational State should be 'ENABLED'")
 488             self.assertEqual(nep['administrative-state'], 'UNLOCKED',
 489                              "Administrative State should be 'UNLOCKED'")
 490
 491 #        self.node_details["node-id"] = "XPDR-C1-XPDR1+DSR"
 492 #        response = test_utils.transportpce_api_rpc_request(
 493 #            'tapi-topology', 'get-node-details', self.node_details)
 494 #        self.assertEqual(response['status_code'], requests.codes.ok)
 495 #        nep_list = response['output']['node']['owned-node-edge-point']
 496 #        for nep in nep_list:
 497 #            self.assertEqual(nep['operational-state'], 'ENABLED',
 498 #                             "Operational State should be 'ENABLED'")
 499 #            self.assertEqual(nep['administrative-state'], 'UNLOCKED',
 500 #                             "Administrative State should be 'UNLOCKED'")
 501         time.sleep(1)
 502
 503     def test_25_check_update_tapi_links_ok(self):
 504         self.tapi_topo["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
 505         response = test_utils.transportpce_api_rpc_request(
 506             'tapi-topology', 'get-topology-details', self.tapi_topo)
 507         time.sleep(2)
 508         link_list = response['output']['topology']['link']
 509         for link in link_list:
 510             self.assertEqual(link['operational-state'], 'ENABLED')
 511             self.assertEqual(link['administrative-state'], 'UNLOCKED')
 512         time.sleep(1)
 513
 514     def test_26_check_update_service1_ok(self):
 515         self.test_12_get_service_Ethernet()
 516
 517     def test_27_check_update_connectivity_service_Ethernet_ok(self):
 518         self.test_13_get_connectivity_service_Ethernet()
 519
 520     def test_28_change_status_port_roadma_srg(self):
 521         self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3/0', 'C1',
 522                                                        {
 523             "port-name": "C1",
 524             "logical-connection-point": "SRG1-PP1",
 525             "port-type": "client",
 526             "circuit-id": "SRG1",
 527             "administrative-state": "outOfService",
 528             "port-qual": "roadm-external"
 529         }))
 530         time.sleep(2)
 531
 532     def test_29_check_update_portmapping(self):
 533         response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
 534         self.assertEqual(response['status_code'], requests.codes.ok)
 535         mapping_list = response['nodes'][0]['mapping']
 536         for mapping in mapping_list:
 537             if mapping['logical-connection-point'] == 'SRG1-PP1-TXRX':
 538                 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
 539                                  "Operational State should be 'OutOfService'")
 540                 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
 541                                  "Administrative State should be 'OutOfService'")
 542             else:
 543                 self.assertEqual(mapping['port-oper-state'], 'InService',
 544                                  "Operational State should be 'InService'")
 545                 self.assertEqual(mapping['port-admin-state'], 'InService',
 546                                  "Administrative State should be 'InService'")
 547         time.sleep(1)
 548
 549     def test_30_check_update_openroadm_topo(self):
 550         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
 551         self.assertEqual(response['status_code'], requests.codes.ok)
 552         node_list = response['network'][0]['node']
 553         nb_updated_tp = 0
 554         for node in node_list:
 555             self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
 556             self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
 557             tp_list = node['ietf-network-topology:termination-point']
 558             for tp in tp_list:
 559                 if node['node-id'] == 'ROADM-A1-SRG1' and tp['tp-id'] == 'SRG1-PP1-TXRX':
 560                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
 561                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
 562                     nb_updated_tp += 1
 563                 else:
 564                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
 565                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
 566         self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
 567
 568         link_list = response['network'][0]['ietf-network-topology:link']
 569         updated_links = ['XPDR-A1-XPDR1-XPDR1-NETWORK1toROADM-A1-SRG1-SRG1-PP1-TXRX',
 570                          'ROADM-A1-SRG1-SRG1-PP1-TXRXtoXPDR-A1-XPDR1-XPDR1-NETWORK1']
 571         nb_updated_link = 0
 572         for link in link_list:
 573             if link['link-id'] in updated_links:
 574                 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'outOfService')
 575                 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'outOfService')
 576                 nb_updated_link += 1
 577             else:
 578                 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
 579                 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
 580         self.assertEqual(nb_updated_link, 2, "Only two xponder-output/input links should have been modified")
 581         time.sleep(1)
 582
 583     def test_31_check_update_tapi_neps(self):
 584         self.node_details["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
 585 #        self.node_details["node-id"] = "ROADM-A1+PHOTONIC_MEDIA"
 586         self.node_details["node-id"] = "3b726367-6f2d-3e3f-9033-d99b61459075"
 587         response = test_utils.transportpce_api_rpc_request(
 588             'tapi-topology', 'get-node-details', self.node_details)
 589         self.assertEqual(response['status_code'], requests.codes.ok)
 590         nep_list = response['output']['node']['owned-node-edge-point']
 591         nb_updated_neps = 0
 592         for nep in nep_list:
 593             if 'SRG1-PP1-TXRX' in nep['name'][0]['value']:
 594                 self.assertEqual(nep['operational-state'], 'DISABLED',
 595                                  "Operational State should be 'DISABLED'")
 596                 self.assertEqual(nep['administrative-state'], 'LOCKED',
 597                                  "Administrative State should be 'LOCKED'")
 598                 nb_updated_neps += 1
 599             else:
 600                 self.assertEqual(nep['operational-state'], 'ENABLED',
 601                                  "Operational State should be 'ENABLED'")
 602                 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
 603                                  "Administrative State should be 'UNLOCKED'")
 604         self.assertEqual(nb_updated_neps, 2, "Only 2 roadm SRG-PP nep (OTS/MC)should have been modified")
 605         time.sleep(1)
 606
 607     def test_32_check_update_tapi_links(self):
 608         self.tapi_topo["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
 609         response = test_utils.transportpce_api_rpc_request(
 610             'tapi-topology', 'get-topology-details', self.tapi_topo)
 611         time.sleep(2)
 612         link_list = response['output']['topology']['link']
 613         nb_updated_link = 0
 614         for link in link_list:
 615             if all(x in link['name'][0]['value'] for x in ['ROADM-A1', 'SRG1-PP1-TXRX']):
 616                 self.assertEqual(link['operational-state'], 'DISABLED')
 617                 self.assertEqual(link['administrative-state'], 'LOCKED')
 618                 nb_updated_link += 1
 619             else:
 620                 self.assertEqual(link['operational-state'], 'ENABLED')
 621                 self.assertEqual(link['administrative-state'], 'UNLOCKED')
 622         self.assertEqual(nb_updated_link, 1,
 623                          "Only one xponder-output/input link should have been modified")
 624         time.sleep(1)
 625
 626     def test_33_check_update_service_Ethernet(self):
 627         self.test_19_check_update_service_Ethernet()
 628
 629     def test_34_check_update_connectivity_service_Ethernet(self):
 630         self.test_20_check_update_connectivity_service_Ethernet()
 631
 632     def test_35_restore_status_port_roadma_srg(self):
 633         self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3/0', 'C1',
 634                                                        {
 635             "port-name": "C1",
 636             "logical-connection-point": "SRG1-PP1",
 637             "port-type": "client",
 638             "circuit-id": "SRG1",
 639             "administrative-state": "inService",
 640             "port-qual": "roadm-external"
 641         }))
 642         time.sleep(2)
 643
 644     def test_36_check_update_portmapping_ok(self):
 645         response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
 646         self.assertEqual(response['status_code'], requests.codes.ok)
 647         mapping_list = response['nodes'][0]['mapping']
 648         for mapping in mapping_list:
 649             self.assertEqual(mapping['port-oper-state'], 'InService',
 650                              "Operational State should be 'InService'")
 651             self.assertEqual(mapping['port-admin-state'], 'InService',
 652                              "Administrative State should be 'InService'")
 653         time.sleep(1)
 654
 655     def test_37_check_update_openroadm_topo_ok(self):
 656         self.test_23_check_update_openroadm_topo_ok()
 657
 658     def test_38_check_update_tapi_neps_ok(self):
 659         self.node_details["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
 660 #        self.node_details["node-id"] = "ROADM-A1+PHOTONIC_MEDIA"
 661         self.node_details["node-id"] = "3b726367-6f2d-3e3f-9033-d99b61459075"
 662         response = test_utils.transportpce_api_rpc_request(
 663             'tapi-topology', 'get-node-details', self.node_details)
 664         self.assertEqual(response['status_code'], requests.codes.ok)
 665         nep_list = response['output']['node']['owned-node-edge-point']
 666         for nep in nep_list:
 667             self.assertEqual(nep['operational-state'], 'ENABLED',
 668                              "Operational State should be 'ENABLED'")
 669             self.assertEqual(nep['administrative-state'], 'UNLOCKED',
 670                              "Administrative State should be 'UNLOCKED'")
 671
 672         time.sleep(1)
 673
 674     def test_39_check_update_tapi_links_ok(self):
 675         self.test_25_check_update_tapi_links_ok()
 676
 677     def test_40_check_update_service1_ok(self):
 678         self.test_12_get_service_Ethernet()
 679
 680     def test_41_check_update_connectivity_service_Ethernet_ok(self):
 681         self.test_13_get_connectivity_service_Ethernet()
 682
 683     def test_42_change_status_line_port_roadma_deg(self):
 684         self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '2/0', 'L1',
 685                                                        {
 686             "port-name": "L1",
 687             "logical-connection-point": "DEG2-TTP-TXRX",
 688             "port-type": "LINE",
 689             "circuit-id": "1",
 690             "administrative-state": "outOfService",
 691             "port-qual": "roadm-external"
 692         }))
 693         time.sleep(2)
 694
 695     def test_43_check_update_portmapping(self):
 696         response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
 697         self.assertEqual(response['status_code'], requests.codes.ok)
 698         mapping_list = response['nodes'][0]['mapping']
 699         for mapping in mapping_list:
 700             if mapping['logical-connection-point'] == 'DEG2-TTP-TXRX':
 701                 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
 702                                  "Operational State should be 'OutOfService'")
 703                 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
 704                                  "Administrative State should be 'OutOfService'")
 705             else:
 706                 self.assertEqual(mapping['port-oper-state'], 'InService',
 707                                  "Operational State should be 'InService'")
 708                 self.assertEqual(mapping['port-admin-state'], 'InService',
 709                                  "Administrative State should be 'InService'")
 710         time.sleep(1)
 711
 712     def test_44_check_update_openroadm_topo(self):
 713         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
 714         self.assertEqual(response['status_code'], requests.codes.ok)
 715         node_list = response['network'][0]['node']
 716         nb_updated_tp = 0
 717         for node in node_list:
 718             self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
 719             self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
 720             tp_list = node['ietf-network-topology:termination-point']
 721             for tp in tp_list:
 722                 if node['node-id'] == 'ROADM-A1-DEG2' and tp['tp-id'] == 'DEG2-TTP-TXRX':
 723                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
 724                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
 725                     nb_updated_tp += 1
 726                 else:
 727                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
 728                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
 729         self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
 730
 731         link_list = response['network'][0]['ietf-network-topology:link']
 732         updated_links = ['ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX',
 733                          'ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX']
 734         nb_updated_link = 0
 735         for link in link_list:
 736             if link['link-id'] in updated_links:
 737                 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'outOfService')
 738                 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'outOfService')
 739                 nb_updated_link += 1
 740             else:
 741                 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
 742                 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
 743         self.assertEqual(nb_updated_link, 2, "Only two xponder-output/input links should have been modified")
 744         time.sleep(1)
 745
 746     def test_45_check_update_tapi_neps(self):
 747         self.node_details["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
 748 #        self.node_details["node-id"] = "ROADM-A1+PHOTONIC_MEDIA"
 749         self.node_details["node-id"] = "3b726367-6f2d-3e3f-9033-d99b61459075"
 750         response = test_utils.transportpce_api_rpc_request(
 751             'tapi-topology', 'get-node-details', self.node_details)
 752         self.assertEqual(response['status_code'], requests.codes.ok)
 753         nep_list = response['output']['node']['owned-node-edge-point']
 754         nb_updated_neps = 0
 755         for nep in nep_list:
 756             if 'DEG2-TTP-TXRX' in nep['name'][0]['value']:
 757                 self.assertEqual(nep['operational-state'], 'DISABLED',
 758                                  "Operational State should be 'DISABLED'")
 759                 self.assertEqual(nep['administrative-state'], 'LOCKED',
 760                                  "Administrative State should be 'LOCKED'")
 761                 nb_updated_neps += 1
 762             else:
 763                 self.assertEqual(nep['operational-state'], 'ENABLED',
 764                                  "Operational State should be 'ENABLED'")
 765                 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
 766                                  "Administrative State should be 'UNLOCKED'")
 767         self.assertEqual(nb_updated_neps, 4, "4 roadm NEPS should have been modified (OTS/OMS/MC/OTSI_MC")
 768         time.sleep(1)
 769
 770     def test_46_check_update_tapi_links(self):
 771         self.tapi_topo["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
 772         response = test_utils.transportpce_api_rpc_request(
 773             'tapi-topology', 'get-topology-details', self.tapi_topo)
 774         time.sleep(2)
 775         self.assertEqual(response['status_code'], requests.codes.ok)
 776         link_list = response['output']['topology']['link']
 777         nb_updated_link = 0
 778         for link in link_list:
 779             if all(x in link['name'][0]['value'] for x in ['ROADM-A1', 'DEG2-TTP-TXRX']):
 780                 self.assertEqual(link['operational-state'], 'DISABLED')
 781                 self.assertEqual(link['administrative-state'], 'LOCKED')
 782                 nb_updated_link += 1
 783             else:
 784                 self.assertEqual(link['operational-state'], 'ENABLED')
 785                 self.assertEqual(link['administrative-state'], 'UNLOCKED')
 786         self.assertEqual(nb_updated_link, 1,
 787                          "Only one rdm-rdm link should have been modified")
 788         time.sleep(1)
 789
 790     def test_47_check_update_service_Ethernet(self):
 791         self.test_19_check_update_service_Ethernet()
 792
 793     def test_48_check_update_connectivity_service_Ethernet(self):
 794         self.test_20_check_update_connectivity_service_Ethernet()
 795
 796     def test_49_restore_status_line_port_roadma_deg(self):
 797         self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '2/0', 'L1',
 798                                                        {
 799             "port-name": "L1",
 800             "logical-connection-point": "DEG2-TTP-TXRX",
 801             "port-type": "LINE",
 802             "circuit-id": "1",
 803             "administrative-state": "inService",
 804             "port-qual": "roadm-external"
 805         }))
 806         time.sleep(2)
 807
 808     def test_50_check_update_portmapping_ok(self):
 809         self.test_36_check_update_portmapping_ok()
 810
 811     def test_51_check_update_openroadm_topo_ok(self):
 812         self.test_23_check_update_openroadm_topo_ok()
 813
 814     def test_52_check_update_tapi_neps_ok(self):
 815         self.test_38_check_update_tapi_neps_ok()
 816
 817     def test_53_check_update_tapi_links_ok(self):
 818         self.test_25_check_update_tapi_links_ok()
 819
 820     def test_54_check_update_service1_ok(self):
 821         self.test_12_get_service_Ethernet()
 822
 823     def test_55_check_update_connectivity_service_Ethernet_ok(self):
 824         self.test_13_get_connectivity_service_Ethernet()
 825
 826     def test_56_change_status_port_roadma_srg(self):
 827         self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3/0', 'C2',
 828                                                        {
 829             "port-name": "C2",
 830             "logical-connection-point": "SRG1-PP2",
 831             "port-type": "client",
 832             "circuit-id": "SRG1",
 833             "administrative-state": "outOfService",
 834             "port-qual": "roadm-external"
 835         }))
 836         time.sleep(2)
 837
 838     def test_57_check_update_portmapping(self):
 839         response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
 840         self.assertEqual(response['status_code'], requests.codes.ok)
 841         mapping_list = response['nodes'][0]['mapping']
 842         for mapping in mapping_list:
 843             if mapping['logical-connection-point'] == 'SRG1-PP2-TXRX':
 844                 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
 845                                  "Operational State should be 'OutOfService'")
 846                 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
 847                                  "Administrative State should be 'OutOfService'")
 848             else:
 849                 self.assertEqual(mapping['port-oper-state'], 'InService',
 850                                  "Operational State should be 'InService'")
 851                 self.assertEqual(mapping['port-admin-state'], 'InService',
 852                                  "Administrative State should be 'InService'")
 853         time.sleep(1)
 854
 855     def test_58_check_update_openroadm_topo(self):
 856         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
 857         self.assertEqual(response['status_code'], requests.codes.ok)
 858         node_list = response['network'][0]['node']
 859         nb_updated_tp = 0
 860         for node in node_list:
 861             self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
 862             self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
 863             tp_list = node['ietf-network-topology:termination-point']
 864             for tp in tp_list:
 865                 if node['node-id'] == 'ROADM-A1-SRG1' and tp['tp-id'] == 'SRG1-PP2-TXRX':
 866                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
 867                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
 868                     nb_updated_tp += 1
 869                 else:
 870                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
 871                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
 872         self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
 873
 874         link_list = response['network'][0]['ietf-network-topology:link']
 875         nb_updated_link = 0
 876         for link in link_list:
 877             self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
 878             self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
 879         self.assertEqual(nb_updated_link, 0, "No link should have been modified")
 880         time.sleep(1)
 881
 882     def test_59_check_update_tapi_neps(self):
 883         self.node_details["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
 884 #        self.node_details["node-id"] = uuid.UUID(bytes="ROADM-A1+PHOTONIC_MEDIA".bytes)
 885         self.node_details["node-id"] = "3b726367-6f2d-3e3f-9033-d99b61459075"
 886         response = test_utils.transportpce_api_rpc_request(
 887             'tapi-topology', 'get-node-details', self.node_details)
 888         self.assertEqual(response['status_code'], requests.codes.ok)
 889         nep_list = response['output']['node']['owned-node-edge-point']
 890         nb_updated_neps = 0
 891         for nep in nep_list:
 892             if 'SRG1-PP2-TXRX' in nep['name'][0]['value']:
 893                 self.assertEqual(nep['operational-state'], 'DISABLED',
 894                                  "Operational State should be 'DISABLED'")
 895                 self.assertEqual(nep['administrative-state'], 'LOCKED',
 896                                  "Administrative State should be 'LOCKED'")
 897                 nb_updated_neps += 1
 898             else:
 899                 self.assertEqual(nep['operational-state'], 'ENABLED',
 900                                  "Operational State should be 'ENABLED'")
 901                 self.assertEqual(nep['administrative-state'], 'UNLOCKED',
 902                                  "Administrative State should be 'UNLOCKED'")
 903         self.assertEqual(nb_updated_neps, 1, "Only 1 roadm neps OTS should have been modified for SRG2PP")
 904         time.sleep(1)
 905
 906     def test_60_check_update_tapi_links(self):
 907         self.tapi_topo["topology-id"] = test_utils.T0_FULL_MULTILAYER_TOPO_UUID
 908         response = test_utils.transportpce_api_rpc_request(
 909             'tapi-topology', 'get-topology-details', self.tapi_topo)
 910         time.sleep(2)
 911         self.assertEqual(response['status_code'], requests.codes.ok)
 912         link_list = response['output']['topology']['link']
 913         nb_updated_link = 0
 914         for link in link_list:
 915             if all(x in link['name'][0]['value'] for x in ['ROADM-A1', 'SRG1-PP2-TXRX']):
 916                 self.assertEqual(link['operational-state'], 'DISABLED')
 917                 self.assertEqual(link['administrative-state'], 'LOCKED')
 918                 nb_updated_link += 1
 919             else:
 920                 self.assertEqual(link['operational-state'], 'ENABLED')
 921                 self.assertEqual(link['administrative-state'], 'UNLOCKED')
 922         self.assertEqual(nb_updated_link, 0,
 923                          "No link should have been modified")
 924         time.sleep(1)
 925
 926     def test_61_check_update_service1_ok(self):
 927         self.test_12_get_service_Ethernet()
 928
 929     def test_62_check_update_connectivity_service_Ethernet_ok(self):
 930         self.test_13_get_connectivity_service_Ethernet()
 931
 932     def test_63_delete_connectivity_service_Ethernet(self):
 933         self.del_serv_input_data["uuid"] = str(self.uuid_services.eth)
 934         response = test_utils.transportpce_api_rpc_request(
 935             'tapi-connectivity', 'delete-connectivity-service', self.del_serv_input_data)
 936         self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content))
 937         time.sleep(self.WAITING)
 938
 939     def test_64_disconnect_xponders_from_roadm(self):
 940         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
 941         self.assertEqual(response['status_code'], requests.codes.ok)
 942         links = response['network'][0]['ietf-network-topology:link']
 943         for link in links:
 944             if link["org-openroadm-common-network:link-type"] in ('XPONDER-OUTPUT', 'XPONDER-INPUT'):
 945                 response = test_utils.del_ietf_network_link_request(
 946                     'openroadm-topology', link['link-id'], 'config')
 947                 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 948
 949     def test_65_disconnect_XPDRA(self):
 950         response = test_utils.unmount_device("XPDR-A1")
 951         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 952
 953     def test_66_disconnect_XPDRC(self):
 954         response = test_utils.unmount_device("XPDR-C1")
 955         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 956
 957     def test_67_disconnect_ROADMA(self):
 958         response = test_utils.unmount_device("ROADM-A1")
 959         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 960
 961     def test_68_disconnect_ROADMC(self):
 962         response = test_utils.unmount_device("ROADM-C1")
 963         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 964
 965     def test_69_restore_status_port_roadma_srg(self):
 966         self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3/0', 'C2',
 967                                                        {
 968             "port-name": "C2",
 969             "logical-connection-point": "SRG1-PP2",
 970             "port-type": "client",
 971             "circuit-id": "SRG1",
 972             "administrative-state": "inService",
 973             "port-qual": "roadm-external"
 974         }))
 975         time.sleep(2)
 976
 977     def test_70_clean_openroadm_topology(self):
 978         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
 979         self.assertEqual(response['status_code'], requests.codes.ok)
 980         links = response['network'][0]['ietf-network-topology:link']
 981         for link in links:
 982             if link["org-openroadm-common-network:link-type"] in ('XPONDER-OUTPUT', 'XPONDER-INPUT', 'ROADM-TO-ROADM'):
 983                 response = test_utils.del_ietf_network_link_request('openroadm-topology', link['link-id'], 'config')
 984                 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 985
 986
 987 if __name__ == "__main__":
 988     unittest.main(verbosity=2)