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