tests/transportpce_tests/with_docker/test02_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 class TransportNbiNotificationstesting(unittest.TestCase):
  28     processes = []
  29     cr_serv_input_data = {
  30         "sdnc-request-header": {
  31             "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
  32             "rpc-action": "service-create",
  33             "request-system-id": "appname",
  34             "notification-url": "http://localhost:8585/NotificationServer/notify"
  35         },
  36         "service-name": "service1",
  37         "common-id": "ASATT1234567",
  38         "connection-type": "service",
  39         "service-a-end": {
  40             "service-rate": "100",
  41             "node-id": "XPDR-A1",
  42             "service-format": "Ethernet",
  43             "clli": "SNJSCAMCJP8",
  44             "tx-direction": [{"index": 0}],
  45             "rx-direction": [{"index": 0}],
  46             "optic-type": "gray"
  47         },
  48         "service-z-end": {
  49             "service-rate": "100",
  50             "node-id": "XPDR-C1",
  51             "service-format": "Ethernet",
  52             "clli": "SNJSCAMCJT4",
  53             "tx-direction": [{"index": 0}],
  54             "rx-direction": [{"index": 0}],
  55             "optic-type": "gray"
  56         },
  57         "due-date": "2016-11-28T00:00:01Z",
  58         "operator-contact": "pw1234"
  59     }
  60
  61     del_serv_input_data = {
  62         "sdnc-request-header": {
  63             "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
  64             "rpc-action": "service-delete",
  65             "request-system-id": "appname",
  66             "notification-url": "http://localhost:8585/NotificationServer/notify"},
  67         "service-delete-req-info": {
  68             "service-name": "TBD",
  69             "tail-retention": "no"}
  70     }
  71
  72     nbi_notif_input_data = {
  73         "connection-type": "service",
  74         "id-consumer": "consumer",
  75         "group-id": "transportpceTest"
  76     }
  77
  78     WAITING = 20  # nominal value is 300
  79     NODE_VERSION = '2.2.1'
  80
  81     @classmethod
  82     def setUpClass(cls):
  83         # pylint: disable=unsubscriptable-object
  84         # TODO: for lighty manage the activation of NBI notification feature
  85         cls.init_failed = False
  86         cls.processes = test_utils.start_tpce()
  87         # NBI notification feature is not installed by default in Karaf
  88         if "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
  89             print("installing NBI notification feature...")
  90             result = test_utils.install_karaf_feature("odl-transportpce-nbinotifications")
  91             if result.returncode != 0:
  92                 cls.init_failed = True
  93         if cls.init_failed:
  94             print("NBI notification installation feature failed...")
  95             test_utils.shutdown_process(cls.processes[0])
  96             sys.exit(2)
  97         cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION),
  98                                                ('roadma', cls.NODE_VERSION),
  99                                                ('roadmc', cls.NODE_VERSION),
 100                                                ('xpdrc', cls.NODE_VERSION)])
 101
 102     @classmethod
 103     def tearDownClass(cls):
 104         # pylint: disable=not-an-iterable
 105         for process in cls.processes:
 106             test_utils.shutdown_process(process)
 107         print("all processes killed")
 108
 109     def setUp(self):  # instruction executed before each test method
 110         # pylint: disable=consider-using-f-string
 111         print("execution of {}".format(self.id().split(".")[-1]))
 112
 113     def test_01_connect_xpdrA(self):
 114         response = test_utils.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION))
 115         self.assertEqual(response.status_code,
 116                          requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 117
 118     def test_02_connect_xpdrC(self):
 119         response = test_utils.mount_device("XPDR-C1", ('xpdrc', self.NODE_VERSION))
 120         self.assertEqual(response.status_code,
 121                          requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 122
 123     def test_03_connect_rdmA(self):
 124         response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION))
 125         self.assertEqual(response.status_code,
 126                          requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 127
 128     def test_04_connect_rdmC(self):
 129         response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION))
 130         self.assertEqual(response.status_code,
 131                          requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 132
 133     def test_05_connect_xprdA_N1_to_roadmA_PP1(self):
 134         response = test_utils.transportpce_api_rpc_request(
 135             'transportpce-networkutils', 'init-xpdr-rdm-links',
 136             {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1',
 137                              'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 138         self.assertEqual(response['status_code'], requests.codes.ok)
 139         self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
 140         time.sleep(2)
 141
 142     def test_06_connect_roadmA_PP1_to_xpdrA_N1(self):
 143         response = test_utils.transportpce_api_rpc_request(
 144             'transportpce-networkutils', 'init-rdm-xpdr-links',
 145             {'links-input': {'xpdr-node': 'XPDR-A1', 'xpdr-num': '1', 'network-num': '1',
 146                              'rdm-node': 'ROADM-A1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 147         self.assertEqual(response['status_code'], requests.codes.ok)
 148         self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
 149         time.sleep(2)
 150
 151     def test_07_connect_xprdC_N1_to_roadmC_PP1(self):
 152         response = test_utils.transportpce_api_rpc_request(
 153             'transportpce-networkutils', 'init-xpdr-rdm-links',
 154             {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1',
 155                              'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 156         self.assertEqual(response['status_code'], requests.codes.ok)
 157         self.assertIn('Xponder Roadm Link created successfully', response["output"]["result"])
 158         time.sleep(2)
 159
 160     def test_08_connect_roadmC_PP1_to_xpdrC_N1(self):
 161         response = test_utils.transportpce_api_rpc_request(
 162             'transportpce-networkutils', 'init-rdm-xpdr-links',
 163             {'links-input': {'xpdr-node': 'XPDR-C1', 'xpdr-num': '1', 'network-num': '1',
 164                              'rdm-node': 'ROADM-C1', 'srg-num': '1', 'termination-point-num': 'SRG1-PP1-TXRX'}})
 165         self.assertEqual(response['status_code'], requests.codes.ok)
 166         self.assertIn('Roadm Xponder links created successfully', response["output"]["result"])
 167         time.sleep(2)
 168
 169     def test_09_get_notifications_service1(self):
 170         response = test_utils.transportpce_api_rpc_request(
 171             'nbi-notifications', 'get-notifications-process-service', self.nbi_notif_input_data)
 172         self.assertEqual(response['status_code'], requests.codes.no_content)
 173         time.sleep(2)
 174
 175     def test_10_create_eth_service1(self):
 176         self.cr_serv_input_data["service-name"] = "service1"
 177         response = test_utils.transportpce_api_rpc_request(
 178             'org-openroadm-service', 'service-create', self.cr_serv_input_data)
 179         self.assertEqual(response['status_code'], requests.codes.ok)
 180         self.assertIn('PCE calculation in progress',
 181                       response['output']['configuration-response-common']['response-message'])
 182         time.sleep(self.WAITING)
 183
 184     def test_11_get_notifications_service1(self):
 185         response = test_utils.transportpce_api_rpc_request(
 186             'nbi-notifications', 'get-notifications-process-service', self.nbi_notif_input_data)
 187         self.assertEqual(response['status_code'], requests.codes.ok)
 188         self.assertEqual(response['output']['notifications-process-service'][-1]['service-name'], 'service1')
 189         self.assertEqual(response['output']['notifications-process-service'][-1]['connection-type'], 'service')
 190         self.assertEqual(response['output']['notifications-process-service'][-1]['message'],
 191                          'ServiceCreate request failed ...')
 192         self.assertEqual(response['output']['notifications-process-service'][-1]['response-failed'],
 193                          'PCE path computation failed !')
 194         time.sleep(2)
 195
 196     def test_12_add_omsAttributes_ROADMA_ROADMC(self):
 197         # Config ROADMA-ROADMC oms-attributes
 198         data = {"span": {
 199             "auto-spanloss": "true",
 200             "spanloss-base": 11.4,
 201             "spanloss-current": 12,
 202             "engineered-spanloss": 12.2,
 203             "link-concatenation": [{
 204                 "SRLG-Id": 0,
 205                 "fiber-type": "smf",
 206                 "SRLG-length": 100000,
 207                 "pmd": 0.5}]}}
 208         response = test_utils.add_oms_attr_request(
 209             "ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", data)
 210         self.assertEqual(response.status_code, requests.codes.created)
 211
 212     def test_13_add_omsAttributes_ROADMC_ROADMA(self):
 213         # Config ROADMC-ROADMA oms-attributes
 214         data = {"span": {
 215             "auto-spanloss": "true",
 216             "spanloss-base": 11.4,
 217             "spanloss-current": 12,
 218             "engineered-spanloss": 12.2,
 219             "link-concatenation": [{
 220                 "SRLG-Id": 0,
 221                 "fiber-type": "smf",
 222                 "SRLG-length": 100000,
 223                 "pmd": 0.5}]}}
 224         response = test_utils.add_oms_attr_request(
 225             "ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX", data)
 226         self.assertEqual(response.status_code, requests.codes.created)
 227
 228     # test service-create for Eth service from xpdr to xpdr
 229     def test_14_create_eth_service1(self):
 230         self.cr_serv_input_data["service-name"] = "service1"
 231         response = test_utils.transportpce_api_rpc_request(
 232             'org-openroadm-service', 'service-create', self.cr_serv_input_data)
 233         self.assertEqual(response['status_code'], requests.codes.ok)
 234         self.assertIn('PCE calculation in progress',
 235                       response['output']['configuration-response-common']['response-message'])
 236         time.sleep(self.WAITING)
 237
 238     def test_15_get_eth_service1(self):
 239         response = test_utils.get_ordm_serv_list_attr_request("services", "service1")
 240         self.assertEqual(response['status_code'], requests.codes.ok)
 241         self.assertEqual(response['services'][0]['administrative-state'], 'inService')
 242         self.assertEqual(response['services'][0]['service-name'], 'service1')
 243         self.assertEqual(response['services'][0]['connection-type'], 'service')
 244         self.assertEqual(response['services'][0]['lifecycle-state'], 'planned')
 245
 246     def test_16_get_notifications_service1(self):
 247         response = test_utils.transportpce_api_rpc_request(
 248             'nbi-notifications', 'get-notifications-process-service', self.nbi_notif_input_data)
 249         self.assertEqual(response['status_code'], requests.codes.ok)
 250         self.assertEqual(response['output']['notifications-process-service'][-1]['service-name'], 'service1')
 251         self.assertEqual(response['output']['notifications-process-service'][-1]['connection-type'], 'service')
 252         self.assertEqual(response['output']['notifications-process-service'][-1]['message'], 'Service implemented !')
 253
 254     def test_17_get_notifications_alarm_service1(self):
 255         response = test_utils.transportpce_api_rpc_request(
 256             'nbi-notifications', 'get-notifications-alarm-service', self.nbi_notif_input_data)
 257         self.assertEqual(response['status_code'], requests.codes.ok)
 258         self.assertEqual(response['output']['notifications-alarm-service'][-1]['service-name'], 'service1')
 259         self.assertEqual(response['output']['notifications-alarm-service'][-1]['connection-type'], 'service')
 260         self.assertEqual(response['output']['notifications-alarm-service'][-1]['operational-state'], 'inService')
 261         self.assertEqual(response['output']['notifications-alarm-service'][-1]['message'],
 262                          'The service is now inService')
 263
 264     def test_18_change_status_port_roadma_srg(self):
 265         url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
 266         body = {"ports": [{
 267             "port-name": "C1",
 268             "logical-connection-point": "SRG1-PP1",
 269             "port-type": "client",
 270             "circuit-id": "SRG1",
 271             "administrative-state": "outOfService",
 272             "port-qual": "roadm-external"}]}
 273         response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
 274                                     data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
 275                                     auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
 276                                     timeout=test_utils.REQUEST_TIMEOUT)
 277         self.assertEqual(response.status_code, requests.codes.ok)
 278         time.sleep(2)
 279
 280     def test_19_get_notifications_alarm_service1(self):
 281         response = test_utils.transportpce_api_rpc_request(
 282             'nbi-notifications', 'get-notifications-alarm-service', self.nbi_notif_input_data)
 283         self.assertEqual(response['status_code'], requests.codes.ok)
 284         self.assertEqual(response['output']['notifications-alarm-service'][-1]['service-name'], 'service1')
 285         self.assertEqual(response['output']['notifications-alarm-service'][-1]['connection-type'], 'service')
 286         self.assertEqual(response['output']['notifications-alarm-service'][-1]['operational-state'], 'outOfService')
 287         self.assertEqual(response['output']['notifications-alarm-service'][-1]['message'],
 288                          'The service is now outOfService')
 289
 290     def test_20_restore_status_port_roadma_srg(self):
 291         url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
 292         body = {"ports": [{
 293             "port-name": "C1",
 294             "logical-connection-point": "SRG1-PP1",
 295             "port-type": "client",
 296             "circuit-id": "SRG1",
 297             "administrative-state": "inService",
 298             "port-qual": "roadm-external"}]}
 299         response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
 300                                     data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
 301                                     auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
 302                                     timeout=test_utils.REQUEST_TIMEOUT)
 303         self.assertEqual(response.status_code, requests.codes.ok)
 304         time.sleep(2)
 305
 306     def test_21_get_notifications_alarm_service1(self):
 307         self.test_17_get_notifications_alarm_service1()
 308
 309     def test_22_delete_eth_service1(self):
 310         self.del_serv_input_data["service-delete-req-info"]["service-name"] = "service1"
 311         response = test_utils.transportpce_api_rpc_request(
 312             'org-openroadm-service', 'service-delete', self.del_serv_input_data)
 313         self.assertEqual(response['status_code'], requests.codes.ok)
 314         self.assertIn('Renderer service delete in progress',
 315                       response['output']['configuration-response-common']['response-message'])
 316         time.sleep(20)
 317
 318     def test_23_get_notifications_service1(self):
 319         response = test_utils.transportpce_api_rpc_request(
 320             'nbi-notifications', 'get-notifications-process-service', self.nbi_notif_input_data)
 321         self.assertEqual(response['status_code'], requests.codes.ok)
 322         self.assertEqual(response['output']['notifications-process-service'][-1]['service-name'], 'service1')
 323         self.assertEqual(response['output']['notifications-process-service'][-1]['connection-type'], 'service')
 324         self.assertEqual(response['output']['notifications-process-service'][-1]['message'], 'Service deleted !')
 325         time.sleep(2)
 326
 327     def test_24_disconnect_XPDRA(self):
 328         response = test_utils.unmount_device("XPDR-A1")
 329         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 330
 331     def test_25_disconnect_XPDRC(self):
 332         response = test_utils.unmount_device("XPDR-C1")
 333         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 334
 335     def test_26_disconnect_ROADMA(self):
 336         response = test_utils.unmount_device("ROADM-A1")
 337         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 338
 339     def test_27_disconnect_ROADMC(self):
 340         response = test_utils.unmount_device("ROADM-C1")
 341         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 342
 343
 344 if __name__ == "__main__":
 345     unittest.main(verbosity=2)