tests/transportpce_tests/2.2.1/test12_end2end.py

   1 #!/usr/bin/env python
   2 ##############################################################################
   3 # Copyright (c) 2017 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
  14 import base64
  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
  26 class TransportPCEFulltesting(unittest.TestCase):
  27
  28     processes = None
  29     cr_serv_input_data = {
  30         "sdnc-request-header": {
  31             "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
  32             "rpc-action": "service-create",
  33             "request-system-id": "appname",
  34             "notification-url": "http://localhost:8585/NotificationServer/notify"
  35         },
  36         "service-name": "service1",
  37         "common-id": "ASATT1234567",
  38         "connection-type": "service",
  39         "service-a-end": {
  40             "service-rate": "100",
  41             "node-id": "XPDR-A1",
  42             "service-format": "Ethernet",
  43             "clli": "NodeA",
  44             "tx-direction": [{
  45                 "port": {
  46                     "port-device-name": "XPDR-A1-XPDR1",
  47                     "port-type": "fixed",
  48                     "port-name": "1/0/2-PLUG-CLIENT",
  49                     "port-rack": "000000.00",
  50                     "port-shelf": "00"
  51                 },
  52                 "lgx": {
  53                     "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00",
  54                     "lgx-port-name": "LGX Back.3",
  55                     "lgx-port-rack": "000000.00",
  56                     "lgx-port-shelf": "00"
  57                 },
  58                 "index": 0
  59             }],
  60             "rx-direction": [{
  61                 "port": {
  62                     "port-device-name": "XPDR-A1-XPDR1",
  63                     "port-type": "fixed",
  64                     "port-name": "1/0/2-PLUG-CLIENT",
  65                     "port-rack": "000000.00",
  66                     "port-shelf": "00"
  67                 },
  68                 "lgx": {
  69                     "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00",
  70                     "lgx-port-name": "LGX Back.4",
  71                     "lgx-port-rack": "000000.00",
  72                     "lgx-port-shelf": "00"
  73                 },
  74                 "index": 0
  75             }],
  76             "optic-type": "gray"
  77         },
  78         "service-z-end": {
  79             "service-rate": "100",
  80             "node-id": "XPDR-C1",
  81             "service-format": "Ethernet",
  82             "clli": "NodeC",
  83                     "tx-direction": [{
  84                         "port": {
  85                             "port-device-name": "XPDR-C1-XPDR2",
  86                             "port-type": "router",
  87                             "port-name": "1/0/2-PLUG-CLIENT",
  88                             "port-rack": "000000.00",
  89                             "port-shelf": "00"
  90                         },
  91                         "lgx": {
  92                             "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00",
  93                             "lgx-port-name": "LGX Back.29",
  94                             "lgx-port-rack": "000000.00",
  95                             "lgx-port-shelf": "00"
  96                         },
  97                         "index": 0
  98                     }],
  99             "rx-direction": [{
 100                 "port": {
 101                     "port-device-name": "XPDR-C1-XPDR2",
 102                     "port-type": "router",
 103                     "port-name": "1/0/2-PLUG-CLIENT",
 104                     "port-rack": "000000.00",
 105                     "port-shelf": "00"
 106                 },
 107                 "lgx": {
 108                     "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00",
 109                     "lgx-port-name": "LGX Back.29",
 110                     "lgx-port-rack": "000000.00",
 111                     "lgx-port-shelf": "00"
 112                 },
 113                 "index": 0
 114             }],
 115             "optic-type": "gray"
 116         },
 117         "due-date": "2016-11-28T00:00:01Z",
 118         "operator-contact": "pw1234"
 119     }
 120
 121     del_serv_input_data = {
 122         "sdnc-request-header": {
 123             "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
 124             "rpc-action": "service-delete",
 125             "request-system-id": "appname",
 126             "notification-url": "http://localhost:8585/NotificationServer/notify"},
 127         "service-delete-req-info": {
 128             "service-name": "TBD",
 129             "tail-retention": "no"}
 130     }
 131
 132     WAITING = 20  # nominal value is 300
 133     NODE_VERSION = '2.2.1'
 134
 135     @classmethod
 136     def setUpClass(cls):
 137         cls.processes = test_utils.start_tpce()
 138         cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION),
 139                                                ('roadma', cls.NODE_VERSION),
 140                                                ('roadmc', cls.NODE_VERSION),
 141                                                ('xpdrc', cls.NODE_VERSION)])
 142
 143     @classmethod
 144     def tearDownClass(cls):
 145         # pylint: disable=not-an-iterable
 146         for process in cls.processes:
 147             test_utils.shutdown_process(process)
 148         print("all processes killed")
 149
 150     def setUp(self):  # instruction executed before each test method
 151         # pylint: disable=consider-using-f-string
 152         print("execution of {}".format(self.id().split(".")[-1]))
 153
 154     def test_01_connect_xpdrA(self):
 155         response = test_utils.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION))
 156         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 157
 158     def test_02_connect_xpdrC(self):
 159         response = test_utils.mount_device("XPDR-C1", ('xpdrc', self.NODE_VERSION))
 160         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 161
 162     def test_03_connect_rdmA(self):
 163         response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION))
 164         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 165
 166     def test_04_connect_rdmC(self):
 167         response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION))
 168         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 169
 170     def test_05_connect_xpdrA_N1_to_roadmA_PP1(self):
 171         response = test_utils.transportpce_api_rpc_request(
 172             'transportpce-networkutils', 'init-xpdr-rdm-links',
 173             {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1',
 174                              'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 175         self.assertEqual(response['status_code'], requests.codes.ok)
 176         self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
 177         time.sleep(2)
 178
 179     def test_06_connect_roadmA_PP1_to_xpdrA_N1(self):
 180         response = test_utils.transportpce_api_rpc_request(
 181             'transportpce-networkutils', 'init-rdm-xpdr-links',
 182             {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1',
 183                              'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 184         self.assertEqual(response['status_code'], requests.codes.ok)
 185         self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
 186         time.sleep(2)
 187
 188     def test_07_connect_xpdrC_xpdr1_N1_to_roadmC_PP1(self):
 189         response = test_utils.transportpce_api_rpc_request(
 190             'transportpce-networkutils', 'init-xpdr-rdm-links',
 191             {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1',
 192                              'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 193         self.assertEqual(response['status_code'], requests.codes.ok)
 194         self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
 195         time.sleep(2)
 196
 197     def test_08_connect_roadmC_PP1_to_xpdrC_xpdr1_N1(self):
 198         response = test_utils.transportpce_api_rpc_request(
 199             'transportpce-networkutils', 'init-rdm-xpdr-links',
 200             {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1',
 201                              'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 202         self.assertEqual(response['status_code'], requests.codes.ok)
 203         self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
 204         time.sleep(2)
 205
 206     def test_09_connect_xpdrA_N2_to_roadmA_PP2(self):
 207         response = test_utils.transportpce_api_rpc_request(
 208             'transportpce-networkutils', 'init-xpdr-rdm-links',
 209             {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '2',
 210                              'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}})
 211         self.assertEqual(response['status_code'], requests.codes.ok)
 212         self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
 213         time.sleep(2)
 214
 215     def test_10_connect_roadmA_PP2_to_xpdrA_N2(self):
 216         response = test_utils.transportpce_api_rpc_request(
 217             'transportpce-networkutils', 'init-rdm-xpdr-links',
 218             {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '2',
 219                              'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}})
 220         self.assertEqual(response['status_code'], requests.codes.ok)
 221         self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
 222         time.sleep(2)
 223
 224     def test_11_connect_xpdrC_xpdr2_N1_to_roadmC_PP2(self):
 225         response = test_utils.transportpce_api_rpc_request(
 226             'transportpce-networkutils', 'init-xpdr-rdm-links',
 227             {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '2', 'network-num': '1',
 228                              'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}})
 229         self.assertEqual(response['status_code'], requests.codes.ok)
 230         self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
 231         time.sleep(2)
 232
 233     def test_12_connect_roadmC_PP2_to_xpdrC_xpdr2_N1(self):
 234         response = test_utils.transportpce_api_rpc_request(
 235             'transportpce-networkutils', 'init-rdm-xpdr-links',
 236             {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '2', 'network-num': '1',
 237                              'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP2-TXRX'}})
 238         self.assertEqual(response['status_code'], requests.codes.ok)
 239         self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
 240         time.sleep(2)
 241
 242     def test_13_add_omsAttributes_ROADMA_ROADMC(self):
 243         # Config ROADMA-ROADMC oms-attributes
 244         data = {"span": {
 245             "auto-spanloss": "true",
 246             "spanloss-base": 11.4,
 247             "spanloss-current": 12,
 248             "engineered-spanloss": 12.2,
 249             "link-concatenation": [{
 250                 "SRLG-Id": 0,
 251                 "fiber-type": "smf",
 252                 "SRLG-length": 100000,
 253                 "pmd": 0.5}]}}
 254         response = test_utils.add_oms_attr_request("ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX",
 255                                                    data)
 256         self.assertEqual(response.status_code, requests.codes.created)
 257
 258     def test_14_add_omsAttributes_ROADMC_ROADMA(self):
 259         # Config ROADMC-ROADMA oms-attributes
 260         data = {"span": {
 261             "auto-spanloss": "true",
 262             "spanloss-base": 11.4,
 263             "spanloss-current": 12,
 264             "engineered-spanloss": 12.2,
 265             "link-concatenation": [{
 266                 "SRLG-Id": 0,
 267                 "fiber-type": "smf",
 268                 "SRLG-length": 100000,
 269                 "pmd": 0.5}]}}
 270         response = test_utils.add_oms_attr_request("ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX",
 271                                                    data)
 272         self.assertEqual(response.status_code, requests.codes.created)
 273
 274 # test service-create for Eth service from xpdr to xpdr
 275     def test_15_create_eth_service2(self):
 276         self.cr_serv_input_data["service-name"] = "service2"
 277         response = test_utils.transportpce_api_rpc_request(
 278             'org-openroadm-service', 'service-create',
 279             self.cr_serv_input_data)
 280         self.assertEqual(response['status_code'], requests.codes.ok)
 281         self.assertIn('PCE calculation in progress',
 282                       response['output']['configuration-response-common']['response-message'])
 283         time.sleep(self.WAITING)
 284
 285     def test_16_get_eth_service2(self):
 286         response = test_utils.get_ordm_serv_list_attr_request("services", "service2")
 287         self.assertEqual(response['status_code'], requests.codes.ok)
 288         self.assertEqual(response['services'][0]['administrative-state'], 'inService')
 289         self.assertEqual(response['services'][0]['service-name'], 'service2')
 290         self.assertEqual(response['services'][0]['connection-type'], 'service')
 291         self.assertEqual(response['services'][0]['lifecycle-state'], 'planned')
 292         time.sleep(1)
 293
 294     def test_17_check_xc1_ROADMA(self):
 295         response = test_utils.check_node_attribute_request(
 296             "ROADM-A1", "roadm-connections", "SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
 297         self.assertEqual(response['status_code'], requests.codes.ok)
 298         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 299         self.assertDictEqual(
 300             dict({
 301                 'connection-name': 'SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768',
 302                 'opticalControlMode': 'gainLoss',
 303                 'target-output-power': -3.0
 304             }, **response['roadm-connections'][0]), response['roadm-connections'][0])
 305         self.assertDictEqual({'src-if': 'SRG1-PP1-TXRX-nmc-761:768'}, response['roadm-connections'][0]['source'])
 306         self.assertDictEqual({'dst-if': 'DEG2-TTP-TXRX-nmc-761:768'}, response['roadm-connections'][0]['destination'])
 307         time.sleep(1)
 308
 309     def test_18_check_xc1_ROADMC(self):
 310         response = test_utils.check_node_attribute_request(
 311             "ROADM-C1", "roadm-connections", "SRG1-PP2-TXRX-DEG1-TTP-TXRX-761:768")
 312         self.assertEqual(response['status_code'], requests.codes.ok)
 313         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 314         self.assertDictEqual(
 315             dict({
 316                 'connection-name': 'SRG1-PP2-TXRX-DEG1-TTP-TXRX-761:768',
 317                 'opticalControlMode': 'gainLoss',
 318                 'target-output-power': -3.0
 319             }, **response['roadm-connections'][0]), response['roadm-connections'][0])
 320         self.assertDictEqual({'src-if': 'SRG1-PP2-TXRX-nmc-761:768'}, response['roadm-connections'][0]['source'])
 321         self.assertDictEqual({'dst-if': 'DEG1-TTP-TXRX-nmc-761:768'}, response['roadm-connections'][0]['destination'])
 322         time.sleep(1)
 323
 324     def test_19_check_topo_XPDRA(self):
 325         response = test_utils.get_ietf_network_node_request('openroadm-topology', 'XPDR-A1-XPDR1', 'config')
 326         self.assertEqual(response['status_code'], requests.codes.ok)
 327         liste_tp = response['node']['ietf-network-topology:termination-point']
 328         for ele in liste_tp:
 329             if ele['tp-id'] == 'XPDR1-NETWORK1':
 330                 self.assertEqual(
 331                     196.1,
 332                     float(ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength']['frequency']))
 333                 self.assertEqual(
 334                     40.0,
 335                     float(ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength']['width']))
 336             elif ele['tp-id'] in ('XPDR1-CLIENT1', 'XPDR1-CLIENT2'):
 337                 self.assertNotIn('org-openroadm-network-topology:xpdr-client-attributes', dict.keys(ele))
 338             elif ele['tp-id'] == 'XPDR1-NETWORK2':
 339                 self.assertIn('org-openroadm-network-topology:xpdr-network-attributes', dict.keys(ele))
 340         time.sleep(1)
 341
 342     def test_20_check_topo_ROADMA_SRG1(self):
 343         response = test_utils.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-SRG1', 'config')
 344         self.assertEqual(response['status_code'], requests.codes.ok)
 345         freq_map = base64.b64decode(
 346             response['node']['org-openroadm-network-topology:srg-attributes']['avail-freq-maps'][0]['freq-map'])
 347         freq_map_array = [int(x) for x in freq_map]
 348         self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
 349         liste_tp = response['node']['ietf-network-topology:termination-point']
 350         for ele in liste_tp:
 351             if ele['tp-id'] == 'SRG1-PP1-TXRX':
 352                 freq_map = base64.b64decode(
 353                     ele['org-openroadm-network-topology:pp-attributes']['avail-freq-maps'][0]['freq-map'])
 354                 freq_map_array = [int(x) for x in freq_map]
 355                 self.assertEqual(freq_map_array[95], 0, "Index 1 should not be available")
 356             elif ele['tp-id'] == 'SRG1-PP2-TXRX':
 357                 self.assertNotIn('avail-freq-maps', dict.keys(ele))
 358         time.sleep(1)
 359
 360     def test_21_check_topo_ROADMA_DEG2(self):
 361         response = test_utils.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-DEG2', 'config')
 362         self.assertEqual(response['status_code'], requests.codes.ok)
 363         freq_map = base64.b64decode(
 364             response['node']['org-openroadm-network-topology:degree-attributes']['avail-freq-maps'][0]['freq-map'])
 365         freq_map_array = [int(x) for x in freq_map]
 366         self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
 367         liste_tp = response['node']['ietf-network-topology:termination-point']
 368         for ele in liste_tp:
 369             if ele['tp-id'] == 'DEG2-CTP-TXRX':
 370                 freq_map = base64.b64decode(
 371                     ele['org-openroadm-network-topology:ctp-attributes']['avail-freq-maps'][0]['freq-map'])
 372                 freq_map_array = [int(x) for x in freq_map]
 373                 self.assertEqual(freq_map_array[95], 0, "Index 1 should not be available")
 374             elif ele['tp-id'] == 'DEG2-TTP-TXRX':
 375                 freq_map = base64.b64decode(
 376                     ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map'])
 377                 freq_map_array = [int(x) for x in freq_map]
 378                 self.assertEqual(freq_map_array[95], 0, "Index 1 should not be available")
 379         time.sleep(1)
 380
 381     def test_22_create_eth_service1(self):
 382         self.cr_serv_input_data["service-name"] = "service1"
 383         del self.cr_serv_input_data["service-a-end"]["tx-direction"][0]["port"]["port-device-name"]
 384         del self.cr_serv_input_data["service-a-end"]["tx-direction"][0]["port"]["port-name"]
 385         del self.cr_serv_input_data["service-a-end"]["rx-direction"][0]["port"]["port-device-name"]
 386         del self.cr_serv_input_data["service-a-end"]["rx-direction"][0]["port"]["port-name"]
 387         del self.cr_serv_input_data["service-z-end"]["tx-direction"][0]["port"]["port-device-name"]
 388         del self.cr_serv_input_data["service-z-end"]["tx-direction"][0]["port"]["port-name"]
 389         del self.cr_serv_input_data["service-z-end"]["rx-direction"][0]["port"]["port-device-name"]
 390         del self.cr_serv_input_data["service-z-end"]["rx-direction"][0]["port"]["port-name"]
 391         response = test_utils.transportpce_api_rpc_request(
 392             'org-openroadm-service', 'service-create',
 393             self.cr_serv_input_data)
 394         self.assertEqual(response['status_code'], requests.codes.ok)
 395         self.assertIn('PCE calculation in progress',
 396                       response['output']['configuration-response-common']['response-message'])
 397         time.sleep(self.WAITING)
 398
 399     def test_23_get_eth_service1(self):
 400         response = test_utils.get_ordm_serv_list_attr_request("services", "service1")
 401         self.assertEqual(response['status_code'], requests.codes.ok)
 402         self.assertEqual(response['services'][0]['administrative-state'], 'inService')
 403         self.assertEqual(response['services'][0]['service-name'], 'service1')
 404         self.assertEqual(response['services'][0]['connection-type'], 'service')
 405         self.assertEqual(response['services'][0]['lifecycle-state'], 'planned')
 406         time.sleep(1)
 407
 408     def test_24_check_xc1_ROADMA(self):
 409         response = test_utils.check_node_attribute_request(
 410             "ROADM-A1", "roadm-connections", "DEG2-TTP-TXRX-SRG1-PP2-TXRX-753:760")
 411         self.assertEqual(response['status_code'], requests.codes.ok)
 412         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 413         self.assertDictEqual(
 414             dict({
 415                 'connection-name': 'DEG2-TTP-TXRX-SRG1-PP2-TXRX-753:760',
 416                 'opticalControlMode': 'power'
 417             }, **response['roadm-connections'][0]), response['roadm-connections'][0])
 418         self.assertDictEqual({'src-if': 'DEG2-TTP-TXRX-nmc-753:760'}, response['roadm-connections'][0]['source'])
 419         self.assertDictEqual({'dst-if': 'SRG1-PP2-TXRX-nmc-753:760'}, response['roadm-connections'][0]['destination'])
 420         time.sleep(1)
 421
 422     def test_25_check_topo_XPDRA(self):
 423         response = test_utils.get_ietf_network_node_request('openroadm-topology', 'XPDR-A1-XPDR1', 'config')
 424         self.assertEqual(response['status_code'], requests.codes.ok)
 425         liste_tp = response['node']['ietf-network-topology:termination-point']
 426         for ele in liste_tp:
 427             if ele['tp-id'] == 'XPDR1-NETWORK2':
 428                 self.assertEqual(
 429                     196.05,
 430                     float(ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength']['frequency']))
 431                 self.assertEqual(
 432                     40.0,
 433                     float(ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength']['width']))
 434             elif ele['tp-id'] in ('XPDR1-CLIENT1'):
 435                 self.assertNotIn('org-openroadm-network-topology:xpdr-client-attributes', dict.keys(ele))
 436         time.sleep(1)
 437
 438     def test_26_check_topo_ROADMA_SRG1(self):
 439         response = test_utils.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-SRG1', 'config')
 440         self.assertEqual(response['status_code'], requests.codes.ok)
 441         freq_map = base64.b64decode(
 442             response['node']['org-openroadm-network-topology:srg-attributes']['avail-freq-maps'][0]['freq-map'])
 443         freq_map_array = [int(x) for x in freq_map]
 444         self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
 445         self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available")
 446         liste_tp = response['node']['ietf-network-topology:termination-point']
 447         for ele in liste_tp:
 448             if ele['tp-id'] == 'SRG1-PP1-TXRX':
 449                 freq_map = base64.b64decode(
 450                     ele['org-openroadm-network-topology:pp-attributes']['avail-freq-maps'][0]['freq-map'])
 451                 freq_map_array = [int(x) for x in freq_map]
 452                 self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
 453                 self.assertEqual(freq_map_array[94], 255, "Lambda 2 should be available")
 454             elif ele['tp-id'] == 'SRG1-PP2-TXRX':
 455                 freq_map = base64.b64decode(
 456                     ele['org-openroadm-network-topology:pp-attributes']['avail-freq-maps'][0]['freq-map'])
 457                 freq_map_array = [int(x) for x in freq_map]
 458                 self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available")
 459                 self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available")
 460             elif ele['tp-id'] == 'SRG1-PP3-TXRX':
 461                 self.assertNotIn('org-openroadm-network-topology:pp-attributes', dict.keys(ele))
 462         time.sleep(1)
 463
 464     def test_27_check_topo_ROADMA_DEG2(self):
 465         response = test_utils.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-DEG2', 'config')
 466         self.assertEqual(response['status_code'], requests.codes.ok)
 467         freq_map = base64.b64decode(
 468             response['node']['org-openroadm-network-topology:degree-attributes']['avail-freq-maps'][0]['freq-map'])
 469         freq_map_array = [int(x) for x in freq_map]
 470         self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
 471         self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available")
 472         liste_tp = response['node']['ietf-network-topology:termination-point']
 473         for ele in liste_tp:
 474             if ele['tp-id'] == 'DEG2-CTP-TXRX':
 475                 freq_map = base64.b64decode(
 476                     ele['org-openroadm-network-topology:ctp-attributes']['avail-freq-maps'][0]['freq-map'])
 477                 freq_map_array = [int(x) for x in freq_map]
 478                 self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
 479                 self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available")
 480             elif ele['tp-id'] == 'DEG2-TTP-TXRX':
 481                 freq_map = base64.b64decode(
 482                     ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map'])
 483                 freq_map_array = [int(x) for x in freq_map]
 484                 self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
 485                 self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available")
 486         time.sleep(1)
 487
 488 #     creation service test on a non-available resource
 489     def test_28_create_eth_service3(self):
 490         self.cr_serv_input_data["service-name"] = "service3"
 491         response = test_utils.transportpce_api_rpc_request(
 492             'org-openroadm-service', 'service-create',
 493             self.cr_serv_input_data)
 494         self.assertEqual(response['status_code'], requests.codes.ok)
 495         self.assertIn('PCE calculation in progress',
 496                       response['output']['configuration-response-common']['response-message'])
 497         self.assertIn('200', response['output']['configuration-response-common']['response-code'])
 498         time.sleep(self.WAITING)
 499
 500 # add a test that check the openroadm-service-list still only contains 2 elements
 501     def test_29_delete_eth_service3(self):
 502         self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service3"
 503         response = test_utils.transportpce_api_rpc_request(
 504             'org-openroadm-service', 'service-delete',
 505             self.del_serv_input_data)
 506         self.assertEqual(response['status_code'], requests.codes.ok)
 507         self.assertIn('Service \'service3\' does not exist in datastore',
 508                       response['output']['configuration-response-common']['response-message'])
 509         self.assertIn('500', response['output']['configuration-response-common']['response-code'])
 510         time.sleep(3)
 511
 512     def test_30_delete_eth_service1(self):
 513         self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service1"
 514         response = test_utils.transportpce_api_rpc_request(
 515             'org-openroadm-service', 'service-delete',
 516             self.del_serv_input_data)
 517         self.assertEqual(response['status_code'], requests.codes.ok)
 518         self.assertIn('Renderer service delete in progress',
 519                       response['output']['configuration-response-common']['response-message'])
 520         time.sleep(self.WAITING)
 521
 522     def test_31_delete_eth_service2(self):
 523         self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service2"
 524         response = test_utils.transportpce_api_rpc_request(
 525             'org-openroadm-service', 'service-delete',
 526             self.del_serv_input_data)
 527         self.assertEqual(response['status_code'], requests.codes.ok)
 528         self.assertIn('Renderer service delete in progress',
 529                       response['output']['configuration-response-common']['response-message'])
 530         time.sleep(self.WAITING)
 531
 532     def test_32_check_no_xc_ROADMA(self):
 533         response = test_utils.check_node_request("ROADM-A1")
 534         self.assertEqual(response['status_code'], requests.codes.ok)
 535         self.assertNotIn('roadm-connections',
 536                          dict.keys(response['org-openroadm-device']))
 537         time.sleep(2)
 538
 539     def test_33_check_topo_XPDRA(self):
 540         response = test_utils.get_ietf_network_node_request('openroadm-topology', 'XPDR-A1-XPDR1', 'config')
 541         self.assertEqual(response['status_code'], requests.codes.ok)
 542         liste_tp = response['node']['ietf-network-topology:termination-point']
 543         for ele in liste_tp:
 544             if ele['org-openroadm-common-network:tp-type'] == 'XPONDER-CLIENT':
 545                 self.assertNotIn('org-openroadm-network-topology:xpdr-client-attributes', dict.keys(ele))
 546             elif ele['org-openroadm-common-network:tp-type'] == 'XPONDER-NETWORK':
 547                 self.assertIn('tail-equipment-id',
 548                               dict.keys(ele['org-openroadm-network-topology:xpdr-network-attributes']))
 549                 self.assertNotIn('wavelength',
 550                                  dict.keys(ele['org-openroadm-network-topology:xpdr-network-attributes']))
 551         time.sleep(1)
 552
 553     def test_34_check_topo_ROADMA_SRG1(self):
 554         response = test_utils.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-SRG1', 'config')
 555         self.assertEqual(response['status_code'], requests.codes.ok)
 556         freq_map = base64.b64decode(
 557             response['node']['org-openroadm-network-topology:srg-attributes']['avail-freq-maps'][0]['freq-map'])
 558         freq_map_array = [int(x) for x in freq_map]
 559         self.assertEqual(freq_map_array[95], 255, "Index 1 should  be available")
 560         self.assertEqual(freq_map_array[94], 255, "Index 2 should  be available")
 561         liste_tp = response['node']['ietf-network-topology:termination-point']
 562         for ele in liste_tp:
 563             if ele['tp-id'] in ('SRG1-PP1-TXRX', 'SRG1-PP2-TXRX'):
 564                 freq_map = base64.b64decode(
 565                     ele['org-openroadm-network-topology:pp-attributes']['avail-freq-maps'][0]['freq-map'])
 566                 freq_map_array = [int(x) for x in freq_map]
 567                 self.assertEqual(freq_map_array[95], 255, "Index 1 should  be available")
 568                 self.assertEqual(freq_map_array[94], 255, "Index 2 should  be available")
 569             elif ele['tp-id'] == 'SRG1-CP-TXRX':
 570                 freq_map = base64.b64decode(
 571                     ele['org-openroadm-network-topology:cp-attributes']['avail-freq-maps'][0]['freq-map'])
 572                 freq_map_array = [int(x) for x in freq_map]
 573                 self.assertEqual(freq_map_array[95], 255, "Index 1 should  be available")
 574                 self.assertEqual(freq_map_array[94], 255, "Index 2 should  be available")
 575             else:
 576                 self.assertNotIn('org-openroadm-network-topology:pp-attributes', dict.keys(ele))
 577         time.sleep(1)
 578
 579     def test_35_check_topo_ROADMA_DEG2(self):
 580         response = test_utils.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-DEG2', 'config')
 581         self.assertEqual(response['status_code'], requests.codes.ok)
 582         freq_map = base64.b64decode(
 583             response['node']['org-openroadm-network-topology:degree-attributes']['avail-freq-maps'][0]['freq-map'])
 584         freq_map_array = [int(x) for x in freq_map]
 585         self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available")
 586         self.assertEqual(freq_map_array[94], 255, "Lambda 2 should be available")
 587         liste_tp = response['node']['ietf-network-topology:termination-point']
 588         for ele in liste_tp:
 589             if ele['tp-id'] == 'DEG2-CTP-TXRX':
 590                 freq_map = base64.b64decode(
 591                     ele['org-openroadm-network-topology:ctp-attributes']['avail-freq-maps'][0]['freq-map'])
 592                 freq_map_array = [int(x) for x in freq_map]
 593                 self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available")
 594                 self.assertEqual(freq_map_array[94], 255, "Lambda 2 should be available")
 595             elif ele['tp-id'] == 'DEG2-TTP-TXRX':
 596                 freq_map = base64.b64decode(
 597                     ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map'])
 598                 freq_map_array = [int(x) for x in freq_map]
 599                 self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available")
 600                 self.assertEqual(freq_map_array[94], 255, "Lambda 2 should be available")
 601         time.sleep(1)
 602
 603 # test service-create for Optical Channel (OC) service from srg-pp to srg-pp
 604     def test_36_create_oc_service1(self):
 605         self.cr_serv_input_data["service-name"] = "service1"
 606         self.cr_serv_input_data["connection-type"] = "roadm-line"
 607         self.cr_serv_input_data["service-a-end"]["node-id"] = "ROADM-A1"
 608         self.cr_serv_input_data["service-a-end"]["service-format"] = "OC"
 609         self.cr_serv_input_data["service-z-end"]["node-id"] = "ROADM-C1"
 610         self.cr_serv_input_data["service-z-end"]["service-format"] = "OC"
 611         response = test_utils.transportpce_api_rpc_request(
 612             'org-openroadm-service', 'service-create',
 613             self.cr_serv_input_data)
 614         self.assertEqual(response['status_code'], requests.codes.ok)
 615         self.assertIn('PCE calculation in progress',
 616                       response['output']['configuration-response-common']['response-message'])
 617         time.sleep(self.WAITING)
 618
 619     def test_37_get_oc_service1(self):
 620         response = test_utils.get_ordm_serv_list_attr_request("services", "service1")
 621         self.assertEqual(response['status_code'], requests.codes.ok)
 622         self.assertEqual(response['services'][0]['administrative-state'], 'inService')
 623         self.assertEqual(response['services'][0]['service-name'], 'service1')
 624         self.assertEqual(response['services'][0]['connection-type'], 'roadm-line')
 625         self.assertEqual(response['services'][0]['lifecycle-state'], 'planned')
 626         time.sleep(1)
 627
 628     def test_38_check_xc1_ROADMA(self):
 629         response = test_utils.check_node_attribute_request(
 630             "ROADM-A1", "roadm-connections", "SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
 631         self.assertEqual(response['status_code'], requests.codes.ok)
 632         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 633         self.assertDictEqual(
 634             dict({
 635                 'connection-name': 'SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768',
 636                 'opticalControlMode': 'gainLoss',
 637                 'target-output-power': -3.0
 638             }, **response['roadm-connections'][0]), response['roadm-connections'][0])
 639         self.assertDictEqual({'src-if': 'SRG1-PP1-TXRX-nmc-761:768'}, response['roadm-connections'][0]['source'])
 640         self.assertDictEqual({'dst-if': 'DEG2-TTP-TXRX-nmc-761:768'}, response['roadm-connections'][0]['destination'])
 641         time.sleep(1)
 642
 643     def test_39_check_xc1_ROADMC(self):
 644         response = test_utils.check_node_attribute_request(
 645             "ROADM-C1", "roadm-connections", "SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768")
 646         self.assertEqual(response['status_code'], requests.codes.ok)
 647         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 648         self.assertDictEqual(
 649             dict({
 650                 'connection-name': 'SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768',
 651                 'opticalControlMode': 'gainLoss',
 652                 'target-output-power': -3.0
 653             }, **response['roadm-connections'][0]), response['roadm-connections'][0])
 654         self.assertDictEqual({'src-if': 'SRG1-PP1-TXRX-nmc-761:768'}, response['roadm-connections'][0]['source'])
 655         self.assertDictEqual({'dst-if': 'DEG1-TTP-TXRX-nmc-761:768'}, response['roadm-connections'][0]['destination'])
 656         time.sleep(1)
 657
 658     def test_40_create_oc_service2(self):
 659         self.cr_serv_input_data["service-name"] = "service2"
 660         self.cr_serv_input_data["connection-type"] = "roadm-line"
 661         self.cr_serv_input_data["service-a-end"]["node-id"] = "ROADM-A1"
 662         self.cr_serv_input_data["service-a-end"]["service-format"] = "OC"
 663         self.cr_serv_input_data["service-z-end"]["node-id"] = "ROADM-C1"
 664         self.cr_serv_input_data["service-z-end"]["service-format"] = "OC"
 665         response = test_utils.transportpce_api_rpc_request(
 666             'org-openroadm-service', 'service-create',
 667             self.cr_serv_input_data)
 668         self.assertEqual(response['status_code'], requests.codes.ok)
 669         self.assertIn('PCE calculation in progress',
 670                       response['output']['configuration-response-common']['response-message'])
 671         time.sleep(self.WAITING)
 672
 673     def test_41_get_oc_service2(self):
 674         response = test_utils.get_ordm_serv_list_attr_request("services", "service2")
 675         self.assertEqual(response['status_code'], requests.codes.ok)
 676         self.assertEqual(response['services'][0]['administrative-state'], 'inService')
 677         self.assertEqual(response['services'][0]['service-name'], 'service2')
 678         self.assertEqual(response['services'][0]['connection-type'], 'roadm-line')
 679         self.assertEqual(response['services'][0]['lifecycle-state'], 'planned')
 680         time.sleep(1)
 681
 682     def test_42_check_xc2_ROADMA(self):
 683         response = test_utils.check_node_attribute_request(
 684             "ROADM-A1", "roadm-connections", "SRG1-PP2-TXRX-DEG2-TTP-TXRX-753:760")
 685         self.assertEqual(response['status_code'], requests.codes.ok)
 686         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 687         self.assertDictEqual(
 688             dict({
 689                 'connection-name': 'SRG1-PP2-TXRX-DEG2-TTP-TXRX-753:760',
 690                 'opticalControlMode': 'gainLoss',
 691                 'target-output-power': -3.0
 692             }, **response['roadm-connections'][0]), response['roadm-connections'][0])
 693         self.assertDictEqual({'src-if': 'SRG1-PP2-TXRX-nmc-753:760'}, response['roadm-connections'][0]['source'])
 694         self.assertDictEqual({'dst-if': 'DEG2-TTP-TXRX-nmc-753:760'}, response['roadm-connections'][0]['destination'])
 695         time.sleep(1)
 696
 697     def test_43_check_topo_ROADMA(self):
 698         response = test_utils.get_ietf_network_node_request('openroadm-topology', 'ROADM-A1-SRG1', 'config')
 699         self.assertEqual(response['status_code'], requests.codes.ok)
 700         freq_map = base64.b64decode(
 701             response['node']['org-openroadm-network-topology:srg-attributes']['avail-freq-maps'][0]['freq-map'])
 702         freq_map_array = [int(x) for x in freq_map]
 703         self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
 704         self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available")
 705         liste_tp = response['node']['ietf-network-topology:termination-point']
 706         for ele in liste_tp:
 707             if ele['tp-id'] == 'SRG1-PP1-TXRX':
 708                 freq_map = base64.b64decode(
 709                     ele['org-openroadm-network-topology:pp-attributes']['avail-freq-maps'][0]['freq-map'])
 710                 freq_map_array = [int(x) for x in freq_map]
 711                 self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
 712                 self.assertEqual(freq_map_array[94], 255, "Lambda 2 should be available")
 713             elif ele['tp-id'] == 'SRG1-PP2-TXRX':
 714                 freq_map = base64.b64decode(
 715                     ele['org-openroadm-network-topology:pp-attributes']['avail-freq-maps'][0]['freq-map'])
 716                 freq_map_array = [int(x) for x in freq_map]
 717                 self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available")
 718                 self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available")
 719             elif ele['tp-id'] == 'SRG1-PP3-TXRX':
 720                 self.assertNotIn('org-openroadm-network-topology:pp-attributes', dict.keys(ele))
 721         time.sleep(1)
 722         self.test_27_check_topo_ROADMA_DEG2()
 723         time.sleep(1)
 724
 725     def test_44_delete_oc_service1(self):
 726         self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service1"
 727         response = test_utils.transportpce_api_rpc_request(
 728             'org-openroadm-service', 'service-delete',
 729             self.del_serv_input_data)
 730         self.assertEqual(response['status_code'], requests.codes.ok)
 731         self.assertIn('Renderer service delete in progress',
 732                       response['output']['configuration-response-common']['response-message'])
 733         time.sleep(self.WAITING)
 734
 735     def test_45_delete_oc_service2(self):
 736         self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service2"
 737         response = test_utils.transportpce_api_rpc_request(
 738             'org-openroadm-service', 'service-delete',
 739             self.del_serv_input_data)
 740         self.assertEqual(response['status_code'], requests.codes.ok)
 741         self.assertIn('Renderer service delete in progress',
 742                       response['output']['configuration-response-common']['response-message'])
 743         time.sleep(self.WAITING)
 744
 745     def test_46_get_no_oc_services(self):
 746         response = test_utils.get_ordm_serv_list_request()
 747         self.assertEqual(response['status_code'], requests.codes.conflict)
 748         self.assertIn(response['service-list'], (
 749             {
 750                 "error-type": "protocol",
 751                 "error-tag": "data-missing",
 752                 "error-message":
 753                     "Request could not be completed because the relevant data "
 754                     "model content does not exist"
 755             }, {
 756                 "error-type": "application",
 757                 "error-tag": "data-missing",
 758                 "error-message":
 759                     "Request could not be completed because the relevant data "
 760                     "model content does not exist"
 761             }))
 762         time.sleep(1)
 763
 764     def test_47_get_no_xc_ROADMA(self):
 765         response = test_utils.check_node_request("ROADM-A1")
 766         self.assertEqual(response['status_code'], requests.codes.ok)
 767         self.assertNotIn('roadm-connections', dict.keys(response['org-openroadm-device']))
 768         time.sleep(1)
 769
 770     def test_48_check_topo_ROADMA(self):
 771         self.test_34_check_topo_ROADMA_SRG1()
 772         self.test_35_check_topo_ROADMA_DEG2()
 773
 774     def test_49_loop_create_oc_service(self):
 775         for i in range(1, 3):
 776             # pylint: disable=consider-using-f-string
 777             print("iteration number {}".format(i))
 778             print("oc service creation")
 779             self.test_36_create_oc_service1()
 780             print("check xc in ROADM-A1")
 781             self.test_38_check_xc1_ROADMA()
 782             print("check xc in ROADM-C1")
 783             self.test_39_check_xc1_ROADMC()
 784             print("oc service deletion\n")
 785             self.test_44_delete_oc_service1()
 786
 787     def test_50_loop_create_eth_service(self):
 788         response = test_utils.get_ordm_serv_list_attr_request("services", "service1")
 789         if response['status_code'] != 404:
 790             self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service1"
 791             response = test_utils.transportpce_api_rpc_request(
 792                 'org-openroadm-service', 'service-delete',
 793                 self.del_serv_input_data)
 794             time.sleep(5)
 795         self.cr_serv_input_data["connection-type"] = "service"
 796         self.cr_serv_input_data["service-a-end"]["node-id"] = "XPDR-A1"
 797         self.cr_serv_input_data["service-a-end"]["service-format"] = "Ethernet"
 798         self.cr_serv_input_data["service-z-end"]["node-id"] = "XPDR-C1"
 799         self.cr_serv_input_data["service-z-end"]["service-format"] = "Ethernet"
 800         for i in range(1, 3):
 801             # pylint: disable=consider-using-f-string
 802             print("iteration number {}".format(i))
 803             print("eth service creation")
 804             self.test_15_create_eth_service2()
 805             print("check xc in ROADM-A1")
 806             self.test_17_check_xc1_ROADMA()
 807             print("eth service deletion\n")
 808             self.test_31_delete_eth_service2()
 809
 810     def test_51_disconnect_XPDRA(self):
 811         response = test_utils.unmount_device("XPDR-A1")
 812         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 813
 814     def test_52_disconnect_XPDRC(self):
 815         response = test_utils.unmount_device("XPDR-C1")
 816         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 817
 818     def test_53_disconnect_ROADMA(self):
 819         response = test_utils.unmount_device("ROADM-A1")
 820         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 821
 822     def test_54_disconnect_ROADMC(self):
 823         response = test_utils.unmount_device("ROADM-C1")
 824         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 825
 826
 827 if __name__ == "__main__":
 828     unittest.main(verbosity=2)