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