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