tests/transportpce_tests/2.2.1/test_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 sys
  16 import unittest
  17 import time
  18 import requests
  19 from common import test_utils
  20
  21
  22 class TransportNbiNotificationstesting(unittest.TestCase):
  23     processes = None
  24     cr_serv_sample_data = {"input": {
  25         "sdnc-request-header": {
  26             "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
  27             "rpc-action": "service-create",
  28             "request-system-id": "appname",
  29             "notification-url": "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": "XPDR-A1",
  37             "service-format": "Ethernet",
  38             "clli": "SNJSCAMCJP8",
  39             "tx-direction": {
  40                 "port": {
  41                     "port-device-name": "ROUTER_SNJSCAMCJP8_000000.00_00",
  42                     "port-type": "router",
  43                     "port-name": "Gigabit Ethernet_Tx.ge-5/0/0.0",
  44                     "port-rack": "000000.00",
  45                     "port-shelf": "00"
  46                 },
  47                 "lgx": {
  48                     "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00",
  49                     "lgx-port-name": "LGX Back.3",
  50                     "lgx-port-rack": "000000.00",
  51                     "lgx-port-shelf": "00"
  52                 }
  53             },
  54             "rx-direction": {
  55                 "port": {
  56                     "port-device-name": "ROUTER_SNJSCAMCJP8_000000.00_00",
  57                     "port-type": "router",
  58                     "port-name": "Gigabit Ethernet_Rx.ge-5/0/0.0",
  59                     "port-rack": "000000.00",
  60                     "port-shelf": "00"
  61                 },
  62                 "lgx": {
  63                     "lgx-device-name": "LGX Panel_SNJSCAMCJP8_000000.00_00",
  64                     "lgx-port-name": "LGX Back.4",
  65                     "lgx-port-rack": "000000.00",
  66                     "lgx-port-shelf": "00"
  67                 }
  68             },
  69             "optic-type": "gray"
  70         },
  71         "service-z-end": {
  72             "service-rate": "100",
  73             "node-id": "XPDR-C1",
  74             "service-format": "Ethernet",
  75             "clli": "SNJSCAMCJT4",
  76             "tx-direction": {
  77                 "port": {
  78                     "port-device-name": "ROUTER_SNJSCAMCJT4_000000.00_00",
  79                     "port-type": "router",
  80                     "port-name": "Gigabit Ethernet_Tx.ge-1/0/0.0",
  81                     "port-rack": "000000.00",
  82                     "port-shelf": "00"
  83                 },
  84                 "lgx": {
  85                     "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00",
  86                     "lgx-port-name": "LGX Back.29",
  87                     "lgx-port-rack": "000000.00",
  88                     "lgx-port-shelf": "00"
  89                 }
  90             },
  91             "rx-direction": {
  92                 "port": {
  93                     "port-device-name": "ROUTER_SNJSCAMCJT4_000000.00_00",
  94                     "port-type": "router",
  95                     "port-name": "Gigabit Ethernet_Rx.ge-1/0/0.0",
  96                     "port-rack": "000000.00",
  97                     "port-shelf": "00"
  98                 },
  99                 "lgx": {
 100                     "lgx-device-name": "LGX Panel_SNJSCAMCJT4_000000.00_00",
 101                     "lgx-port-name": "LGX Back.30",
 102                     "lgx-port-rack": "000000.00",
 103                     "lgx-port-shelf": "00"
 104                 }
 105             },
 106             "optic-type": "gray"
 107         },
 108         "due-date": "2016-11-28T00:00:01Z",
 109         "operator-contact": "pw1234"
 110     }
 111     }
 112
 113     WAITING = 20  # nominal value is 300
 114
 115     @classmethod
 116     def setUpClass(cls):
 117         # TODO: for lighty manage the activation of NBI notification feature
 118         cls.processes = test_utils.start_tpce()
 119         # NBI notification feature is not installed by default in Karaf
 120         if "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
 121             print("installing NBI notification feature...")
 122             result = test_utils.install_karaf_feature("odl-transportpce-nbinotifications")
 123             if result.returncode != 0:
 124                 cls.init_failed = True
 125             print("Restarting OpenDaylight...")
 126             test_utils.shutdown_process(cls.processes[0])
 127             cls.processes[0] = test_utils.start_karaf()
 128             test_utils.process_list[0] = cls.processes[0]
 129             cls.init_failed = not test_utils.wait_until_log_contains(
 130                 test_utils.KARAF_LOG, test_utils.KARAF_OK_START_MSG, time_to_wait=60)
 131         if cls.init_failed:
 132             print("NBI notification installation feature failed...")
 133             test_utils.shutdown_process(cls.processes[0])
 134             sys.exit(2)
 135         cls.processes = test_utils.start_sims(['xpdra', 'roadma', 'roadmc', 'xpdrc'])
 136
 137     @classmethod
 138     def tearDownClass(cls):
 139         # pylint: disable=not-an-iterable
 140         for process in cls.processes:
 141             test_utils.shutdown_process(process)
 142         print("all processes killed")
 143
 144     def setUp(self):  # instruction executed before each test method
 145         print("execution of {}".format(self.id().split(".")[-1]))
 146
 147     def test_01_connect_xpdrA(self):
 148         response = test_utils.mount_device("XPDR-A1", 'xpdra')
 149         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 150
 151     def test_02_connect_xpdrC(self):
 152         response = test_utils.mount_device("XPDR-C1", 'xpdrc')
 153         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 154
 155     def test_03_connect_rdmA(self):
 156         response = test_utils.mount_device("ROADM-A1", 'roadma')
 157         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 158
 159     def test_04_connect_rdmC(self):
 160         response = test_utils.mount_device("ROADM-C1", 'roadmc')
 161         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 162
 163     def test_05_connect_xprdA_N1_to_roadmA_PP1(self):
 164         response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "1",
 165                                                           "ROADM-A1", "1", "SRG1-PP1-TXRX")
 166         self.assertEqual(response.status_code, requests.codes.ok)
 167         res = response.json()
 168         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
 169         time.sleep(2)
 170
 171     def test_06_connect_roadmA_PP1_to_xpdrA_N1(self):
 172         response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "1",
 173                                                           "ROADM-A1", "1", "SRG1-PP1-TXRX")
 174         self.assertEqual(response.status_code, requests.codes.ok)
 175         res = response.json()
 176         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
 177         time.sleep(2)
 178
 179     def test_07_connect_xprdC_N1_to_roadmC_PP1(self):
 180         response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "1",
 181                                                           "ROADM-C1", "1", "SRG1-PP1-TXRX")
 182         self.assertEqual(response.status_code, requests.codes.ok)
 183         res = response.json()
 184         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
 185         time.sleep(2)
 186
 187     def test_08_connect_roadmC_PP1_to_xpdrC_N1(self):
 188         response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "1",
 189                                                           "ROADM-C1", "1", "SRG1-PP1-TXRX")
 190         self.assertEqual(response.status_code, requests.codes.ok)
 191         res = response.json()
 192         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
 193         time.sleep(2)
 194
 195     def test_09_get_notifications_service1(self):
 196         data = {
 197             "input": {
 198                 "connection-type": "service",
 199                 "id-consumer": "consumer",
 200                 "group-id": "transportpceTest"
 201             }
 202         }
 203         response = test_utils.get_notifications_service_request(data)
 204         self.assertEqual(response.status_code, requests.codes.no_content)
 205         time.sleep(2)
 206
 207     def test_10_create_eth_service1(self):
 208         self.cr_serv_sample_data["input"]["service-name"] = "service1"
 209         response = test_utils.service_create_request(self.cr_serv_sample_data)
 210         self.assertEqual(response.status_code, requests.codes.ok)
 211         res = response.json()
 212         self.assertIn('PCE calculation in progress',
 213                       res['output']['configuration-response-common']['response-message'])
 214         time.sleep(self.WAITING)
 215
 216     def test_11_get_notifications_service1(self):
 217         data = {
 218             "input": {
 219                 "connection-type": "service",
 220                 "id-consumer": "consumer",
 221                 "group-id": "transportpceTest"
 222             }
 223         }
 224         response = test_utils.get_notifications_service_request(data)
 225         self.assertEqual(response.status_code, requests.codes.ok)
 226         res = response.json()
 227         self.assertEqual(res['output']['notification-service'][-2]['service-name'], 'service1')
 228         self.assertEqual(res['output']['notification-service'][-2]['connection-type'], 'service')
 229         self.assertEqual(res['output']['notification-service'][-2]['message'], 'ServiceCreate request received ...')
 230         self.assertEqual(res['output']['notification-service'][-1]['service-name'], 'service1')
 231         self.assertEqual(res['output']['notification-service'][-1]['message'], 'ServiceCreate request failed ...')
 232         self.assertEqual(res['output']['notification-service'][-1]['response-failed'],
 233                          'PCE path computation failed !')
 234         time.sleep(2)
 235
 236     def test_12_add_omsAttributes_ROADMA_ROADMC(self):
 237         # Config ROADMA-ROADMC oms-attributes
 238         data = {"span": {
 239             "auto-spanloss": "true",
 240             "spanloss-base": 11.4,
 241             "spanloss-current": 12,
 242             "engineered-spanloss": 12.2,
 243             "link-concatenation": [{
 244                 "SRLG-Id": 0,
 245                 "fiber-type": "smf",
 246                 "SRLG-length": 100000,
 247                 "pmd": 0.5}]}}
 248         response = test_utils.add_oms_attr_request("ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", data)
 249         self.assertEqual(response.status_code, requests.codes.created)
 250
 251     def test_13_add_omsAttributes_ROADMC_ROADMA(self):
 252         # Config ROADMC-ROADMA oms-attributes
 253         data = {"span": {
 254             "auto-spanloss": "true",
 255             "spanloss-base": 11.4,
 256             "spanloss-current": 12,
 257             "engineered-spanloss": 12.2,
 258             "link-concatenation": [{
 259                 "SRLG-Id": 0,
 260                 "fiber-type": "smf",
 261                 "SRLG-length": 100000,
 262                 "pmd": 0.5}]}}
 263         response = test_utils.add_oms_attr_request("ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX", data)
 264         self.assertEqual(response.status_code, requests.codes.created)
 265
 266     # test service-create for Eth service from xpdr to xpdr
 267     def test_14_create_eth_service1(self):
 268         self.cr_serv_sample_data["input"]["service-name"] = "service1"
 269         response = test_utils.service_create_request(self.cr_serv_sample_data)
 270         self.assertEqual(response.status_code, requests.codes.ok)
 271         res = response.json()
 272         self.assertIn('PCE calculation in progress',
 273                       res['output']['configuration-response-common']['response-message'])
 274         time.sleep(self.WAITING)
 275
 276     def test_15_get_eth_service1(self):
 277         response = test_utils.get_service_list_request("services/service1")
 278         self.assertEqual(response.status_code, requests.codes.ok)
 279         res = response.json()
 280         self.assertEqual(
 281             res['services'][0]['administrative-state'], 'inService')
 282         self.assertEqual(
 283             res['services'][0]['service-name'], 'service1')
 284         self.assertEqual(
 285             res['services'][0]['connection-type'], 'service')
 286         self.assertEqual(
 287             res['services'][0]['lifecycle-state'], 'planned')
 288         time.sleep(2)
 289
 290     def test_16_get_notifications_service1(self):
 291         data = {
 292             "input": {
 293                 "connection-type": "service",
 294                 "id-consumer": "consumer",
 295                 "group-id": "transportpceTest"
 296             }
 297         }
 298         response = test_utils.get_notifications_service_request(data)
 299         self.assertEqual(response.status_code, requests.codes.ok)
 300         res = response.json()
 301         self.assertEqual(res['output']['notification-service'][-3]['service-name'], 'service1')
 302         self.assertEqual(res['output']['notification-service'][-3]['connection-type'], 'service')
 303         self.assertEqual(res['output']['notification-service'][-3]['message'], 'ServiceCreate request received ...')
 304         self.assertEqual(res['output']['notification-service'][-2]['service-name'], 'service1')
 305         self.assertEqual(res['output']['notification-service'][-2]['message'], 'PCE calculation done OK !')
 306         self.assertEqual(res['output']['notification-service'][-1]['service-name'], 'service1')
 307         self.assertEqual(res['output']['notification-service'][-1]['message'], 'Service implemented !')
 308         time.sleep(2)
 309
 310     def test_17_delete_eth_service1(self):
 311         response = test_utils.service_delete_request("service1")
 312         self.assertEqual(response.status_code, requests.codes.ok)
 313         res = response.json()
 314         self.assertIn('Renderer service delete in progress',
 315                       res['output']['configuration-response-common']['response-message'])
 316         time.sleep(20)
 317
 318     def test_18_get_notifications_service1(self):
 319         data = {
 320             "input": {
 321                 "connection-type": "service",
 322                 "id-consumer": "consumer",
 323                 "group-id": "transportpceTest"
 324             }
 325         }
 326         response = test_utils.get_notifications_service_request(data)
 327         self.assertEqual(response.status_code, requests.codes.ok)
 328         res = response.json()
 329         self.assertEqual(res['output']['notification-service'][-2]['service-name'], 'service1')
 330         self.assertEqual(res['output']['notification-service'][-2]['connection-type'], 'service')
 331         self.assertEqual(res['output']['notification-service'][-2]['message'], 'ServiceDelete request received ...')
 332         self.assertEqual(res['output']['notification-service'][-1]['service-name'], 'service1')
 333         self.assertEqual(res['output']['notification-service'][-1]['message'], 'Service deleted !')
 334         time.sleep(2)
 335
 336     def test_19_disconnect_XPDRA(self):
 337         response = test_utils.unmount_device("XPDR-A1")
 338         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 339
 340     def test_20_disconnect_XPDRC(self):
 341         response = test_utils.unmount_device("XPDR-C1")
 342         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 343
 344     def test_21_disconnect_ROADMA(self):
 345         response = test_utils.unmount_device("ROADM-A1")
 346         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 347
 348     def test_22_disconnect_ROADMC(self):
 349         response = test_utils.unmount_device("ROADM-C1")
 350         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 351
 352
 353 if __name__ == "__main__":
 354     unittest.main(verbosity=2)