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