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