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