osm_mon/tests/plugins/OpenStack/unit/test_alarm_req.py

   1 # Copyright 2017 iIntel Research and Development Ireland Limited
   2 # **************************************************************
   3
   4 # This file is part of OSM Monitoring module
   5 # All Rights Reserved to Intel Corporation
   6
   7 # Licensed under the Apache License, Version 2.0 (the 'License'); you may
   8 # not use this file except in compliance with the License. You may obtain
   9 # a copy of the License at
  10
  11 #         http://www.apache.org/licenses/LICENSE-2.0
  12
  13 # Unless required by applicable law or agreed to in writing, software
  14 # distributed under the License is distributed on an 'AS IS' BASIS, WITHOUT
  15 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  16 # License for the specific language governing permissions and limitations
  17 # under the License.
  18
  19 # For those usages not covered by the Apache License, Version 2.0 please
  20 # contact: helena.mcgough@intel.com or adrian.hoban@intel.com
  21 ##
  22 """Tests for all alarm request message keys."""
  23
  24 import json
  25 import logging
  26 import unittest
  27 from io import UnsupportedOperation
  28
  29 import mock
  30
  31 from osm_mon.core.auth import AuthManager
  32 from osm_mon.core.database import VimCredentials, DatabaseManager
  33 from osm_mon.plugins.OpenStack.Aodh import alarm_handler as alarm_req
  34 from osm_mon.plugins.OpenStack.Aodh.alarm_handler import OpenstackAlarmHandler
  35 from osm_mon.plugins.OpenStack.common import Common
  36
  37 log = logging.getLogger(__name__)
  38
  39 mock_creds = VimCredentials()
  40 mock_creds.config = '{}'
  41
  42
  43 class Message(object):
  44     """A class to mock a message object value for alarm requests."""
  45
  46     def __init__(self):
  47         """Initialize a mocked message instance."""
  48         self.topic = 'alarm_request'
  49         self.key = None
  50         self.value = json.dumps({'mock_value': 'mock_details'})
  51
  52
  53 class TestAlarmKeys(unittest.TestCase):
  54     """Integration test for alarm request keys."""
  55
  56     def setUp(self):
  57         """Setup the tests for alarm request keys."""
  58         super(TestAlarmKeys, self).setUp()
  59         self.alarming = alarm_req.OpenstackAlarmHandler()
  60         self.alarming.common = Common()
  61
  62     @mock.patch.object(AuthManager, 'get_credentials')
  63     @mock.patch.object(Common, 'get_endpoint')
  64     @mock.patch.object(Common, 'get_auth_token')
  65     def test_alarming_authentication(self, get_token, get_endpoint, get_creds):
  66         """Test getting an auth_token and endpoint for alarm requests."""
  67         # if auth_token is None environment variables are used to authenticate
  68         get_creds.return_value = mock_creds
  69
  70         with self.assertRaises(UnsupportedOperation):
  71             self.alarming.handle_message('', {}, 'test_id')
  72
  73         get_token.assert_called_with('test_id', verify_ssl=True)
  74         get_endpoint.assert_any_call('alarming', 'test_id', verify_ssl=True)
  75
  76     @mock.patch.object(Common, 'get_endpoint', mock.Mock())
  77     @mock.patch.object(Common, 'get_auth_token', mock.Mock())
  78     @mock.patch.object(AuthManager, 'get_credentials')
  79     @mock.patch.object(alarm_req.OpenstackAlarmHandler, 'delete_alarm')
  80     def test_delete_alarm_key(self, del_alarm, get_creds):
  81         """Test the functionality for a create alarm request."""
  82         value = {'alarm_delete_request': {
  83             'correlation_id': 1,
  84             'alarm_uuid': 'my_alarm_id'
  85         }}
  86
  87         get_creds.return_value = mock_creds
  88         del_alarm.return_value = {}
  89
  90         # Call the alarming functionality and check delete request
  91         self.alarming.handle_message('delete_alarm_request', value, 'test_id')
  92         del_alarm.assert_called_with(mock.ANY, mock.ANY, 'my_alarm_id', True)
  93
  94     @mock.patch.object(Common, 'get_endpoint', mock.Mock())
  95     @mock.patch.object(Common, 'get_auth_token', mock.Mock())
  96     @mock.patch.object(AuthManager, 'get_credentials')
  97     @mock.patch.object(alarm_req.OpenstackAlarmHandler, 'list_alarms')
  98     def test_list_alarm_key(self, list_alarm, get_creds):
  99         """Test the functionality for a list alarm request."""
 100         value = {'alarm_list_request': {'correlation_id': 1}}
 101
 102         get_creds.return_value = mock_creds
 103
 104         list_alarm.return_value = []
 105
 106         # Call the alarming functionality and check list functionality
 107         self.alarming.handle_message('list_alarm_request', value, 'test_id')
 108         list_alarm.assert_called_with(mock.ANY, mock.ANY, {'correlation_id': 1}, True)
 109
 110     @mock.patch.object(Common, 'get_auth_token', mock.Mock())
 111     @mock.patch.object(Common, 'get_endpoint', mock.Mock())
 112     @mock.patch.object(AuthManager, 'get_credentials')
 113     @mock.patch.object(alarm_req.OpenstackAlarmHandler, 'update_alarm_state')
 114     def test_ack_alarm_key(self, ack_alarm, get_creds):
 115         """Test the functionality for an acknowledge alarm request."""
 116         value = {'ack_details': {'alarm_uuid': 'my_alarm_id'}}
 117
 118         get_creds.return_value = mock_creds
 119
 120         # Call alarming functionality and check acknowledge functionality
 121         self.alarming.handle_message('acknowledge_alarm_request', value, 'test_id')
 122         ack_alarm.assert_called_with(mock.ANY, mock.ANY, 'my_alarm_id', True)
 123
 124     @mock.patch.object(Common, 'get_auth_token', mock.Mock())
 125     @mock.patch.object(Common, 'get_endpoint', mock.Mock())
 126     @mock.patch.object(DatabaseManager, 'save_alarm', mock.Mock())
 127     @mock.patch.object(Common, "perform_request")
 128     @mock.patch.object(AuthManager, 'get_credentials')
 129     @mock.patch.object(alarm_req.OpenstackAlarmHandler, 'configure_alarm')
 130     def test_config_alarm_key(self, config_alarm, get_creds, perf_req):
 131         """Test the functionality for a create alarm request."""
 132         value = {'alarm_create_request': {'correlation_id': 1, 'threshold_value': 50,
 133                                           'operation': 'GT', 'metric_name': 'cpu_utilization',
 134                                           'vdu_name': 'vdu',
 135                                           'vnf_member_index': '1',
 136                                           'ns_id': '1',
 137                                           'resource_uuid': '123'}}
 138         mock_perf_req_return_value = {"metrics": {"cpu_util": 123}}
 139         perf_req.return_value = type('obj', (object,), {'text': json.dumps(mock_perf_req_return_value, sort_keys=True)})
 140         get_creds.return_value = mock_creds
 141
 142         # Call alarming functionality and check config alarm call
 143         config_alarm.return_value = 'my_alarm_id'
 144         self.alarming.handle_message('create_alarm_request', value, 'test_id')
 145         config_alarm.assert_called_with(mock.ANY, mock.ANY, {'correlation_id': 1, 'threshold_value': 50,
 146                                                              'operation': 'GT',
 147                                                              'metric_name': 'cpu_utilization',
 148                                                              'vdu_name': 'vdu',
 149                                                              'vnf_member_index': '1', 'ns_id': '1',
 150                                                              'resource_uuid': '123'}, {}, True)