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_sample_data = {
  49         "input": {
  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-layer": "ETH",
  90                 "service-type": "POINT_TO_POINT_CONNECTIVITY",
  91                 "service-level": "Some service-level",
  92                 "requested-capacity": {
  93                     "total-size": {
  94                         "value": "100",
  95                         "unit": "GB"
  96                     }
  97                 }
  98             },
  99             "state": "Some state"
 100         }
 101     }
 102
 103     cr_notif_subs_sample_data = {
 104         "input": {
 105             "subscription-filter": {
 106                 "requested-notification-types": [
 107                     "ALARM_EVENT"
 108                 ],
 109                 "requested-object-types": [
 110                     "CONNECTIVITY_SERVICE"
 111                 ],
 112                 "requested-layer-protocols": [
 113                     "ETH"
 114                 ],
 115                 "requested-object-identifier": [
 116                     "76d8f07b-ead5-4132-8eb8-cf3fdef7e079"
 117                 ],
 118                 "include-content": True,
 119                 "local-id": "localId",
 120                 "name": [
 121                     {
 122                         "value-name": "Subscription name",
 123                         "value": "test subscription"
 124                     }
 125                 ]
 126             },
 127             "subscription-state": "ACTIVE"
 128         }
 129     }
 130
 131     cr_get_notif_list_sample_data = {
 132         "input": {
 133             "subscription-id-or-name": "c07e7fd1-0377-4fbf-8928-36c17b0d0d68",
 134             "time-period": "time-period"
 135         }
 136     }
 137
 138     processes = []
 139     uuid_services = UuidServices()
 140     uuid_subscriptions = UuidSubscriptions()
 141     WAITING = 25  # nominal value is 300
 142     NODE_VERSION_221 = '2.2.1'
 143
 144     @classmethod
 145     def setUpClass(cls):
 146         # pylint: disable=unsubscriptable-object
 147         # TODO: for lighty manage the activation of NBI notification feature
 148         cls.init_failed_nbi = False
 149         cls.init_failed_tapi = False
 150         os.environ['JAVA_MIN_MEM'] = '1024M'
 151         os.environ['JAVA_MAX_MEM'] = '4096M'
 152         cls.processes = test_utils.start_tpce()
 153         # NBI notification feature is not installed by default in Karaf
 154         if "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
 155             print("installing NBI notification feature...")
 156             result = test_utils.install_karaf_feature("odl-transportpce-nbinotifications")
 157             if result.returncode != 0:
 158                 cls.init_failed_nbi = True
 159             print("installing tapi feature...")
 160             result = test_utils.install_karaf_feature("odl-transportpce-tapi")
 161             if result.returncode != 0:
 162                 cls.init_failed_tapi = True
 163             print("Restarting OpenDaylight...")
 164             test_utils.shutdown_process(cls.processes[0])
 165             cls.processes[0] = test_utils.start_karaf()
 166             test_utils.process_list[0] = cls.processes[0]
 167             cls.init_failed = not test_utils.wait_until_log_contains(
 168                 test_utils.KARAF_LOG, test_utils.KARAF_OK_START_MSG, time_to_wait=60)
 169         if cls.init_failed_nbi:
 170             print("NBI notification installation feature failed...")
 171             test_utils.shutdown_process(cls.processes[0])
 172             sys.exit(2)
 173         if cls.init_failed_tapi:
 174             print("tapi installation feature failed...")
 175             test_utils.shutdown_process(cls.processes[0])
 176             sys.exit(2)
 177         cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION_221),
 178                                                ('roadma', cls.NODE_VERSION_221),
 179                                                ('roadmc', cls.NODE_VERSION_221),
 180                                                ('xpdrc', cls.NODE_VERSION_221)])
 181
 182     @classmethod
 183     def tearDownClass(cls):
 184         # pylint: disable=not-an-iterable
 185         for process in cls.processes:
 186             test_utils.shutdown_process(process)
 187         print("all processes killed")
 188
 189     def setUp(self):  # instruction executed before each test method
 190         # pylint: disable=consider-using-f-string
 191         print("execution of {}".format(self.id().split(".")[-1]))
 192
 193     def test_01_connect_xpdrA(self):
 194         response = test_utils.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION_221))
 195         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 196
 197     def test_02_connect_xpdrC(self):
 198         response = test_utils.mount_device("XPDR-C1", ('xpdrc', self.NODE_VERSION_221))
 199         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 200
 201     def test_03_connect_rdmA(self):
 202         response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION_221))
 203         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 204
 205     def test_04_connect_rdmC(self):
 206         response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION_221))
 207         self.assertEqual(response.status_code, 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.connect_xpdr_to_rdm_request("XPDR-A1", "1", "1",
 211                                                           "ROADM-A1", "1", "SRG1-PP1-TXRX")
 212         self.assertEqual(response.status_code, requests.codes.ok)
 213         res = response.json()
 214         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
 215         time.sleep(2)
 216
 217     def test_06_connect_roadmA_PP1_to_xpdrA_N1(self):
 218         response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "1",
 219                                                           "ROADM-A1", "1", "SRG1-PP1-TXRX")
 220         self.assertEqual(response.status_code, requests.codes.ok)
 221         res = response.json()
 222         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
 223         time.sleep(2)
 224
 225     def test_07_connect_xprdC_N1_to_roadmC_PP1(self):
 226         response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "1",
 227                                                           "ROADM-C1", "1", "SRG1-PP1-TXRX")
 228         self.assertEqual(response.status_code, requests.codes.ok)
 229         res = response.json()
 230         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
 231         time.sleep(2)
 232
 233     def test_08_connect_roadmC_PP1_to_xpdrC_N1(self):
 234         response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "1",
 235                                                           "ROADM-C1", "1", "SRG1-PP1-TXRX")
 236         self.assertEqual(response.status_code, requests.codes.ok)
 237         res = response.json()
 238         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
 239         time.sleep(2)
 240
 241     def test_09_add_omsAttributes_ROADMA_ROADMC(self):
 242         # Config ROADMA-ROADMC oms-attributes
 243         data = {"span": {
 244             "auto-spanloss": "true",
 245             "spanloss-base": 11.4,
 246             "spanloss-current": 12,
 247             "engineered-spanloss": 12.2,
 248             "link-concatenation": [{
 249                 "SRLG-Id": 0,
 250                 "fiber-type": "smf",
 251                 "SRLG-length": 100000,
 252                 "pmd": 0.5}]}}
 253         response = test_utils.add_oms_attr_request("ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", data)
 254         self.assertEqual(response.status_code, requests.codes.created)
 255
 256     def test_10_add_omsAttributes_ROADMC_ROADMA(self):
 257         # Config ROADMC-ROADMA oms-attributes
 258         data = {"span": {
 259             "auto-spanloss": "true",
 260             "spanloss-base": 11.4,
 261             "spanloss-current": 12,
 262             "engineered-spanloss": 12.2,
 263             "link-concatenation": [{
 264                 "SRLG-Id": 0,
 265                 "fiber-type": "smf",
 266                 "SRLG-length": 100000,
 267                 "pmd": 0.5}]}}
 268         response = test_utils.add_oms_attr_request("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.tapi_create_connectivity_request(self.cr_serv_sample_data)
 274         time.sleep(self.WAITING)
 275         self.assertEqual(response.status_code, requests.codes.ok)
 276         res = response.json()
 277         self.uuid_services.eth = res['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, **res['output']['service']),
 293                              res['output']['service'])
 294         self.assertDictEqual(dict(input_dict_2, **res['output']['service']['end-point'][0]['name'][0]),
 295                              res['output']['service']['end-point'][0]['name'][0])
 296         self.assertDictEqual(dict(input_dict_3, **res['output']['service']['end-point'][1]['name'][0]),
 297                              res['output']['service']['end-point'][1]['name'][0])
 298         # If the gate fails is because of the waiting time not being enough
 299         time.sleep(self.WAITING)
 300
 301     def test_12_get_service_Ethernet(self):
 302         response = test_utils.get_service_list_request("services/" + str(self.uuid_services.eth))
 303         self.assertEqual(response.status_code, requests.codes.ok)
 304         res = response.json()
 305         self.assertEqual(
 306             res['services'][0]['administrative-state'], 'inService')
 307         self.assertEqual(
 308             res['services'][0]['service-name'], self.uuid_services.eth)
 309         self.assertEqual(
 310             res['services'][0]['connection-type'], 'service')
 311         self.assertEqual(
 312             res['services'][0]['lifecycle-state'], 'planned')
 313         time.sleep(2)
 314
 315     def test_13_get_connectivity_service_Ethernet(self):
 316         response = test_utils.tapi_get_connectivity_request(str(self.uuid_services.eth))
 317         self.assertEqual(response.status_code, requests.codes.ok)
 318         res = response.json()
 319         self.assertEqual(
 320             res['output']['service']['operational-state'], 'ENABLED')
 321         self.assertEqual(
 322             res['output']['service']['name'][0]['value'], self.uuid_services.eth)
 323         self.assertEqual(
 324             res['output']['service']['administrative-state'], 'UNLOCKED')
 325         self.assertEqual(
 326             res['output']['service']['lifecycle-state'], 'INSTALLED')
 327         time.sleep(2)
 328
 329     def test_14_create_notifications_subscription_service(self):
 330         self.cr_notif_subs_sample_data["input"]["subscription-filter"]["requested-object-identifier"][0] = str(
 331             self.uuid_services.eth)
 332         response = test_utils.tapi_create_notification_subscription_service_request(self.cr_notif_subs_sample_data)
 333         self.assertEqual(response.status_code, requests.codes.ok)
 334         res = response.json()
 335         self.uuid_subscriptions.eth = res['output']['subscription-service']['uuid']
 336         self.assertEqual(res['output']['subscription-service']['subscription-filter']
 337                          ['requested-object-types'][0], 'CONNECTIVITY_SERVICE')
 338         self.assertEqual(res['output']['subscription-service']['subscription-filter']
 339                          ['requested-notification-types'][0], 'ALARM_EVENT')
 340         self.assertEqual(res['output']['subscription-service']['subscription-filter']
 341                          ['requested-object-identifier'][0], str(self.uuid_services.eth))
 342         time.sleep(2)
 343
 344     def test_15_change_status_port_roadma_srg(self):
 345         url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
 346         body = {"ports": [{
 347             "port-name": "C1",
 348             "logical-connection-point": "SRG1-PP1",
 349             "port-type": "client",
 350             "circuit-id": "SRG1",
 351             "administrative-state": "outOfService",
 352             "port-qual": "roadm-external"}]}
 353         response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
 354                                     data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
 355                                     auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
 356                                     timeout=test_utils.REQUEST_TIMEOUT)
 357         self.assertEqual(response.status_code, requests.codes.ok)
 358         # If the gate fails is because of the waiting time not being enough
 359         time.sleep(2)
 360
 361     def test_16_get_tapi_notifications_connectivity_service_Ethernet(self):
 362         self.cr_get_notif_list_sample_data["input"]["subscription-id-or-name"] = str(self.uuid_subscriptions.eth)
 363         response = test_utils.tapi_get_notifications_list_request(self.cr_get_notif_list_sample_data)
 364         self.assertEqual(response.status_code, requests.codes.ok)
 365         res = response.json()
 366         self.assertEqual(res['output']['notification'][0]['target-object-identifier'], str(self.uuid_services.eth))
 367         self.assertEqual(res['output']['notification'][0]['target-object-type'], 'CONNECTIVITY_SERVICE')
 368         self.assertEqual(res['output']['notification'][0]['changed-attributes'][0]['new-value'], 'LOCKED')
 369         self.assertEqual(res['output']['notification'][0]['changed-attributes'][1]['new-value'], 'DISABLED')
 370         time.sleep(2)
 371
 372     def test_17_restore_status_port_roadma_srg(self):
 373         url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
 374         body = {"ports": [{
 375             "port-name": "C1",
 376             "logical-connection-point": "SRG1-PP1",
 377             "port-type": "client",
 378             "circuit-id": "SRG1",
 379             "administrative-state": "inService",
 380             "port-qual": "roadm-external"}]}
 381         response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
 382                                     data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
 383                                     auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
 384                                     timeout=test_utils.REQUEST_TIMEOUT)
 385         self.assertEqual(response.status_code, requests.codes.ok)
 386         # If the gate fails is because of the waiting time not being enough
 387         time.sleep(2)
 388
 389     def test_18_get_tapi_notifications_connectivity_service_Ethernet(self):
 390         self.cr_get_notif_list_sample_data["input"]["subscription-id-or-name"] = str(self.uuid_subscriptions.eth)
 391         response = test_utils.tapi_get_notifications_list_request(self.cr_get_notif_list_sample_data)
 392         self.assertEqual(response.status_code, requests.codes.ok)
 393         res = response.json()
 394         self.assertEqual(res['output']['notification'][1]['target-object-identifier'], str(self.uuid_services.eth))
 395         self.assertEqual(res['output']['notification'][1]['target-object-type'], 'CONNECTIVITY_SERVICE')
 396         self.assertEqual(res['output']['notification'][1]['changed-attributes'][0]['new-value'], 'UNLOCKED')
 397         self.assertEqual(res['output']['notification'][1]['changed-attributes'][1]['new-value'], 'ENABLED')
 398         time.sleep(2)
 399
 400     def test_19_delete_connectivity_service_Ethernet(self):
 401         response = test_utils.tapi_delete_connectivity_request(str(self.uuid_services.eth))
 402         self.assertEqual(response.status_code, requests.codes.no_content)
 403         time.sleep(self.WAITING)
 404
 405     def test_20_disconnect_XPDRA(self):
 406         response = test_utils.unmount_device("XPDR-A1")
 407         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 408
 409     def test_21_disconnect_XPDRC(self):
 410         response = test_utils.unmount_device("XPDR-C1")
 411         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 412
 413     def test_22_disconnect_ROADMA(self):
 414         response = test_utils.unmount_device("ROADM-A1")
 415         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 416
 417     def test_23_disconnect_ROADMC(self):
 418         response = test_utils.unmount_device("ROADM-C1")
 419         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 420
 421
 422 if __name__ == "__main__":
 423     unittest.main(verbosity=2)