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