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