tests/transportpce_tests/hybrid/test01_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 unittest
  14 import time
  15 import requests
  16 # pylint: disable=wrong-import-order
  17 import sys
  18 sys.path.append('transportpce_tests/common/')
  19 # pylint: disable=wrong-import-position
  20 # pylint: disable=import-error
  21 import test_utils  # nopep8
  22
  23
  24 class TransportPCEFulltesting(unittest.TestCase):
  25
  26     processes = None
  27     cr_serv_input_data = {
  28         "sdnc-request-header": {
  29             "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
  30             "rpc-action": "service-create",
  31             "request-system-id": "appname",
  32             "notification-url": "http://localhost:8585/NotificationServer/notify"
  33         },
  34         "service-name": "service1",
  35         "common-id": "ASATT1234567",
  36         "connection-type": "service",
  37         "service-a-end": {
  38             "service-rate": "100",
  39             "node-id": "XPDRA01",
  40             "service-format": "Ethernet",
  41             "clli": "SNJSCAMCJP8",
  42             "tx-direction": [{"index": 0}],
  43             "rx-direction": [{"index": 0}],
  44             "optic-type": "gray"
  45         },
  46         "service-z-end": {
  47             "service-rate": "100",
  48             "node-id": "XPDR-C1",
  49             "service-format": "Ethernet",
  50             "clli": "SNJSCAMCJT4",
  51             "tx-direction": [{"index": 0}],
  52             "rx-direction": [{"index": 0}],
  53             "optic-type": "gray"
  54         },
  55         "due-date": "2016-11-28T00:00:01Z",
  56         "operator-contact": "pw1234"
  57     }
  58
  59     del_serv_input_data = {
  60         "sdnc-request-header": {
  61             "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
  62             "rpc-action": "service-delete",
  63             "request-system-id": "appname",
  64             "notification-url": "http://localhost:8585/NotificationServer/notify"},
  65         "service-delete-req-info": {
  66             "service-name": "TBD",
  67             "tail-retention": "no"}
  68     }
  69
  70     WAITING = 25  # nominal value is 300
  71     NODE_VERSION_121 = '1.2.1'
  72     NODE_VERSION_221 = '2.2.1'
  73     NODE_VERSION_71 = '7.1'
  74
  75     @classmethod
  76     def setUpClass(cls):
  77         cls.processes = test_utils.start_tpce()
  78         cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION_121),
  79                                                ('roadma', cls.NODE_VERSION_221),
  80                                                ('roadmc', cls.NODE_VERSION_221),
  81                                                ('xpdrc', cls.NODE_VERSION_71)])
  82
  83     @classmethod
  84     def tearDownClass(cls):
  85         # pylint: disable=not-an-iterable
  86         for process in cls.processes:
  87             test_utils.shutdown_process(process)
  88         print("all processes killed")
  89         time.sleep(3)
  90
  91     def setUp(self):  # instruction executed before each test method
  92         # pylint: disable=consider-using-f-string
  93         print("execution of {}".format(self.id().split(".")[-1]))
  94
  95     def test_01_connect_xpdrA(self):
  96         response = test_utils.mount_device("XPDRA01", ('xpdra', self.NODE_VERSION_121))
  97         self.assertEqual(response.status_code,
  98                          requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  99
 100     def test_02_connect_xpdrC(self):
 101         response = test_utils.mount_device("XPDR-C1", ('xpdrc', self.NODE_VERSION_71))
 102         self.assertEqual(response.status_code,
 103                          requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 104
 105     def test_03_connect_rdmA(self):
 106         response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION_221))
 107         self.assertEqual(response.status_code,
 108                          requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 109
 110     def test_04_connect_rdmC(self):
 111         response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION_221))
 112         self.assertEqual(response.status_code,
 113                          requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 114
 115     def test_05_connect_xpdrA_N1_to_roadmA_PP1(self):
 116         response = test_utils.transportpce_api_rpc_request(
 117             'transportpce-networkutils', 'init-xpdr-rdm-links',
 118             {'links-input': {'xpdr-node': 'XPDRA01', 'xpdr-num': '1', 'network-num': '1',
 119                              'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 120         self.assertEqual(response['status_code'], requests.codes.ok)
 121         self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
 122
 123     def test_06_connect_roadmA_PP1_to_xpdrA_N1(self):
 124         response = test_utils.transportpce_api_rpc_request(
 125             'transportpce-networkutils', 'init-rdm-xpdr-links',
 126             {'links-input': {'xpdr-node': 'XPDRA01', 'xpdr-num': '1', 'network-num': '1',
 127                              'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 128         self.assertEqual(response['status_code'], requests.codes.ok)
 129         self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
 130
 131     def test_07_connect_xpdrC_N1_to_roadmC_PP1(self):
 132         response = test_utils.transportpce_api_rpc_request(
 133             'transportpce-networkutils', 'init-xpdr-rdm-links',
 134             {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1',
 135                              'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 136         self.assertEqual(response['status_code'], requests.codes.ok)
 137         self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
 138
 139     def test_08_connect_roadmC_PP1_to_xpdrC_N1(self):
 140         response = test_utils.transportpce_api_rpc_request(
 141             'transportpce-networkutils', 'init-rdm-xpdr-links',
 142             {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1',
 143                              'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 144         self.assertEqual(response['status_code'], requests.codes.ok)
 145         self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
 146
 147     def test_09_add_omsAttributes_ROADMA_ROADMC(self):
 148         # Config ROADMA-ROADMC oms-attributes
 149         data = {"span": {
 150             "auto-spanloss": "true",
 151             "spanloss-base": 11.4,
 152             "spanloss-current": 12,
 153             "engineered-spanloss": 12.2,
 154             "link-concatenation": [{
 155                 "SRLG-Id": 0,
 156                 "fiber-type": "smf",
 157                 "SRLG-length": 100000,
 158                 "pmd": 0.5}]}}
 159         response = test_utils.add_oms_attr_request(
 160             "ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", data)
 161         self.assertEqual(response.status_code, requests.codes.created)
 162
 163     def test_10_add_omsAttributes_ROADMC_ROADMA(self):
 164         # Config ROADMC-ROADMA oms-attributes
 165         data = {"span": {
 166             "auto-spanloss": "true",
 167             "spanloss-base": 11.4,
 168             "spanloss-current": 12,
 169             "engineered-spanloss": 12.2,
 170             "link-concatenation": [{
 171                 "SRLG-Id": 0,
 172                 "fiber-type": "smf",
 173                 "SRLG-length": 100000,
 174                 "pmd": 0.5}]}}
 175         response = test_utils.add_oms_attr_request(
 176             "ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX", data)
 177         self.assertEqual(response.status_code, requests.codes.created)
 178
 179 # test service-create for Eth service from xpdr to xpdr
 180     def test_11_create_eth_service1(self):
 181         response = test_utils.transportpce_api_rpc_request(
 182             'org-openroadm-service', 'service-create',
 183             self.cr_serv_input_data)
 184         self.assertEqual(response['status_code'], requests.codes.ok)
 185         self.assertIn('PCE calculation in progress',
 186                       response['output']['configuration-response-common']['response-message'])
 187         time.sleep(self.WAITING)
 188
 189     def test_12_get_eth_service1(self):
 190         response = test_utils.get_ordm_serv_list_attr_request(
 191             "services", "service1")
 192         self.assertEqual(response['status_code'], requests.codes.ok)
 193         self.assertEqual(response['services'][0]['administrative-state'], 'inService')
 194         self.assertEqual(response['services'][0]['connection-type'], 'service')
 195         self.assertEqual(response['services'][0]['service-name'], 'service1')
 196         self.assertEqual(response['services'][0]['lifecycle-state'], 'planned')
 197         time.sleep(1)
 198
 199     def test_13_change_status_line_port_xpdra(self):
 200         self.assertTrue(test_utils.sims_update_cp_port(('xpdra', self.NODE_VERSION_121), '1%2F0%2F1-PLUG-NET', '1',
 201                                                        {
 202             "port-name": "1",
 203             "logical-connection-point": "XPDR1-NETWORK1",
 204             "port-type": "CFP2",
 205             "circuit-id": "XPDRA-NETWORK",
 206             "administrative-state": "outOfService",
 207             "port-qual": "xpdr-network"
 208         }))
 209         time.sleep(2)
 210
 211     def test_14_check_update_portmapping(self):
 212         response = test_utils.get_portmapping_node_attr("XPDRA01", None, None)
 213         self.assertEqual(response['status_code'], requests.codes.ok)
 214         mapping_list = response['nodes'][0]['mapping']
 215         for mapping in mapping_list:
 216             if mapping['logical-connection-point'] == 'XPDR1-NETWORK1':
 217                 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
 218                                  "Operational State should be 'OutOfService'")
 219                 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
 220                                  "Administrative State should be 'OutOfService'")
 221             else:
 222                 self.assertEqual(mapping['port-oper-state'], 'InService',
 223                                  "Operational State should be 'InService'")
 224                 self.assertEqual(mapping['port-admin-state'], 'InService',
 225                                  "Administrative State should be 'InService'")
 226         time.sleep(1)
 227
 228     def test_15_check_update_openroadm_topo(self):
 229         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
 230         self.assertEqual(response['status_code'], requests.codes.ok)
 231         node_list = response['network'][0]['node']
 232         nb_updated_tp = 0
 233         for node in node_list:
 234             self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
 235             self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
 236             tp_list = node['ietf-network-topology:termination-point']
 237             for tp in tp_list:
 238                 if node['node-id'] == 'XPDRA01-XPDR1' and tp['tp-id'] == 'XPDR1-NETWORK1':
 239                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
 240                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
 241                     nb_updated_tp += 1
 242                 else:
 243                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
 244                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
 245         self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
 246
 247         link_list = response['network'][0]['ietf-network-topology:link']
 248         updated_links = ['XPDRA01-XPDR1-XPDR1-NETWORK1toROADM-A1-SRG1-SRG1-PP1-TXRX',
 249                          'ROADM-A1-SRG1-SRG1-PP1-TXRXtoXPDRA01-XPDR1-XPDR1-NETWORK1']
 250         nb_updated_link = 0
 251         for link in link_list:
 252             if link['link-id'] in updated_links:
 253                 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'outOfService')
 254                 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'outOfService')
 255                 nb_updated_link += 1
 256             else:
 257                 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
 258                 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
 259         self.assertEqual(nb_updated_link, 2, "Only two xponder-output/input links should have been modified")
 260         time.sleep(1)
 261
 262     def test_16_check_update_service1(self):
 263         response = test_utils.get_ordm_serv_list_attr_request("services", "service1")
 264         self.assertEqual(response['status_code'], requests.codes.ok)
 265         self.assertEqual(response['services'][0]['operational-state'], 'outOfService')
 266         self.assertEqual(response['services'][0]['administrative-state'], 'inService')
 267         time.sleep(1)
 268
 269     def test_17_restore_status_line_port_xpdra(self):
 270         self.assertTrue(test_utils.sims_update_cp_port(('xpdra', self.NODE_VERSION_121), '1%2F0%2F1-PLUG-NET', '1',
 271                                                        {
 272             "port-name": "1",
 273             "logical-connection-point": "XPDR1-NETWORK1",
 274             "port-type": "CFP2",
 275             "circuit-id": "XPDRA-NETWORK",
 276             "administrative-state": "inService",
 277             "port-qual": "xpdr-network"
 278         }))
 279         time.sleep(2)
 280
 281     def test_18_check_update_portmapping_ok(self):
 282         response = test_utils.get_portmapping_node_attr("XPDRA01", None, None)
 283         self.assertEqual(response['status_code'], requests.codes.ok)
 284         mapping_list = response['nodes'][0]['mapping']
 285         for mapping in mapping_list:
 286             self.assertEqual(mapping['port-oper-state'], 'InService',
 287                              "Operational State should be 'InService'")
 288             self.assertEqual(mapping['port-admin-state'], 'InService',
 289                              "Administrative State should be 'InService'")
 290         time.sleep(1)
 291
 292     def test_19_check_update_openroadm_topo_ok(self):
 293         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
 294         self.assertEqual(response['status_code'], requests.codes.ok)
 295         node_list = response['network'][0]['node']
 296         for node in node_list:
 297             self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
 298             self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
 299             tp_list = node['ietf-network-topology:termination-point']
 300             for tp in tp_list:
 301                 self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
 302                 self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
 303
 304         link_list = response['network'][0]['ietf-network-topology:link']
 305         for link in link_list:
 306             self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
 307             self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
 308         time.sleep(1)
 309
 310     def test_20_check_update_service1_ok(self):
 311         self.test_12_get_eth_service1()
 312
 313     def test_21_change_status_port_roadma_srg(self):
 314         self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3%2F0', 'C1',
 315                                                        {
 316             "port-name": "C1",
 317             "logical-connection-point": "SRG1-PP1",
 318             "port-type": "client",
 319             "circuit-id": "SRG1",
 320             "administrative-state": "outOfService",
 321             "port-qual": "roadm-external"
 322         }))
 323         time.sleep(2)
 324
 325     def test_22_check_update_portmapping(self):
 326         response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
 327         self.assertEqual(response['status_code'], requests.codes.ok)
 328         mapping_list = response['nodes'][0]['mapping']
 329         for mapping in mapping_list:
 330             if mapping['logical-connection-point'] == 'SRG1-PP1-TXRX':
 331                 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
 332                                  "Operational State should be 'OutOfService'")
 333                 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
 334                                  "Administrative State should be 'OutOfService'")
 335             else:
 336                 self.assertEqual(mapping['port-oper-state'], 'InService',
 337                                  "Operational State should be 'InService'")
 338                 self.assertEqual(mapping['port-admin-state'], 'InService',
 339                                  "Administrative State should be 'InService'")
 340         time.sleep(1)
 341
 342     def test_23_check_update_openroadm_topo(self):
 343         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
 344         self.assertEqual(response['status_code'], requests.codes.ok)
 345         node_list = response['network'][0]['node']
 346         nb_updated_tp = 0
 347         for node in node_list:
 348             self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
 349             self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
 350             tp_list = node['ietf-network-topology:termination-point']
 351             for tp in tp_list:
 352                 if node['node-id'] == 'ROADM-A1-SRG1' and tp['tp-id'] == 'SRG1-PP1-TXRX':
 353                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
 354                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
 355                     nb_updated_tp += 1
 356                 else:
 357                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
 358                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
 359         self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
 360
 361         link_list = response['network'][0]['ietf-network-topology:link']
 362         updated_links = ['XPDRA01-XPDR1-XPDR1-NETWORK1toROADM-A1-SRG1-SRG1-PP1-TXRX',
 363                          'ROADM-A1-SRG1-SRG1-PP1-TXRXtoXPDRA01-XPDR1-XPDR1-NETWORK1']
 364         nb_updated_link = 0
 365         for link in link_list:
 366             if link['link-id'] in updated_links:
 367                 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'outOfService')
 368                 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'outOfService')
 369                 nb_updated_link += 1
 370             else:
 371                 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
 372                 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
 373         self.assertEqual(nb_updated_link, 2, "Only two xponder-output/input links should have been modified")
 374         time.sleep(1)
 375
 376     def test_24_restore_status_port_roadma_srg(self):
 377         self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3%2F0', 'C1',
 378                                                        {
 379             "port-name": "C1",
 380             "logical-connection-point": "SRG1-PP1",
 381             "port-type": "client",
 382             "circuit-id": "SRG1",
 383             "administrative-state": "inService",
 384             "port-qual": "roadm-external"
 385         }))
 386         time.sleep(2)
 387
 388     def test_25_check_update_portmapping_ok(self):
 389         self.test_18_check_update_portmapping_ok()
 390
 391     def test_26_check_update_openroadm_topo_ok(self):
 392         self.test_19_check_update_openroadm_topo_ok()
 393
 394     def test_27_check_update_service1_ok(self):
 395         self.test_12_get_eth_service1()
 396
 397     def test_28_change_status_line_port_roadma_deg(self):
 398         self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '2%2F0', 'L1',
 399                                                        {
 400             "port-name": "L1",
 401             "logical-connection-point": "DEG2-TTP-TXRX",
 402             "port-type": "LINE",
 403             "circuit-id": "1",
 404             "administrative-state": "outOfService",
 405             "port-qual": "roadm-external"
 406         }))
 407         time.sleep(2)
 408
 409     def test_29_check_update_portmapping(self):
 410         response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
 411         self.assertEqual(response['status_code'], requests.codes.ok)
 412         mapping_list = response['nodes'][0]['mapping']
 413         for mapping in mapping_list:
 414             if mapping['logical-connection-point'] == 'DEG2-TTP-TXRX':
 415                 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
 416                                  "Operational State should be 'OutOfService'")
 417                 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
 418                                  "Administrative State should be 'OutOfService'")
 419             else:
 420                 self.assertEqual(mapping['port-oper-state'], 'InService',
 421                                  "Operational State should be 'InService'")
 422                 self.assertEqual(mapping['port-admin-state'], 'InService',
 423                                  "Administrative State should be 'InService'")
 424         time.sleep(1)
 425
 426     def test_30_check_update_openroadm_topo(self):
 427         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
 428         self.assertEqual(response['status_code'], requests.codes.ok)
 429         node_list = response['network'][0]['node']
 430         nb_updated_tp = 0
 431         for node in node_list:
 432             self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
 433             self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
 434             tp_list = node['ietf-network-topology:termination-point']
 435             for tp in tp_list:
 436                 if node['node-id'] == 'ROADM-A1-DEG2' and tp['tp-id'] == 'DEG2-TTP-TXRX':
 437                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
 438                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
 439                     nb_updated_tp += 1
 440                 else:
 441                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
 442                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
 443         self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
 444
 445         link_list = response['network'][0]['ietf-network-topology:link']
 446         updated_links = ['ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX',
 447                          'ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX']
 448         nb_updated_link = 0
 449         for link in link_list:
 450             if link['link-id'] in updated_links:
 451                 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'outOfService')
 452                 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'outOfService')
 453                 nb_updated_link += 1
 454             else:
 455                 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
 456                 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
 457         self.assertEqual(nb_updated_link, 2, "Only two xponder-output/input links should have been modified")
 458         time.sleep(1)
 459
 460     def test_31_restore_status_line_port_roadma_srg(self):
 461         self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '2%2F0', 'L1',
 462                                                        {
 463             "port-name": "L1",
 464             "logical-connection-point": "DEG2-TTP-TXRX",
 465             "port-type": "LINE",
 466             "circuit-id": "1",
 467             "administrative-state": "inService",
 468             "port-qual": "roadm-external"
 469         }))
 470         time.sleep(2)
 471
 472     def test_32_check_update_portmapping_ok(self):
 473         self.test_18_check_update_portmapping_ok()
 474
 475     def test_33_check_update_openroadm_topo_ok(self):
 476         self.test_19_check_update_openroadm_topo_ok()
 477
 478     def test_34_check_update_service1_ok(self):
 479         self.test_12_get_eth_service1()
 480
 481     def test_35_change_status_line_port_xpdrc(self):
 482         self.assertTrue(test_utils.sims_update_cp_port(('xpdrc', self.NODE_VERSION_71), '1%2F0%2F1-PLUG-NET', '1',
 483                                                        {
 484             "port-name": "1",
 485             "port-type": "CFP2",
 486             "administrative-state": "outOfService",
 487             "port-qual": "xpdr-network"
 488         }))
 489         time.sleep(2)
 490
 491     def test_36_check_update_portmapping(self):
 492         response = test_utils.get_portmapping_node_attr("XPDR-C1", None, None)
 493         self.assertEqual(response['status_code'], requests.codes.ok)
 494         mapping_list = response['nodes'][0]['mapping']
 495         for mapping in mapping_list:
 496             if mapping['logical-connection-point'] == 'XPDR1-NETWORK1':
 497                 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
 498                                  "Operational State should be 'OutOfService'")
 499                 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
 500                                  "Administrative State should be 'OutOfService'")
 501             else:
 502                 self.assertEqual(mapping['port-oper-state'], 'InService',
 503                                  "Operational State should be 'InService'")
 504                 self.assertEqual(mapping['port-admin-state'], 'InService',
 505                                  "Administrative State should be 'InService'")
 506         time.sleep(1)
 507
 508     def test_37_check_update_openroadm_topo(self):
 509         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
 510         self.assertEqual(response['status_code'], requests.codes.ok)
 511         node_list = response['network'][0]['node']
 512         nb_updated_tp = 0
 513         for node in node_list:
 514             self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
 515             self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
 516             tp_list = node['ietf-network-topology:termination-point']
 517             for tp in tp_list:
 518                 if node['node-id'] == 'XPDR-C1-XPDR1' and tp['tp-id'] == 'XPDR1-NETWORK1':
 519                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
 520                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
 521                     nb_updated_tp += 1
 522                 else:
 523                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
 524                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
 525         self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
 526
 527         link_list = response['network'][0]['ietf-network-topology:link']
 528         updated_links = ['XPDR-C1-XPDR1-XPDR1-NETWORK1toROADM-C1-SRG1-SRG1-PP1-TXRX',
 529                          'ROADM-C1-SRG1-SRG1-PP1-TXRXtoXPDR-C1-XPDR1-XPDR1-NETWORK1']
 530         nb_updated_link = 0
 531         for link in link_list:
 532             if link['link-id'] in updated_links:
 533                 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'outOfService')
 534                 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'outOfService')
 535                 nb_updated_link += 1
 536             else:
 537                 self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
 538                 self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
 539         self.assertEqual(nb_updated_link, 2, "Only two xponder-output/input links should have been modified")
 540         time.sleep(1)
 541
 542     def test_38_restore_status_line_port_xpdrc(self):
 543         self.assertTrue(test_utils.sims_update_cp_port(('xpdrc', self.NODE_VERSION_71), '1%2F0%2F1-PLUG-NET', '1',
 544                                                        {
 545             "port-name": "1",
 546             "port-type": "CFP2",
 547             "administrative-state": "inService",
 548             "port-qual": "xpdr-network"
 549         }))
 550         time.sleep(2)
 551
 552     def test_39_check_update_portmapping_ok(self):
 553         self.test_18_check_update_portmapping_ok()
 554
 555     def test_40_check_update_openroadm_topo_ok(self):
 556         self.test_19_check_update_openroadm_topo_ok()
 557
 558     def test_41_check_update_service1_ok(self):
 559         self.test_12_get_eth_service1()
 560
 561     def test_42_change_status_port_roadma_srg(self):
 562         self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3%2F0', 'C2',
 563                                                        {
 564             "port-name": "C2",
 565             "logical-connection-point": "SRG1-PP2",
 566             "port-type": "client",
 567             "circuit-id": "SRG1",
 568             "administrative-state": "outOfService",
 569             "port-qual": "roadm-external"
 570         }))
 571         time.sleep(2)
 572
 573     def test_43_check_update_portmapping(self):
 574         response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
 575         self.assertEqual(response['status_code'], requests.codes.ok)
 576         mapping_list = response['nodes'][0]['mapping']
 577         for mapping in mapping_list:
 578             if mapping['logical-connection-point'] == 'SRG1-PP2-TXRX':
 579                 self.assertEqual(mapping['port-oper-state'], 'OutOfService',
 580                                  "Operational State should be 'OutOfService'")
 581                 self.assertEqual(mapping['port-admin-state'], 'OutOfService',
 582                                  "Administrative State should be 'OutOfService'")
 583             else:
 584                 self.assertEqual(mapping['port-oper-state'], 'InService',
 585                                  "Operational State should be 'InService'")
 586                 self.assertEqual(mapping['port-admin-state'], 'InService',
 587                                  "Administrative State should be 'InService'")
 588         time.sleep(1)
 589
 590     def test_44_check_update_openroadm_topo(self):
 591         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
 592         self.assertEqual(response['status_code'], requests.codes.ok)
 593         node_list = response['network'][0]['node']
 594         nb_updated_tp = 0
 595         for node in node_list:
 596             self.assertEqual(node['org-openroadm-common-network:operational-state'], 'inService')
 597             self.assertEqual(node['org-openroadm-common-network:administrative-state'], 'inService')
 598             tp_list = node['ietf-network-topology:termination-point']
 599             for tp in tp_list:
 600                 if node['node-id'] == 'ROADM-A1-SRG1' and tp['tp-id'] == 'SRG1-PP2-TXRX':
 601                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'outOfService')
 602                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'outOfService')
 603                     nb_updated_tp += 1
 604                 else:
 605                     self.assertEqual(tp['org-openroadm-common-network:operational-state'], 'inService')
 606                     self.assertEqual(tp['org-openroadm-common-network:administrative-state'], 'inService')
 607         self.assertEqual(nb_updated_tp, 1, "Only one termination-point should have been modified")
 608
 609         link_list = response['network'][0]['ietf-network-topology:link']
 610         nb_updated_link = 0
 611         for link in link_list:
 612             self.assertEqual(link['org-openroadm-common-network:operational-state'], 'inService')
 613             self.assertEqual(link['org-openroadm-common-network:administrative-state'], 'inService')
 614         self.assertEqual(nb_updated_link, 0, "No link should have been modified")
 615         time.sleep(1)
 616
 617     def test_45_check_update_service1_ok(self):
 618         self.test_12_get_eth_service1()
 619
 620     def test_46_delete_eth_service1(self):
 621         self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service1"
 622         response = test_utils.transportpce_api_rpc_request(
 623             'org-openroadm-service', 'service-delete',
 624             self.del_serv_input_data)
 625         self.assertEqual(response['status_code'], requests.codes.ok)
 626         self.assertIn('Renderer service delete in progress',
 627                       response['output']['configuration-response-common']['response-message'])
 628         time.sleep(self.WAITING)
 629
 630     def test_47_disconnect_xponders_from_roadm(self):
 631         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
 632         self.assertEqual(response['status_code'], requests.codes.ok)
 633         links = response['network'][0]['ietf-network-topology:link']
 634         for link in links:
 635             if link["org-openroadm-common-network:link-type"] in ('XPONDER-OUTPUT', 'XPONDER-INPUT'):
 636                 response = test_utils.del_ietf_network_link_request(
 637                     'openroadm-topology', link['link-id'], 'config')
 638                 self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 639
 640     def test_48_disconnect_XPDRA(self):
 641         response = test_utils.unmount_device("XPDRA01")
 642         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 643
 644     def test_49_disconnect_XPDRC(self):
 645         response = test_utils.unmount_device("XPDR-C1")
 646         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 647
 648     def test_50_disconnect_ROADMA(self):
 649         response = test_utils.unmount_device("ROADM-A1")
 650         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 651
 652     def test_51_disconnect_ROADMC(self):
 653         response = test_utils.unmount_device("ROADM-C1")
 654         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 655
 656
 657 if __name__ == "__main__":
 658     unittest.main(verbosity=2)