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         if cls.init_failed_nbi:
 160             print("NBI notification installation feature failed...")
 161             test_utils.shutdown_process(cls.processes[0])
 162             sys.exit(2)
 163         if cls.init_failed_tapi:
 164             print("tapi installation feature failed...")
 165             test_utils.shutdown_process(cls.processes[0])
 166             sys.exit(2)
 167         cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION_221),
 168                                                ('roadma', cls.NODE_VERSION_221),
 169                                                ('roadmc', cls.NODE_VERSION_221),
 170                                                ('xpdrc', cls.NODE_VERSION_221)])
 171
 172     @classmethod
 173     def tearDownClass(cls):
 174         # pylint: disable=not-an-iterable
 175         for process in cls.processes:
 176             test_utils.shutdown_process(process)
 177         print("all processes killed")
 178
 179     def setUp(self):  # instruction executed before each test method
 180         # pylint: disable=consider-using-f-string
 181         print("execution of {}".format(self.id().split(".")[-1]))
 182
 183     def test_01_connect_xpdrA(self):
 184         response = test_utils.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION_221))
 185         self.assertEqual(response.status_code,
 186                          requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 187
 188     def test_02_connect_xpdrC(self):
 189         response = test_utils.mount_device("XPDR-C1", ('xpdrc', self.NODE_VERSION_221))
 190         self.assertEqual(response.status_code,
 191                          requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 192
 193     def test_03_connect_rdmA(self):
 194         response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION_221))
 195         self.assertEqual(response.status_code,
 196                          requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 197
 198     def test_04_connect_rdmC(self):
 199         response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION_221))
 200         self.assertEqual(response.status_code,
 201                          requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 202
 203     def test_05_connect_xprdA_N1_to_roadmA_PP1(self):
 204         response = test_utils.transportpce_api_rpc_request(
 205             'transportpce-networkutils', 'init-xpdr-rdm-links',
 206             {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1',
 207                              'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 208         self.assertEqual(response['status_code'], requests.codes.ok)
 209         self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
 210         time.sleep(2)
 211
 212     def test_06_connect_roadmA_PP1_to_xpdrA_N1(self):
 213         response = test_utils.transportpce_api_rpc_request(
 214             'transportpce-networkutils', 'init-rdm-xpdr-links',
 215             {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1',
 216                              'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 217         self.assertEqual(response['status_code'], requests.codes.ok)
 218         self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
 219         time.sleep(2)
 220
 221     def test_07_connect_xprdC_N1_to_roadmC_PP1(self):
 222         response = test_utils.transportpce_api_rpc_request(
 223             'transportpce-networkutils', 'init-xpdr-rdm-links',
 224             {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1',
 225                              'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 226         self.assertEqual(response['status_code'], requests.codes.ok)
 227         self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
 228         time.sleep(2)
 229
 230     def test_08_connect_roadmC_PP1_to_xpdrC_N1(self):
 231         response = test_utils.transportpce_api_rpc_request(
 232             'transportpce-networkutils', 'init-rdm-xpdr-links',
 233             {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1',
 234                              'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 235         self.assertEqual(response['status_code'], requests.codes.ok)
 236         self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
 237         time.sleep(2)
 238
 239     def test_09_add_omsAttributes_ROADMA_ROADMC(self):
 240         # Config ROADMA-ROADMC oms-attributes
 241         data = {"span": {
 242             "auto-spanloss": "true",
 243             "spanloss-base": 11.4,
 244             "spanloss-current": 12,
 245             "engineered-spanloss": 12.2,
 246             "link-concatenation": [{
 247                 "SRLG-Id": 0,
 248                 "fiber-type": "smf",
 249                 "SRLG-length": 100000,
 250                 "pmd": 0.5}]}}
 251         response = test_utils.add_oms_attr_request(
 252             "ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", data)
 253         self.assertEqual(response.status_code, requests.codes.created)
 254
 255     def test_10_add_omsAttributes_ROADMC_ROADMA(self):
 256         # Config ROADMC-ROADMA oms-attributes
 257         data = {"span": {
 258             "auto-spanloss": "true",
 259             "spanloss-base": 11.4,
 260             "spanloss-current": 12,
 261             "engineered-spanloss": 12.2,
 262             "link-concatenation": [{
 263                 "SRLG-Id": 0,
 264                 "fiber-type": "smf",
 265                 "SRLG-length": 100000,
 266                 "pmd": 0.5}]}}
 267         response = test_utils.add_oms_attr_request(
 268             "ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX", data)
 269         self.assertEqual(response.status_code, requests.codes.created)
 270
 271     # test service-create for Eth service from xpdr to xpdr
 272     def test_11_create_connectivity_service_Ethernet(self):
 273         response = test_utils.transportpce_api_rpc_request(
 274             'tapi-connectivity', 'create-connectivity-service', self.cr_serv_input_data)
 275         time.sleep(self.WAITING)
 276         self.assertEqual(response['status_code'], requests.codes.ok)
 277         self.uuid_services.eth = response['output']['service']['uuid']
 278         # pylint: disable=consider-using-f-string
 279
 280         input_dict_1 = {'administrative-state': 'LOCKED',
 281                         'lifecycle-state': 'PLANNED',
 282                         'operational-state': 'DISABLED',
 283                         'service-type': 'POINT_TO_POINT_CONNECTIVITY',
 284                         'service-layer': 'ETH',
 285                         'connectivity-direction': 'BIDIRECTIONAL'
 286                         }
 287         input_dict_2 = {'value-name': 'OpenROADM node id',
 288                         'value': 'XPDR-C1-XPDR1'}
 289         input_dict_3 = {'value-name': 'OpenROADM node id',
 290                         'value': 'XPDR-A1-XPDR1'}
 291
 292         self.assertDictEqual(dict(input_dict_1, **response['output']['service']),
 293                              response['output']['service'])
 294         self.assertDictEqual(dict(input_dict_2, **response['output']['service']['end-point'][0]['name'][0]),
 295                              response['output']['service']['end-point'][0]['name'][0])
 296         self.assertDictEqual(dict(input_dict_3, **response['output']['service']['end-point'][1]['name'][0]),
 297                              response['output']['service']['end-point'][1]['name'][0])
 298
 299     def test_12_get_service_Ethernet(self):
 300         response = test_utils.get_ordm_serv_list_attr_request("services", str(self.uuid_services.eth))
 301         self.assertEqual(response['status_code'], requests.codes.ok)
 302         self.assertEqual(response['services'][0]['administrative-state'], 'inService')
 303         self.assertEqual(response['services'][0]['service-name'], str(self.uuid_services.eth))
 304         self.assertEqual(response['services'][0]['connection-type'], 'service')
 305         self.assertEqual(response['services'][0]['lifecycle-state'], 'planned')
 306         time.sleep(1)
 307
 308     def test_13_get_connectivity_service_Ethernet(self):
 309         self.tapi_serv_details["service-id-or-name"] = str(self.uuid_services.eth)
 310         response = test_utils.transportpce_api_rpc_request(
 311             'tapi-connectivity', 'get-connectivity-service-details', self.tapi_serv_details)
 312         self.assertEqual(response['status_code'], requests.codes.ok)
 313         self.assertEqual(response['output']['service']['operational-state'], 'ENABLED')
 314         self.assertEqual(response['output']['service']['name'][0]['value'], self.uuid_services.eth)
 315         self.assertEqual(response['output']['service']['administrative-state'], 'UNLOCKED')
 316         self.assertEqual(response['output']['service']['lifecycle-state'], 'INSTALLED')
 317
 318     def test_14_create_notifications_subscription_service(self):
 319         self.cr_notif_subs_input_data["subscription-filter"]["requested-object-identifier"][0] =\
 320             str(self.uuid_services.eth)
 321         response = test_utils.transportpce_api_rpc_request(
 322             'tapi-notification', 'create-notification-subscription-service', self.cr_notif_subs_input_data)
 323         self.assertEqual(response['status_code'], requests.codes.ok)
 324         self.uuid_subscriptions.eth = response['output']['subscription-service']['uuid']
 325         self.assertEqual(response['output']['subscription-service']['subscription-filter']
 326                          ['requested-object-types'][0], 'CONNECTIVITY_SERVICE')
 327         self.assertEqual(response['output']['subscription-service']['subscription-filter']
 328                          ['requested-notification-types'][0], 'ALARM_EVENT')
 329         self.assertEqual(response['output']['subscription-service']['subscription-filter']
 330                          ['requested-object-identifier'][0], str(self.uuid_services.eth))
 331         time.sleep(2)
 332
 333     def test_15_change_status_port_roadma_srg(self):
 334         url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
 335         body = {"ports": [{
 336             "port-name": "C1",
 337             "logical-connection-point": "SRG1-PP1",
 338             "port-type": "client",
 339             "circuit-id": "SRG1",
 340             "administrative-state": "outOfService",
 341             "port-qual": "roadm-external"}]}
 342         response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
 343                                     data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
 344                                     auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
 345                                     timeout=test_utils.REQUEST_TIMEOUT)
 346         self.assertEqual(response.status_code, requests.codes.ok)
 347         # If the gate fails is because of the waiting time not being enough
 348         time.sleep(2)
 349
 350     def test_16_get_tapi_notifications_connectivity_service_Ethernet(self):
 351         self.cr_get_notif_list_input_data["subscription-id-or-name"] = str(self.uuid_subscriptions.eth)
 352         response = test_utils.transportpce_api_rpc_request(
 353             'tapi-notification', 'get-notification-list', self.cr_get_notif_list_input_data)
 354         self.assertEqual(response['status_code'], requests.codes.ok)
 355         self.assertEqual(response['output']['notification'][0]['target-object-identifier'], str(self.uuid_services.eth))
 356         self.assertEqual(response['output']['notification'][0]['target-object-type'], 'CONNECTIVITY_SERVICE')
 357         self.assertEqual(response['output']['notification'][0]['changed-attributes'][0]['new-value'], 'LOCKED')
 358         self.assertEqual(response['output']['notification'][0]['changed-attributes'][1]['new-value'], 'DISABLED')
 359         time.sleep(2)
 360
 361     def test_17_restore_status_port_roadma_srg(self):
 362         url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
 363         body = {"ports": [{
 364             "port-name": "C1",
 365             "logical-connection-point": "SRG1-PP1",
 366             "port-type": "client",
 367             "circuit-id": "SRG1",
 368             "administrative-state": "inService",
 369             "port-qual": "roadm-external"}]}
 370         response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
 371                                     data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
 372                                     auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
 373                                     timeout=test_utils.REQUEST_TIMEOUT)
 374         self.assertEqual(response.status_code, requests.codes.ok)
 375         # If the gate fails is because of the waiting time not being enough
 376         time.sleep(2)
 377
 378     def test_18_get_tapi_notifications_connectivity_service_Ethernet(self):
 379         self.cr_get_notif_list_input_data["subscription-id-or-name"] = str(self.uuid_subscriptions.eth)
 380         response = test_utils.transportpce_api_rpc_request(
 381             'tapi-notification', 'get-notification-list', self.cr_get_notif_list_input_data)
 382         self.assertEqual(response['status_code'], requests.codes.ok)
 383         self.assertEqual(response['output']['notification'][0]['target-object-identifier'], str(self.uuid_services.eth))
 384         self.assertEqual(response['output']['notification'][0]['target-object-type'], 'CONNECTIVITY_SERVICE')
 385         self.assertEqual(response['output']['notification'][0]['changed-attributes'][0]['new-value'], 'UNLOCKED')
 386         self.assertEqual(response['output']['notification'][0]['changed-attributes'][1]['new-value'], 'ENABLED')
 387         time.sleep(2)
 388
 389     def test_19_delete_connectivity_service_Ethernet(self):
 390         self.tapi_serv_details["service-id-or-name"] = str(self.uuid_services.eth)
 391         response = test_utils.transportpce_api_rpc_request(
 392             'tapi-connectivity', 'delete-connectivity-service', self.tapi_serv_details)
 393         self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content))
 394         time.sleep(self.WAITING)
 395
 396     def test_20_disconnect_XPDRA(self):
 397         response = test_utils.unmount_device("XPDR-A1")
 398         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 399
 400     def test_21_disconnect_XPDRC(self):
 401         response = test_utils.unmount_device("XPDR-C1")
 402         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 403
 404     def test_22_disconnect_ROADMA(self):
 405         response = test_utils.unmount_device("ROADM-A1")
 406         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 407
 408     def test_23_disconnect_ROADMC(self):
 409         response = test_utils.unmount_device("ROADM-C1")
 410         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 411
 412
 413 if __name__ == "__main__":
 414     unittest.main(verbosity=2)