osm_mon/test/OpenStack/test_alarming.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
  26 import logging
  27
  28 import unittest
  29
  30 import mock
  31
  32 from plugins.OpenStack.Aodh import alarming as alarm_req
  33 from plugins.OpenStack.common import Common
  34
  35 __author__ = "Helena McGough"
  36
  37 log = logging.getLogger(__name__)
  38
  39 auth_token = mock.ANY
  40 endpoint = mock.ANY
  41
  42
  43 class Response(object):
  44     """Mock a response message class."""
  45
  46     def __init__(self, result):
  47         """Initialise the response text and status code."""
  48         self.text = json.dumps(result)
  49         self.status_code = "MOCK_STATUS_CODE"
  50
  51
  52 class TestAlarming(unittest.TestCase):
  53     """Tests for alarming class functions."""
  54
  55     maxDiff = None
  56
  57     def setUp(self):
  58         """Setup for tests."""
  59         super(TestAlarming, self).setUp()
  60         self.alarming = alarm_req.Alarming()
  61         self.alarming.common = Common()
  62
  63     @mock.patch.object(alarm_req.Alarming, "check_payload")
  64     @mock.patch.object(alarm_req.Alarming, "check_for_metric")
  65     @mock.patch.object(Common, "_perform_request")
  66     def test_config_invalid_alarm_req(self, perf_req, check_metric, check_pay):
  67         """Test configure an invalid alarm request."""
  68         # Configuring with invalid alarm name results in failure
  69         values = {"alarm_name": "my_alarm",
  70                   "metric_name": "my_metric",
  71                   "resource_uuid": "my_r_id"}
  72         self.alarming.configure_alarm(endpoint, auth_token, values)
  73         perf_req.assert_not_called
  74         perf_req.reset_mock()
  75
  76         # Correct alarm_name will check for metric in Gnocchi
  77         # If there isn't one an alarm won;t be created
  78         values = {"alarm_name": "disk_write_ops",
  79                   "metric_name": "disk_write_ops",
  80                   "resource_uuid": "my_r_id"}
  81
  82         check_metric.return_value = None
  83
  84         self.alarming.configure_alarm(endpoint, auth_token, values)
  85         perf_req.assert_not_called
  86
  87     @mock.patch.object(alarm_req.Alarming, "check_payload")
  88     @mock.patch.object(alarm_req.Alarming, "check_for_metric")
  89     @mock.patch.object(Common, "_perform_request")
  90     def test_config_valid_alarm_req(self, perf_req, check_metric, check_pay):
  91         """Test config a valid alarm."""
  92         # Correct alarm_name will check for metric in Gnocchi
  93         # And conform that the payload is configured correctly
  94         values = {"alarm_name": "disk_write_ops",
  95                   "metric_name": "disk_write_ops",
  96                   "resource_uuid": "my_r_id"}
  97
  98         check_metric.return_value = "my_metric_id"
  99         check_pay.return_value = "my_payload"
 100
 101         self.alarming.configure_alarm(endpoint, auth_token, values)
 102         perf_req.assert_called_with(
 103             "<ANY>/v2/alarms/", auth_token,
 104             req_type="post", payload="my_payload")
 105
 106     @mock.patch.object(Common, "_perform_request")
 107     def test_delete_alarm_req(self, perf_req):
 108         """Test delete alarm request."""
 109         self.alarming.delete_alarm(endpoint, auth_token, "my_alarm_id")
 110
 111         perf_req.assert_called_with(
 112             "<ANY>/v2/alarms/my_alarm_id", auth_token, req_type="delete")
 113
 114     @mock.patch.object(Common, "_perform_request")
 115     def test_invalid_list_alarm_req(self, perf_req):
 116         """Test invalid list alarm_req."""
 117         # Request will not be performed with out a resoure_id
 118         list_details = {"mock_details": "invalid_details"}
 119         self.alarming.list_alarms(endpoint, auth_token, list_details)
 120
 121         perf_req.assert_not_called
 122
 123     @mock.patch.object(Common, "_perform_request")
 124     def test_valid_list_alarm_req(self, perf_req):
 125         """Test valid list alarm request."""
 126         # Minimum requirement for an alarm list is resource_id
 127         list_details = {"resource_uuid": "mock_r_id"}
 128         self.alarming.list_alarms(endpoint, auth_token, list_details)
 129
 130         perf_req.assert_called_with(
 131             "<ANY>/v2/alarms/", auth_token, req_type="get")
 132         perf_req.reset_mock()
 133
 134         # Check list with alarm_name defined
 135         list_details = {"resource_uuid": "mock_r_id",
 136                         "alarm_name": "my_alarm",
 137                         "severity": "critical"}
 138         self.alarming.list_alarms(endpoint, auth_token, list_details)
 139
 140         perf_req.assert_called_with(
 141             "<ANY>/v2/alarms/", auth_token, req_type="get")
 142
 143     @mock.patch.object(Common, "_perform_request")
 144     def test_ack_alarm_req(self, perf_req):
 145         """Test update alarm state for acknowledge alarm request."""
 146         self.alarming.update_alarm_state(endpoint, auth_token, "my_alarm_id")
 147
 148         perf_req.assert_called_with(
 149             "<ANY>/v2/alarms/my_alarm_id/state", auth_token, req_type="put",
 150             payload=json.dumps("ok"))
 151
 152     @mock.patch.object(alarm_req.Alarming, "check_payload")
 153     @mock.patch.object(Common, "_perform_request")
 154     def test_update_alarm_invalid(self, perf_req, check_pay):
 155         """Test update alarm with invalid get response."""
 156         values = {"alarm_uuid": "my_alarm_id"}
 157
 158         self.alarming.update_alarm(endpoint, auth_token, values)
 159
 160         perf_req.assert_called_with(mock.ANY, auth_token, req_type="get")
 161         check_pay.assert_not_called
 162
 163     @mock.patch.object(alarm_req.Alarming, "check_payload")
 164     @mock.patch.object(Common, "_perform_request")
 165     def test_update_alarm_invalid_payload(self, perf_req, check_pay):
 166         """Test update alarm with invalid payload."""
 167         resp = Response({"name": "my_alarm",
 168                          "state": "alarm",
 169                          "gnocchi_resources_threshold_rule":
 170                          {"resource_id": "my_resource_id",
 171                           "metric": "my_metric"}})
 172         perf_req.return_value = resp
 173         check_pay.return_value = None
 174         values = {"alarm_uuid": "my_alarm_id"}
 175
 176         self.alarming.update_alarm(endpoint, auth_token, values)
 177
 178         perf_req.assert_called_with(mock.ANY, auth_token, req_type="get")
 179         self.assertEqual(perf_req.call_count, 1)
 180
 181     @mock.patch.object(alarm_req.Alarming, "check_payload")
 182     @mock.patch.object(Common, "_perform_request")
 183     def test_update_alarm_valid(self, perf_req, check_pay):
 184         """Test valid update alarm request."""
 185         resp = Response({"name": "my_alarm",
 186                          "state": "alarm",
 187                          "gnocchi_resources_threshold_rule":
 188                          {"resource_id": "my_resource_id",
 189                           "metric": "my_metric"}})
 190         perf_req.return_value = resp
 191         values = {"alarm_uuid": "my_alarm_id"}
 192
 193         self.alarming.update_alarm(endpoint, auth_token, values)
 194
 195         check_pay.assert_called_with(values, "my_metric", "my_resource_id",
 196                                      "my_alarm", alarm_state="alarm")
 197
 198         self.assertEqual(perf_req.call_count, 2)
 199         # Second call is the update request
 200         perf_req.assert_called_with(
 201             '<ANY>/v2/alarms/my_alarm_id', auth_token,
 202             req_type="put", payload=check_pay.return_value)
 203
 204     def test_check_valid_payload(self):
 205         """Test the check payload function for a valid payload."""
 206         values = {"severity": "warning",
 207                   "statistic": "COUNT",
 208                   "threshold_value": 12,
 209                   "operation": "GT"}
 210         payload = self.alarming.check_payload(
 211             values, "my_metric", "r_id", "alarm_name")
 212
 213         self.assertEqual(
 214             json.loads(payload), {"name": "alarm_name",
 215                                   "gnocchi_resources_threshold_rule":
 216                                   {"resource_id": "r_id",
 217                                    "metric": "my_metric",
 218                                    "comparison_operator": "gt",
 219                                    "aggregation_method": "count",
 220                                    "threshold": 12,
 221                                    "resource_type": "generic"},
 222                                   "severity": "low",
 223                                   "state": "ok",
 224                                   "type": "gnocchi_resources_threshold",
 225                                   "alarm_actions": ["http://localhost:8662"]})
 226
 227     def test_check_valid_state_payload(self):
 228         """Test the check payload function for a valid payload with state."""
 229         values = {"severity": "warning",
 230                   "statistic": "COUNT",
 231                   "threshold_value": 12,
 232                   "operation": "GT"}
 233         payload = self.alarming.check_payload(
 234             values, "my_metric", "r_id", "alarm_name", alarm_state="alarm")
 235
 236         self.assertEqual(
 237             json.loads(payload), {"name": "alarm_name",
 238                                   "gnocchi_resources_threshold_rule":
 239                                   {"resource_id": "r_id",
 240                                    "metric": "my_metric",
 241                                    "comparison_operator": "gt",
 242                                    "aggregation_method": "count",
 243                                    "threshold": 12,
 244                                    "resource_type": "generic"},
 245                                   "severity": "low",
 246                                   "state": "alarm",
 247                                   "type": "gnocchi_resources_threshold",
 248                                   "alarm_actions": ["http://localhost:8662"]})
 249
 250     def test_check_invalid_payload(self):
 251         """Test the check payload function for an invalid payload."""
 252         values = {"alarm_values": "mock_invalid_details"}
 253         payload = self.alarming.check_payload(
 254             values, "my_metric", "r_id", "alarm_name")
 255
 256         self.assertEqual(payload, None)
 257
 258     @mock.patch.object(Common, "_perform_request")
 259     def test_get_alarm_state(self, perf_req):
 260         """Test the get alarm state function."""
 261         self.alarming.get_alarm_state(endpoint, auth_token, "alarm_id")
 262
 263         perf_req.assert_called_with(
 264             "<ANY>/v2/alarms/alarm_id/state", auth_token, req_type="get")
 265
 266     @mock.patch.object(Common, "get_endpoint")
 267     @mock.patch.object(Common, "_perform_request")
 268     def test_check_for_metric(self, perf_req, get_endpoint):
 269         """Test the check for metric function."""
 270         get_endpoint.return_value = "gnocchi_endpoint"
 271
 272         self.alarming.check_for_metric(auth_token, "metric_name", "r_id")
 273
 274         perf_req.assert_called_with(
 275             "gnocchi_endpoint/v1/metric/", auth_token, req_type="get")