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 = None
  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         # TODO: for lighty manage the activation of NBI notification feature
 124         cls.init_failed = False
 125         cls.processes = test_utils.start_tpce()
 126         # NBI notification feature is not installed by default in Karaf
 127         if "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
 128             print("installing NBI notification feature...")
 129             result = test_utils.install_karaf_feature("odl-transportpce-nbinotifications")
 130             if result.returncode != 0:
 131                 cls.init_failed = True
 132             print("Restarting OpenDaylight...")
 133             test_utils.shutdown_process(cls.processes[0])
 134             cls.processes[0] = test_utils.start_karaf()
 135             test_utils.process_list[0] = cls.processes[0]
 136             cls.init_failed = not test_utils.wait_until_log_contains(
 137                 test_utils.KARAF_LOG, test_utils.KARAF_OK_START_MSG, time_to_wait=60)
 138         if cls.init_failed:
 139             print("NBI notification installation feature failed...")
 140             test_utils.shutdown_process(cls.processes[0])
 141             sys.exit(2)
 142         cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION),
 143                                                ('roadma', cls.NODE_VERSION),
 144                                                ('roadmc', cls.NODE_VERSION),
 145                                                ('xpdrc', cls.NODE_VERSION)])
 146
 147     @classmethod
 148     def tearDownClass(cls):
 149         # pylint: disable=not-an-iterable
 150         for process in cls.processes:
 151             test_utils.shutdown_process(process)
 152         print("all processes killed")
 153
 154     def setUp(self):  # instruction executed before each test method
 155         # pylint: disable=consider-using-f-string
 156         print("execution of {}".format(self.id().split(".")[-1]))
 157
 158     def test_01_connect_xpdrA(self):
 159         response = test_utils.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION))
 160         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 161
 162     def test_02_connect_xpdrC(self):
 163         response = test_utils.mount_device("XPDR-C1", ('xpdrc', self.NODE_VERSION))
 164         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 165
 166     def test_03_connect_rdmA(self):
 167         response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION))
 168         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 169
 170     def test_04_connect_rdmC(self):
 171         response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION))
 172         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 173
 174     def test_05_connect_xprdA_N1_to_roadmA_PP1(self):
 175         response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "1",
 176                                                           "ROADM-A1", "1", "SRG1-PP1-TXRX")
 177         self.assertEqual(response.status_code, requests.codes.ok)
 178         res = response.json()
 179         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
 180         time.sleep(2)
 181
 182     def test_06_connect_roadmA_PP1_to_xpdrA_N1(self):
 183         response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "1",
 184                                                           "ROADM-A1", "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', res["output"]["result"])
 188         time.sleep(2)
 189
 190     def test_07_connect_xprdC_N1_to_roadmC_PP1(self):
 191         response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "1",
 192                                                           "ROADM-C1", "1", "SRG1-PP1-TXRX")
 193         self.assertEqual(response.status_code, requests.codes.ok)
 194         res = response.json()
 195         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
 196         time.sleep(2)
 197
 198     def test_08_connect_roadmC_PP1_to_xpdrC_N1(self):
 199         response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "1",
 200                                                           "ROADM-C1", "1", "SRG1-PP1-TXRX")
 201         self.assertEqual(response.status_code, requests.codes.ok)
 202         res = response.json()
 203         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
 204         time.sleep(2)
 205
 206     def test_09_get_notifications_service1(self):
 207         data = {
 208             "input": {
 209                 "connection-type": "service",
 210                 "id-consumer": "consumer",
 211                 "group-id": "transportpceTest"
 212             }
 213         }
 214         response = test_utils.get_notifications_process_service_request(data)
 215         self.assertEqual(response.status_code, requests.codes.no_content)
 216         time.sleep(2)
 217
 218     def test_10_create_eth_service1(self):
 219         self.cr_serv_sample_data["input"]["service-name"] = "service1"
 220         response = test_utils.service_create_request(self.cr_serv_sample_data)
 221         self.assertEqual(response.status_code, requests.codes.ok)
 222         res = response.json()
 223         self.assertIn('PCE calculation in progress',
 224                       res['output']['configuration-response-common']['response-message'])
 225         time.sleep(self.WAITING)
 226
 227     def test_11_get_notifications_service1(self):
 228         data = {
 229             "input": {
 230                 "connection-type": "service",
 231                 "id-consumer": "consumer",
 232                 "group-id": "transportpceTest"
 233             }
 234         }
 235         response = test_utils.get_notifications_process_service_request(data)
 236         self.assertEqual(response.status_code, requests.codes.ok)
 237         res = response.json()
 238         self.assertEqual(res['output']['notifications-process-service'][-1]['service-name'], 'service1')
 239         self.assertEqual(res['output']['notifications-process-service'][-1]['connection-type'], 'service')
 240         self.assertEqual(res['output']['notifications-process-service'][-1]['message'],
 241                          'ServiceCreate request failed ...')
 242         self.assertEqual(res['output']['notifications-process-service'][-1]['response-failed'],
 243                          'PCE path computation failed !')
 244         time.sleep(2)
 245
 246     def test_12_add_omsAttributes_ROADMA_ROADMC(self):
 247         # Config ROADMA-ROADMC oms-attributes
 248         data = {"span": {
 249             "auto-spanloss": "true",
 250             "spanloss-base": 11.4,
 251             "spanloss-current": 12,
 252             "engineered-spanloss": 12.2,
 253             "link-concatenation": [{
 254                 "SRLG-Id": 0,
 255                 "fiber-type": "smf",
 256                 "SRLG-length": 100000,
 257                 "pmd": 0.5}]}}
 258         response = test_utils.add_oms_attr_request("ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", data)
 259         self.assertEqual(response.status_code, requests.codes.created)
 260
 261     def test_13_add_omsAttributes_ROADMC_ROADMA(self):
 262         # Config ROADMC-ROADMA oms-attributes
 263         data = {"span": {
 264             "auto-spanloss": "true",
 265             "spanloss-base": 11.4,
 266             "spanloss-current": 12,
 267             "engineered-spanloss": 12.2,
 268             "link-concatenation": [{
 269                 "SRLG-Id": 0,
 270                 "fiber-type": "smf",
 271                 "SRLG-length": 100000,
 272                 "pmd": 0.5}]}}
 273         response = test_utils.add_oms_attr_request("ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX", data)
 274         self.assertEqual(response.status_code, requests.codes.created)
 275
 276     # test service-create for Eth service from xpdr to xpdr
 277     def test_14_create_eth_service1(self):
 278         self.cr_serv_sample_data["input"]["service-name"] = "service1"
 279         response = test_utils.service_create_request(self.cr_serv_sample_data)
 280         self.assertEqual(response.status_code, requests.codes.ok)
 281         res = response.json()
 282         self.assertIn('PCE calculation in progress',
 283                       res['output']['configuration-response-common']['response-message'])
 284         time.sleep(self.WAITING)
 285
 286     def test_15_get_eth_service1(self):
 287         response = test_utils.get_service_list_request("services/service1")
 288         self.assertEqual(response.status_code, requests.codes.ok)
 289         res = response.json()
 290         self.assertEqual(
 291             res['services'][0]['administrative-state'], 'inService')
 292         self.assertEqual(
 293             res['services'][0]['service-name'], 'service1')
 294         self.assertEqual(
 295             res['services'][0]['connection-type'], 'service')
 296         self.assertEqual(
 297             res['services'][0]['lifecycle-state'], 'planned')
 298         time.sleep(2)
 299
 300     def test_16_get_notifications_service1(self):
 301         data = {
 302             "input": {
 303                 "connection-type": "service",
 304                 "id-consumer": "consumer",
 305                 "group-id": "transportpceTest"
 306             }
 307         }
 308         response = test_utils.get_notifications_process_service_request(data)
 309         self.assertEqual(response.status_code, requests.codes.ok)
 310         res = response.json()
 311         self.assertEqual(res['output']['notifications-process-service'][-1]['service-name'], 'service1')
 312         self.assertEqual(res['output']['notifications-process-service'][-1]['connection-type'], 'service')
 313         self.assertEqual(res['output']['notifications-process-service'][-1]['message'], 'Service implemented !')
 314         time.sleep(2)
 315
 316     def test_17_get_notifications_alarm_service1(self):
 317         data = {
 318             "input": {
 319                 "connection-type": "service",
 320                 "id-consumer": "consumer",
 321                 "group-id": "transportpceTest"
 322             }
 323         }
 324         response = test_utils.get_notifications_alarm_service_request(data)
 325         self.assertEqual(response.status_code, requests.codes.ok)
 326         res = response.json()
 327         self.assertEqual(res['output']['notifications-alarm-service'][-1]['service-name'], 'service1')
 328         self.assertEqual(res['output']['notifications-alarm-service'][-1]['connection-type'], 'service')
 329         self.assertEqual(res['output']['notifications-alarm-service'][-1]['operational-state'], 'inService')
 330         self.assertEqual(res['output']['notifications-alarm-service'][-1]['message'], 'The service is now inService')
 331         time.sleep(2)
 332
 333     def test_18_change_status_port_roadma_srg(self):
 334         url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
 335         body = {"ports": [{
 336             "port-name": "C1",
 337             "logical-connection-point": "SRG1-PP1",
 338             "port-type": "client",
 339             "circuit-id": "SRG1",
 340             "administrative-state": "outOfService",
 341             "port-qual": "roadm-external"}]}
 342         response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
 343                                     data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
 344                                     auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 345         self.assertEqual(response.status_code, requests.codes.ok)
 346         time.sleep(2)
 347
 348     def test_19_get_notifications_alarm_service1(self):
 349         data = {
 350             "input": {
 351                 "connection-type": "service",
 352                 "id-consumer": "consumer",
 353                 "group-id": "transportpceTest"
 354             }
 355         }
 356         response = test_utils.get_notifications_alarm_service_request(data)
 357         self.assertEqual(response.status_code, requests.codes.ok)
 358         res = response.json()
 359         self.assertEqual(res['output']['notifications-alarm-service'][-1]['service-name'], 'service1')
 360         self.assertEqual(res['output']['notifications-alarm-service'][-1]['connection-type'], 'service')
 361         self.assertEqual(res['output']['notifications-alarm-service'][-1]['operational-state'], 'outOfService')
 362         self.assertEqual(res['output']['notifications-alarm-service'][-1]['message'], 'The service is now outOfService')
 363         time.sleep(2)
 364
 365     def test_20_restore_status_port_roadma_srg(self):
 366         url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
 367         body = {"ports": [{
 368             "port-name": "C1",
 369             "logical-connection-point": "SRG1-PP1",
 370             "port-type": "client",
 371             "circuit-id": "SRG1",
 372             "administrative-state": "inService",
 373             "port-qual": "roadm-external"}]}
 374         response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
 375                                     data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
 376                                     auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 377         self.assertEqual(response.status_code, requests.codes.ok)
 378         time.sleep(2)
 379
 380     def test_21_get_notifications_alarm_service1(self):
 381         self.test_17_get_notifications_alarm_service1()
 382
 383     def test_22_delete_eth_service1(self):
 384         response = test_utils.service_delete_request("service1")
 385         self.assertEqual(response.status_code, requests.codes.ok)
 386         res = response.json()
 387         self.assertIn('Renderer service delete in progress',
 388                       res['output']['configuration-response-common']['response-message'])
 389         time.sleep(20)
 390
 391     def test_23_get_notifications_service1(self):
 392         data = {
 393             "input": {
 394                 "connection-type": "service",
 395                 "id-consumer": "consumer",
 396                 "group-id": "transportpceTest"
 397             }
 398         }
 399         response = test_utils.get_notifications_process_service_request(data)
 400         self.assertEqual(response.status_code, requests.codes.ok)
 401         res = response.json()
 402         self.assertEqual(res['output']['notifications-process-service'][-1]['service-name'], 'service1')
 403         self.assertEqual(res['output']['notifications-process-service'][-1]['connection-type'], 'service')
 404         self.assertEqual(res['output']['notifications-process-service'][-1]['message'], 'Service deleted !')
 405         time.sleep(2)
 406
 407     def test_24_disconnect_XPDRA(self):
 408         response = test_utils.unmount_device("XPDR-A1")
 409         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 410
 411     def test_25_disconnect_XPDRC(self):
 412         response = test_utils.unmount_device("XPDR-C1")
 413         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 414
 415     def test_26_disconnect_ROADMA(self):
 416         response = test_utils.unmount_device("ROADM-A1")
 417         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 418
 419     def test_27_disconnect_ROADMC(self):
 420         response = test_utils.unmount_device("ROADM-C1")
 421         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 422
 423
 424 if __name__ == "__main__":
 425     unittest.main(verbosity=2)