tests/transportpce_tests/with_docker/test02_nbinotifications.py

   1 #!/usr/bin/env python
   2 ##############################################################################
   3 # Copyright (c) 2020 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 os
  15 import json
  16 # pylint: disable=wrong-import-order
  17 import sys
  18 import unittest
  19 import time
  20 import requests
  21 sys.path.append('transportpce_tests/common/')
  22 # pylint: disable=wrong-import-position
  23 # pylint: disable=import-error
  24 import test_utils  # nopep8
  25
  26
  27 class TransportNbiNotificationstesting(unittest.TestCase):
  28     processes = []
  29     cr_serv_sample_data = {"input": {
  30         "sdnc-request-header": {
  31             "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
  32             "rpc-action": "service-create",
  33             "request-system-id": "appname",
  34             "notification-url": "http://localhost:8585/NotificationServer/notify"
  35         },
  36         "service-name": "service1",
  37         "common-id": "ASATT1234567",
  38         "connection-type": "service",
  39         "service-a-end": {
  40             "service-rate": "100",
  41             "node-id": "XPDR-A1",
  42             "service-format": "Ethernet",
  43             "clli": "SNJSCAMCJP8",
  44             "tx-direction": {
  45                 "port": {
  46                     "port-device-name": "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": "LGX Panel_SNJSCAMCJP8_000000.00_00",
  54                     "lgx-port-name": "LGX Back.3",
  55                     "lgx-port-rack": "000000.00",
  56                     "lgx-port-shelf": "00"
  57                 }
  58             },
  59             "rx-direction": {
  60                 "port": {
  61                     "port-device-name": "ROUTER_SNJSCAMCJP8_000000.00_00",
  62                     "port-type": "router",
  63                     "port-name": "Gigabit Ethernet_Rx.ge-5/0/0.0",
  64                     "port-rack": "000000.00",
  65                     "port-shelf": "00"
  66                 },
  67                 "lgx": {
  68                     "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00",
  69                     "lgx-port-name": "LGX Back.4",
  70                     "lgx-port-rack": "000000.00",
  71                     "lgx-port-shelf": "00"
  72                 }
  73             },
  74             "optic-type": "gray"
  75         },
  76         "service-z-end": {
  77             "service-rate": "100",
  78             "node-id": "XPDR-C1",
  79             "service-format": "Ethernet",
  80             "clli": "SNJSCAMCJT4",
  81             "tx-direction": {
  82                 "port": {
  83                     "port-device-name": "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": "LGX Panel_SNJSCAMCJT4_000000.00_00",
  91                     "lgx-port-name": "LGX Back.29",
  92                     "lgx-port-rack": "000000.00",
  93                     "lgx-port-shelf": "00"
  94                 }
  95             },
  96             "rx-direction": {
  97                 "port": {
  98                     "port-device-name": "ROUTER_SNJSCAMCJT4_000000.00_00",
  99                     "port-type": "router",
 100                     "port-name": "Gigabit Ethernet_Rx.ge-1/0/0.0",
 101                     "port-rack": "000000.00",
 102                     "port-shelf": "00"
 103                 },
 104                 "lgx": {
 105                     "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00",
 106                     "lgx-port-name": "LGX Back.30",
 107                     "lgx-port-rack": "000000.00",
 108                     "lgx-port-shelf": "00"
 109                 }
 110             },
 111             "optic-type": "gray"
 112         },
 113         "due-date": "2016-11-28T00:00:01Z",
 114         "operator-contact": "pw1234"
 115     }
 116     }
 117
 118     WAITING = 20  # nominal value is 300
 119     NODE_VERSION = '2.2.1'
 120
 121     @classmethod
 122     def setUpClass(cls):
 123         # pylint: disable=unsubscriptable-object
 124         # TODO: for lighty manage the activation of NBI notification feature
 125         cls.init_failed = False
 126         cls.processes = test_utils.start_tpce()
 127         # NBI notification feature is not installed by default in Karaf
 128         if "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
 129             print("installing NBI notification feature...")
 130             result = test_utils.install_karaf_feature("odl-transportpce-nbinotifications")
 131             if result.returncode != 0:
 132                 cls.init_failed = True
 133             print("Restarting OpenDaylight...")
 134             test_utils.shutdown_process(cls.processes[0])
 135             cls.processes[0] = test_utils.start_karaf()
 136             test_utils.process_list[0] = cls.processes[0]
 137             cls.init_failed = not test_utils.wait_until_log_contains(
 138                 test_utils.KARAF_LOG, test_utils.KARAF_OK_START_MSG, time_to_wait=60)
 139         if cls.init_failed:
 140             print("NBI notification installation feature failed...")
 141             test_utils.shutdown_process(cls.processes[0])
 142             sys.exit(2)
 143         cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION),
 144                                                ('roadma', cls.NODE_VERSION),
 145                                                ('roadmc', cls.NODE_VERSION),
 146                                                ('xpdrc', cls.NODE_VERSION)])
 147
 148     @classmethod
 149     def tearDownClass(cls):
 150         # pylint: disable=not-an-iterable
 151         for process in cls.processes:
 152             test_utils.shutdown_process(process)
 153         print("all processes killed")
 154
 155     def setUp(self):  # instruction executed before each test method
 156         # pylint: disable=consider-using-f-string
 157         print("execution of {}".format(self.id().split(".")[-1]))
 158
 159     def test_01_connect_xpdrA(self):
 160         response = test_utils.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION))
 161         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 162
 163     def test_02_connect_xpdrC(self):
 164         response = test_utils.mount_device("XPDR-C1", ('xpdrc', self.NODE_VERSION))
 165         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 166
 167     def test_03_connect_rdmA(self):
 168         response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION))
 169         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 170
 171     def test_04_connect_rdmC(self):
 172         response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION))
 173         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 174
 175     def test_05_connect_xprdA_N1_to_roadmA_PP1(self):
 176         response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "1",
 177                                                           "ROADM-A1", "1", "SRG1-PP1-TXRX")
 178         self.assertEqual(response.status_code, requests.codes.ok)
 179         res = response.json()
 180         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
 181         time.sleep(2)
 182
 183     def test_06_connect_roadmA_PP1_to_xpdrA_N1(self):
 184         response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "1",
 185                                                           "ROADM-A1", "1", "SRG1-PP1-TXRX")
 186         self.assertEqual(response.status_code, requests.codes.ok)
 187         res = response.json()
 188         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
 189         time.sleep(2)
 190
 191     def test_07_connect_xprdC_N1_to_roadmC_PP1(self):
 192         response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "1",
 193                                                           "ROADM-C1", "1", "SRG1-PP1-TXRX")
 194         self.assertEqual(response.status_code, requests.codes.ok)
 195         res = response.json()
 196         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
 197         time.sleep(2)
 198
 199     def test_08_connect_roadmC_PP1_to_xpdrC_N1(self):
 200         response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "1",
 201                                                           "ROADM-C1", "1", "SRG1-PP1-TXRX")
 202         self.assertEqual(response.status_code, requests.codes.ok)
 203         res = response.json()
 204         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
 205         time.sleep(2)
 206
 207     def test_09_get_notifications_service1(self):
 208         data = {
 209             "input": {
 210                 "connection-type": "service",
 211                 "id-consumer": "consumer",
 212                 "group-id": "transportpceTest"
 213             }
 214         }
 215         response = test_utils.get_notifications_process_service_request(data)
 216         self.assertEqual(response.status_code, requests.codes.no_content)
 217         time.sleep(2)
 218
 219     def test_10_create_eth_service1(self):
 220         self.cr_serv_sample_data["input"]["service-name"] = "service1"
 221         response = test_utils.service_create_request(self.cr_serv_sample_data)
 222         self.assertEqual(response.status_code, requests.codes.ok)
 223         res = response.json()
 224         self.assertIn('PCE calculation in progress',
 225                       res['output']['configuration-response-common']['response-message'])
 226         time.sleep(self.WAITING)
 227
 228     def test_11_get_notifications_service1(self):
 229         data = {
 230             "input": {
 231                 "connection-type": "service",
 232                 "id-consumer": "consumer",
 233                 "group-id": "transportpceTest"
 234             }
 235         }
 236         response = test_utils.get_notifications_process_service_request(data)
 237         self.assertEqual(response.status_code, requests.codes.ok)
 238         res = response.json()
 239         self.assertEqual(res['output']['notifications-process-service'][-1]['service-name'], 'service1')
 240         self.assertEqual(res['output']['notifications-process-service'][-1]['connection-type'], 'service')
 241         self.assertEqual(res['output']['notifications-process-service'][-1]['message'],
 242                          'ServiceCreate request failed ...')
 243         self.assertEqual(res['output']['notifications-process-service'][-1]['response-failed'],
 244                          'PCE path computation failed !')
 245         time.sleep(2)
 246
 247     def test_12_add_omsAttributes_ROADMA_ROADMC(self):
 248         # Config ROADMA-ROADMC oms-attributes
 249         data = {"span": {
 250             "auto-spanloss": "true",
 251             "spanloss-base": 11.4,
 252             "spanloss-current": 12,
 253             "engineered-spanloss": 12.2,
 254             "link-concatenation": [{
 255                 "SRLG-Id": 0,
 256                 "fiber-type": "smf",
 257                 "SRLG-length": 100000,
 258                 "pmd": 0.5}]}}
 259         response = test_utils.add_oms_attr_request("ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", data)
 260         self.assertEqual(response.status_code, requests.codes.created)
 261
 262     def test_13_add_omsAttributes_ROADMC_ROADMA(self):
 263         # Config ROADMC-ROADMA oms-attributes
 264         data = {"span": {
 265             "auto-spanloss": "true",
 266             "spanloss-base": 11.4,
 267             "spanloss-current": 12,
 268             "engineered-spanloss": 12.2,
 269             "link-concatenation": [{
 270                 "SRLG-Id": 0,
 271                 "fiber-type": "smf",
 272                 "SRLG-length": 100000,
 273                 "pmd": 0.5}]}}
 274         response = test_utils.add_oms_attr_request("ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX", data)
 275         self.assertEqual(response.status_code, requests.codes.created)
 276
 277     # test service-create for Eth service from xpdr to xpdr
 278     def test_14_create_eth_service1(self):
 279         self.cr_serv_sample_data["input"]["service-name"] = "service1"
 280         response = test_utils.service_create_request(self.cr_serv_sample_data)
 281         self.assertEqual(response.status_code, requests.codes.ok)
 282         res = response.json()
 283         self.assertIn('PCE calculation in progress',
 284                       res['output']['configuration-response-common']['response-message'])
 285         time.sleep(self.WAITING)
 286
 287     def test_15_get_eth_service1(self):
 288         response = test_utils.get_service_list_request("services/service1")
 289         self.assertEqual(response.status_code, requests.codes.ok)
 290         res = response.json()
 291         self.assertEqual(
 292             res['services'][0]['administrative-state'], 'inService')
 293         self.assertEqual(
 294             res['services'][0]['service-name'], 'service1')
 295         self.assertEqual(
 296             res['services'][0]['connection-type'], 'service')
 297         self.assertEqual(
 298             res['services'][0]['lifecycle-state'], 'planned')
 299         time.sleep(2)
 300
 301     def test_16_get_notifications_service1(self):
 302         data = {
 303             "input": {
 304                 "connection-type": "service",
 305                 "id-consumer": "consumer",
 306                 "group-id": "transportpceTest"
 307             }
 308         }
 309         response = test_utils.get_notifications_process_service_request(data)
 310         self.assertEqual(response.status_code, requests.codes.ok)
 311         res = response.json()
 312         self.assertEqual(res['output']['notifications-process-service'][-1]['service-name'], 'service1')
 313         self.assertEqual(res['output']['notifications-process-service'][-1]['connection-type'], 'service')
 314         self.assertEqual(res['output']['notifications-process-service'][-1]['message'], 'Service implemented !')
 315         time.sleep(2)
 316
 317     def test_17_get_notifications_alarm_service1(self):
 318         data = {
 319             "input": {
 320                 "connection-type": "service",
 321                 "id-consumer": "consumer",
 322                 "group-id": "transportpceTest"
 323             }
 324         }
 325         response = test_utils.get_notifications_alarm_service_request(data)
 326         self.assertEqual(response.status_code, requests.codes.ok)
 327         res = response.json()
 328         self.assertEqual(res['output']['notifications-alarm-service'][-1]['service-name'], 'service1')
 329         self.assertEqual(res['output']['notifications-alarm-service'][-1]['connection-type'], 'service')
 330         self.assertEqual(res['output']['notifications-alarm-service'][-1]['operational-state'], 'inService')
 331         self.assertEqual(res['output']['notifications-alarm-service'][-1]['message'], 'The service is now inService')
 332         time.sleep(2)
 333
 334     def test_18_change_status_port_roadma_srg(self):
 335         url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
 336         body = {"ports": [{
 337             "port-name": "C1",
 338             "logical-connection-point": "SRG1-PP1",
 339             "port-type": "client",
 340             "circuit-id": "SRG1",
 341             "administrative-state": "outOfService",
 342             "port-qual": "roadm-external"}]}
 343         response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
 344                                     data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
 345                                     auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 346         self.assertEqual(response.status_code, requests.codes.ok)
 347         time.sleep(2)
 348
 349     def test_19_get_notifications_alarm_service1(self):
 350         data = {
 351             "input": {
 352                 "connection-type": "service",
 353                 "id-consumer": "consumer",
 354                 "group-id": "transportpceTest"
 355             }
 356         }
 357         response = test_utils.get_notifications_alarm_service_request(data)
 358         self.assertEqual(response.status_code, requests.codes.ok)
 359         res = response.json()
 360         self.assertEqual(res['output']['notifications-alarm-service'][-1]['service-name'], 'service1')
 361         self.assertEqual(res['output']['notifications-alarm-service'][-1]['connection-type'], 'service')
 362         self.assertEqual(res['output']['notifications-alarm-service'][-1]['operational-state'], 'outOfService')
 363         self.assertEqual(res['output']['notifications-alarm-service'][-1]['message'], 'The service is now outOfService')
 364         time.sleep(2)
 365
 366     def test_20_restore_status_port_roadma_srg(self):
 367         url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
 368         body = {"ports": [{
 369             "port-name": "C1",
 370             "logical-connection-point": "SRG1-PP1",
 371             "port-type": "client",
 372             "circuit-id": "SRG1",
 373             "administrative-state": "inService",
 374             "port-qual": "roadm-external"}]}
 375         response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
 376                                     data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
 377                                     auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 378         self.assertEqual(response.status_code, requests.codes.ok)
 379         time.sleep(2)
 380
 381     def test_21_get_notifications_alarm_service1(self):
 382         self.test_17_get_notifications_alarm_service1()
 383
 384     def test_22_delete_eth_service1(self):
 385         response = test_utils.service_delete_request("service1")
 386         self.assertEqual(response.status_code, requests.codes.ok)
 387         res = response.json()
 388         self.assertIn('Renderer service delete in progress',
 389                       res['output']['configuration-response-common']['response-message'])
 390         time.sleep(20)
 391
 392     def test_23_get_notifications_service1(self):
 393         data = {
 394             "input": {
 395                 "connection-type": "service",
 396                 "id-consumer": "consumer",
 397                 "group-id": "transportpceTest"
 398             }
 399         }
 400         response = test_utils.get_notifications_process_service_request(data)
 401         self.assertEqual(response.status_code, requests.codes.ok)
 402         res = response.json()
 403         self.assertEqual(res['output']['notifications-process-service'][-1]['service-name'], 'service1')
 404         self.assertEqual(res['output']['notifications-process-service'][-1]['connection-type'], 'service')
 405         self.assertEqual(res['output']['notifications-process-service'][-1]['message'], 'Service deleted !')
 406         time.sleep(2)
 407
 408     def test_24_disconnect_XPDRA(self):
 409         response = test_utils.unmount_device("XPDR-A1")
 410         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 411
 412     def test_25_disconnect_XPDRC(self):
 413         response = test_utils.unmount_device("XPDR-C1")
 414         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 415
 416     def test_26_disconnect_ROADMA(self):
 417         response = test_utils.unmount_device("ROADM-A1")
 418         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 419
 420     def test_27_disconnect_ROADMC(self):
 421         response = test_utils.unmount_device("ROADM-C1")
 422         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 423
 424
 425 if __name__ == "__main__":
 426     unittest.main(verbosity=2)