tests/transportpce_tests/with_docker/test03_tapi_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 # pylint: disable=wrong-import-order
  16 import sys
  17 import unittest
  18 import time
  19 import requests
  20 sys.path.append('transportpce_tests/common/')
  21 # pylint: disable=wrong-import-position
  22 # pylint: disable=import-error
  23 import test_utils  # nopep8
  24
  25
  26 # pylint: disable=too-few-public-methods
  27 class UuidServices:
  28     def __init__(self):
  29         # pylint: disable=invalid-name
  30         self.pm = None
  31         self.odu = None
  32         self.dsr = None
  33         self.eth = None
  34
  35
  36 # pylint: disable=too-few-public-methods
  37 class UuidSubscriptions:
  38     def __init__(self):
  39         # pylint: disable=invalid-name
  40         self.pm = None
  41         self.odu = None
  42         self.dsr = None
  43         self.eth = None
  44
  45
  46 class TransportNbiNotificationstesting(unittest.TestCase):
  47     cr_serv_input_data = {
  48         "end-point": [
  49             {
  50                 "layer-protocol-name": "DSR",
  51                 "service-interface-point": {
  52                     "service-interface-point-uuid": "b1f4bd3b-7fa9-367b-a8ab-6e80293238df"
  53                 },
  54                 "administrative-state": "UNLOCKED",
  55                 "operational-state": "ENABLED",
  56                 "direction": "BIDIRECTIONAL",
  57                 "role": "SYMMETRIC",
  58                 "protection-role": "WORK",
  59                 "local-id": "XPDR-C1-XPDR1",
  60                 "name": [
  61                         {
  62                             "value-name": "OpenROADM node id",
  63                             "value": "XPDR-C1-XPDR1"
  64                         }
  65                 ]
  66             },
  67             {
  68                 "layer-protocol-name": "DSR",
  69                 "service-interface-point": {
  70                     "service-interface-point-uuid": "b5964ce9-274c-3f68-b4d1-83c0b61bc74e"
  71                 },
  72                 "administrative-state": "UNLOCKED",
  73                 "operational-state": "ENABLED",
  74                 "direction": "BIDIRECTIONAL",
  75                 "role": "SYMMETRIC",
  76                 "protection-role": "WORK",
  77                 "local-id": "XPDR-A1-XPDR1",
  78                 "name": [
  79                         {
  80                             "value-name": "OpenROADM node id",
  81                             "value": "XPDR-A1-XPDR1"
  82                         }
  83                 ]
  84             }
  85         ],
  86         "connectivity-constraint": {
  87             "service-layer": "ETH",
  88             "service-type": "POINT_TO_POINT_CONNECTIVITY",
  89             "service-level": "Some service-level",
  90             "requested-capacity": {
  91                 "total-size": {
  92                     "value": "100",
  93                     "unit": "GB"
  94                 }
  95             }
  96         },
  97         "state": "Some state"
  98     }
  99
 100     tapi_serv_details = {"service-id-or-name": "TBD"}
 101
 102     cr_notif_subs_input_data = {
 103         "subscription-filter": {
 104             "requested-notification-types": [
 105                 "ALARM_EVENT"
 106             ],
 107             "requested-object-types": [
 108                 "CONNECTIVITY_SERVICE"
 109             ],
 110             "requested-layer-protocols": [
 111                 "ETH"
 112             ],
 113             "requested-object-identifier": [
 114                 "76d8f07b-ead5-4132-8eb8-cf3fdef7e079"
 115             ],
 116             "include-content": True,
 117             "local-id": "localId",
 118             "name": [
 119                 {
 120                         "value-name": "Subscription name",
 121                         "value": "test subscription"
 122                 }
 123             ]
 124         },
 125         "subscription-state": "ACTIVE"
 126     }
 127
 128     cr_get_notif_list_input_data = {
 129         "subscription-id-or-name": "c07e7fd1-0377-4fbf-8928-36c17b0d0d68",
 130         "time-period": "time-period"
 131     }
 132
 133     processes = []
 134     uuid_services = UuidServices()
 135     uuid_subscriptions = UuidSubscriptions()
 136     WAITING = 25  # nominal value is 300
 137     NODE_VERSION_221 = '2.2.1'
 138
 139     @classmethod
 140     def setUpClass(cls):
 141         # pylint: disable=unsubscriptable-object
 142         # TODO: for lighty manage the activation of NBI notification feature
 143         cls.init_failed_nbi = False
 144         cls.init_failed_tapi = False
 145         os.environ['JAVA_MIN_MEM'] = '1024M'
 146         os.environ['JAVA_MAX_MEM'] = '4096M'
 147         cls.processes = test_utils.start_tpce()
 148         # NBI notification feature is not installed by default in Karaf
 149         if "NO_ODL_STARTUP" not in os.environ or "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
 150             print("installing NBI notification feature...")
 151             result = test_utils.install_karaf_feature("odl-transportpce-nbinotifications")
 152             if result.returncode != 0:
 153                 cls.init_failed_nbi = True
 154             print("installing tapi feature...")
 155             result = test_utils.install_karaf_feature("odl-transportpce-tapi")
 156             if result.returncode != 0:
 157                 cls.init_failed_tapi = True
 158         if cls.init_failed_nbi:
 159             print("NBI notification installation feature failed...")
 160             test_utils.shutdown_process(cls.processes[0])
 161             sys.exit(2)
 162         if cls.init_failed_tapi:
 163             print("tapi installation feature failed...")
 164             test_utils.shutdown_process(cls.processes[0])
 165             sys.exit(2)
 166         cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION_221),
 167                                                ('roadma', cls.NODE_VERSION_221),
 168                                                ('roadmc', cls.NODE_VERSION_221),
 169                                                ('xpdrc', cls.NODE_VERSION_221)])
 170
 171     @classmethod
 172     def tearDownClass(cls):
 173         # pylint: disable=not-an-iterable
 174         for process in cls.processes:
 175             test_utils.shutdown_process(process)
 176         print("all processes killed")
 177
 178     def setUp(self):  # instruction executed before each test method
 179         # pylint: disable=consider-using-f-string
 180         print("execution of {}".format(self.id().split(".")[-1]))
 181
 182     def test_01_connect_xpdrA(self):
 183         response = test_utils.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION_221))
 184         self.assertEqual(response.status_code,
 185                          requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 186
 187     def test_02_connect_xpdrC(self):
 188         response = test_utils.mount_device("XPDR-C1", ('xpdrc', self.NODE_VERSION_221))
 189         self.assertEqual(response.status_code,
 190                          requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 191
 192     def test_03_connect_rdmA(self):
 193         response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION_221))
 194         self.assertEqual(response.status_code,
 195                          requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 196
 197     def test_04_connect_rdmC(self):
 198         response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION_221))
 199         self.assertEqual(response.status_code,
 200                          requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 201
 202     def test_05_connect_xpdrA_N1_to_roadmA_PP1(self):
 203         response = test_utils.transportpce_api_rpc_request(
 204             'transportpce-networkutils', 'init-xpdr-rdm-links',
 205             {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1',
 206                              'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 207         self.assertEqual(response['status_code'], requests.codes.ok)
 208         self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
 209         time.sleep(2)
 210
 211     def test_06_connect_roadmA_PP1_to_xpdrA_N1(self):
 212         response = test_utils.transportpce_api_rpc_request(
 213             'transportpce-networkutils', 'init-rdm-xpdr-links',
 214             {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1',
 215                              'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 216         self.assertEqual(response['status_code'], requests.codes.ok)
 217         self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
 218         time.sleep(2)
 219
 220     def test_07_connect_xpdrC_N1_to_roadmC_PP1(self):
 221         response = test_utils.transportpce_api_rpc_request(
 222             'transportpce-networkutils', 'init-xpdr-rdm-links',
 223             {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1',
 224                              'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 225         self.assertEqual(response['status_code'], requests.codes.ok)
 226         self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
 227         time.sleep(2)
 228
 229     def test_08_connect_roadmC_PP1_to_xpdrC_N1(self):
 230         response = test_utils.transportpce_api_rpc_request(
 231             'transportpce-networkutils', 'init-rdm-xpdr-links',
 232             {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1',
 233                              'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 234         self.assertEqual(response['status_code'], requests.codes.ok)
 235         self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
 236         time.sleep(2)
 237
 238     def test_09_add_omsAttributes_ROADMA_ROADMC(self):
 239         # Config ROADMA-ROADMC oms-attributes
 240         data = {"span": {
 241             "auto-spanloss": "true",
 242             "spanloss-base": 11.4,
 243             "spanloss-current": 12,
 244             "engineered-spanloss": 12.2,
 245             "link-concatenation": [{
 246                 "SRLG-Id": 0,
 247                 "fiber-type": "smf",
 248                 "SRLG-length": 100000,
 249                 "pmd": 0.5}]}}
 250         response = test_utils.add_oms_attr_request(
 251             "ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", data)
 252         self.assertEqual(response.status_code, requests.codes.created)
 253
 254     def test_10_add_omsAttributes_ROADMC_ROADMA(self):
 255         # Config ROADMC-ROADMA oms-attributes
 256         data = {"span": {
 257             "auto-spanloss": "true",
 258             "spanloss-base": 11.4,
 259             "spanloss-current": 12,
 260             "engineered-spanloss": 12.2,
 261             "link-concatenation": [{
 262                 "SRLG-Id": 0,
 263                 "fiber-type": "smf",
 264                 "SRLG-length": 100000,
 265                 "pmd": 0.5}]}}
 266         response = test_utils.add_oms_attr_request(
 267             "ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX", data)
 268         self.assertEqual(response.status_code, requests.codes.created)
 269
 270     # test service-create for Eth service from xpdr to xpdr
 271     def test_11_create_connectivity_service_Ethernet(self):
 272         response = test_utils.transportpce_api_rpc_request(
 273             'tapi-connectivity', 'create-connectivity-service', self.cr_serv_input_data)
 274         time.sleep(self.WAITING)
 275         self.assertEqual(response['status_code'], requests.codes.ok)
 276         self.uuid_services.eth = response['output']['service']['uuid']
 277         # pylint: disable=consider-using-f-string
 278
 279         input_dict_1 = {'administrative-state': 'LOCKED',
 280                         'lifecycle-state': 'PLANNED',
 281                         'operational-state': 'DISABLED',
 282                         'service-type': 'POINT_TO_POINT_CONNECTIVITY',
 283                         'service-layer': 'ETH',
 284                         'connectivity-direction': 'BIDIRECTIONAL'
 285                         }
 286         input_dict_2 = {'value-name': 'OpenROADM node id',
 287                         'value': 'XPDR-C1-XPDR1'}
 288         input_dict_3 = {'value-name': 'OpenROADM node id',
 289                         'value': 'XPDR-A1-XPDR1'}
 290
 291         self.assertDictEqual(dict(input_dict_1, **response['output']['service']),
 292                              response['output']['service'])
 293         self.assertDictEqual(dict(input_dict_2, **response['output']['service']['end-point'][0]['name'][0]),
 294                              response['output']['service']['end-point'][0]['name'][0])
 295         self.assertDictEqual(dict(input_dict_3, **response['output']['service']['end-point'][1]['name'][0]),
 296                              response['output']['service']['end-point'][1]['name'][0])
 297
 298     def test_12_get_service_Ethernet(self):
 299         response = test_utils.get_ordm_serv_list_attr_request("services", str(self.uuid_services.eth))
 300         self.assertEqual(response['status_code'], requests.codes.ok)
 301         self.assertEqual(response['services'][0]['administrative-state'], 'inService')
 302         self.assertEqual(response['services'][0]['service-name'], str(self.uuid_services.eth))
 303         self.assertEqual(response['services'][0]['connection-type'], 'service')
 304         self.assertEqual(response['services'][0]['lifecycle-state'], 'planned')
 305         time.sleep(1)
 306
 307     def test_13_get_connectivity_service_Ethernet(self):
 308         self.tapi_serv_details["service-id-or-name"] = str(self.uuid_services.eth)
 309         response = test_utils.transportpce_api_rpc_request(
 310             'tapi-connectivity', 'get-connectivity-service-details', self.tapi_serv_details)
 311         self.assertEqual(response['status_code'], requests.codes.ok)
 312         self.assertEqual(response['output']['service']['operational-state'], 'ENABLED')
 313         self.assertEqual(response['output']['service']['name'][0]['value'], self.uuid_services.eth)
 314         self.assertEqual(response['output']['service']['administrative-state'], 'UNLOCKED')
 315         self.assertEqual(response['output']['service']['lifecycle-state'], 'INSTALLED')
 316
 317     def test_14_create_notifications_subscription_service(self):
 318         self.cr_notif_subs_input_data["subscription-filter"]["requested-object-identifier"][0] =\
 319             str(self.uuid_services.eth)
 320         response = test_utils.transportpce_api_rpc_request(
 321             'tapi-notification', 'create-notification-subscription-service', self.cr_notif_subs_input_data)
 322         self.assertEqual(response['status_code'], requests.codes.ok)
 323         self.uuid_subscriptions.eth = response['output']['subscription-service']['uuid']
 324         self.assertEqual(response['output']['subscription-service']['subscription-filter']
 325                          ['requested-object-types'][0], 'CONNECTIVITY_SERVICE')
 326         self.assertEqual(response['output']['subscription-service']['subscription-filter']
 327                          ['requested-notification-types'][0], 'ALARM_EVENT')
 328         self.assertEqual(response['output']['subscription-service']['subscription-filter']
 329                          ['requested-object-identifier'][0], str(self.uuid_services.eth))
 330         time.sleep(2)
 331
 332     def test_15_change_status_port_roadma_srg(self):
 333         self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3%2F0', 'C1',
 334                                                        {
 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         }))
 342
 343     def test_16_get_tapi_notifications_connectivity_service_Ethernet(self):
 344         self.cr_get_notif_list_input_data["subscription-id-or-name"] = str(self.uuid_subscriptions.eth)
 345         response = test_utils.transportpce_api_rpc_request(
 346             'tapi-notification', 'get-notification-list', self.cr_get_notif_list_input_data)
 347         self.assertEqual(response['status_code'], requests.codes.ok)
 348         self.assertEqual(response['output']['notification'][0]['target-object-identifier'], str(self.uuid_services.eth))
 349         self.assertEqual(response['output']['notification'][0]['target-object-type'], 'CONNECTIVITY_SERVICE')
 350         self.assertEqual(response['output']['notification'][0]['changed-attributes'][0]['new-value'], 'LOCKED')
 351         self.assertEqual(response['output']['notification'][0]['changed-attributes'][1]['new-value'], 'DISABLED')
 352         time.sleep(2)
 353
 354     def test_17_restore_status_port_roadma_srg(self):
 355         self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3%2F0', 'C1',
 356                                                        {
 357             "port-name": "C1",
 358             "logical-connection-point": "SRG1-PP1",
 359             "port-type": "client",
 360             "circuit-id": "SRG1",
 361             "administrative-state": "inService",
 362             "port-qual": "roadm-external"
 363         }))
 364
 365     def test_18_get_tapi_notifications_connectivity_service_Ethernet(self):
 366         self.cr_get_notif_list_input_data["subscription-id-or-name"] = str(self.uuid_subscriptions.eth)
 367         response = test_utils.transportpce_api_rpc_request(
 368             'tapi-notification', 'get-notification-list', self.cr_get_notif_list_input_data)
 369         self.assertEqual(response['status_code'], requests.codes.ok)
 370         self.assertEqual(response['output']['notification'][0]['target-object-identifier'], str(self.uuid_services.eth))
 371         self.assertEqual(response['output']['notification'][0]['target-object-type'], 'CONNECTIVITY_SERVICE')
 372         self.assertEqual(response['output']['notification'][0]['changed-attributes'][0]['new-value'], 'UNLOCKED')
 373         self.assertEqual(response['output']['notification'][0]['changed-attributes'][1]['new-value'], 'ENABLED')
 374         time.sleep(2)
 375
 376     def test_19_delete_connectivity_service_Ethernet(self):
 377         self.tapi_serv_details["service-id-or-name"] = str(self.uuid_services.eth)
 378         response = test_utils.transportpce_api_rpc_request(
 379             'tapi-connectivity', 'delete-connectivity-service', self.tapi_serv_details)
 380         self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content))
 381         time.sleep(self.WAITING)
 382
 383     def test_20_disconnect_XPDRA(self):
 384         response = test_utils.unmount_device("XPDR-A1")
 385         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 386
 387     def test_21_disconnect_XPDRC(self):
 388         response = test_utils.unmount_device("XPDR-C1")
 389         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 390
 391     def test_22_disconnect_ROADMA(self):
 392         response = test_utils.unmount_device("ROADM-A1")
 393         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 394
 395     def test_23_disconnect_ROADMC(self):
 396         response = test_utils.unmount_device("ROADM-C1")
 397         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 398
 399
 400 if __name__ == "__main__":
 401     unittest.main(verbosity=2)