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