c45c05201775abc8b72d031d009ccf4db10d42b6
1 # Copyright 2017 iIntel Research and Development Ireland Limited
2 # **************************************************************
4 # This file is part of OSM Monitoring module
5 # All Rights Reserved to Intel Corporation
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
11 # http://www.apache.org/licenses/LICENSE-2.0
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
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
22 """Tests for all alarm request message keys."""
30 from osm_mon
.core
.settings
import Config
31 from osm_mon
.plugins
.OpenStack
.Aodh
import alarming
as alarm_req
32 from osm_mon
.plugins
.OpenStack
.common
import Common
34 log
= logging
.getLogger(__name__
)
37 alarm_endpoint
= "alarm_endpoint"
38 metric_endpoint
= "metric_endpoint"
41 class Response(object):
42 """Mock a response message class."""
44 def __init__(self
, result
):
45 """Initialise the response text and status code."""
46 self
.text
= json
.dumps(result
)
47 self
.status_code
= "MOCK_STATUS_CODE"
50 class TestAlarming(unittest
.TestCase
):
51 """Tests for alarming class functions."""
56 """Setup for tests."""
57 super(TestAlarming
, self
).setUp()
58 self
.alarming
= alarm_req
.Alarming()
60 @mock.patch
.object(alarm_req
.Alarming
, "check_payload")
61 @mock.patch
.object(alarm_req
.Alarming
, "check_for_metric")
62 @mock.patch
.object(Common
, "perform_request")
63 def test_config_invalid_alarm_req(self
, perf_req
, check_metric
, check_pay
):
64 """Test configure an invalid alarm request."""
65 # Configuring with invalid alarm name results in failure
66 values
= {"alarm_name": "my_alarm",
67 "metric_name": "my_metric",
68 "resource_uuid": "my_r_id"}
69 self
.alarming
.configure_alarm(alarm_endpoint
, metric_endpoint
, auth_token
, values
, {})
70 perf_req
.assert_not_called()
73 # Correct alarm_name will check for metric in Gnocchi
74 # If there isn't one an alarm won;t be created
75 values
= {"alarm_name": "disk_write_ops",
76 "metric_name": "disk_write_ops",
77 "resource_uuid": "my_r_id"}
79 check_metric
.return_value
= None
81 self
.alarming
.configure_alarm(alarm_endpoint
, metric_endpoint
, auth_token
, values
, {})
82 perf_req
.assert_not_called()
84 @mock.patch
.object(alarm_req
.Alarming
, "check_payload")
85 @mock.patch
.object(alarm_req
.Alarming
, "check_for_metric")
86 @mock.patch
.object(Common
, "perform_request")
87 def test_config_valid_alarm_req(self
, perf_req
, check_metric
, check_pay
):
88 """Test config a valid alarm."""
89 # Correct alarm_name will check for metric in Gnocchi
90 # And conform that the payload is configured correctly
91 values
= {"alarm_name": "disk_write_ops",
92 "metric_name": "disk_write_ops",
93 "resource_uuid": "my_r_id"}
95 check_metric
.return_value
= "my_metric_id"
96 check_pay
.return_value
= "my_payload"
98 perf_req
.return_value
= type('obj', (object,), {'text': '{"alarm_id":"1"}'})
100 self
.alarming
.configure_alarm(alarm_endpoint
, metric_endpoint
, auth_token
, values
, {})
101 perf_req
.assert_called_with(
102 "alarm_endpoint/v2/alarms/", auth_token
,
103 req_type
="post", payload
="my_payload")
105 @mock.patch
.object(Common
, "perform_request")
106 def test_delete_alarm_req(self
, perf_req
):
107 """Test delete alarm request."""
108 self
.alarming
.delete_alarm(alarm_endpoint
, auth_token
, "my_alarm_id")
110 perf_req
.assert_called_with(
111 "alarm_endpoint/v2/alarms/my_alarm_id", auth_token
, req_type
="delete")
113 @mock.patch
.object(Common
, "perform_request")
114 def test_invalid_list_alarm_req(self
, perf_req
):
115 """Test invalid list alarm_req."""
116 # Request will not be performed with out a resoure_id
117 list_details
= {"mock_details": "invalid_details"}
118 self
.alarming
.list_alarms(alarm_endpoint
, auth_token
, list_details
)
120 perf_req
.assert_not_called()
122 @mock.patch
.object(Common
, "perform_request")
123 def test_valid_list_alarm_req(self
, perf_req
):
124 """Test valid list alarm request."""
125 # Minimum requirement for an alarm list is resource_id
126 list_details
= {"resource_uuid": "mock_r_id"}
127 self
.alarming
.list_alarms(alarm_endpoint
, auth_token
, list_details
)
129 perf_req
.assert_called_with(
130 "alarm_endpoint/v2/alarms/", auth_token
, req_type
="get")
131 perf_req
.reset_mock()
133 # Check list with alarm_name defined
134 list_details
= {"resource_uuid": "mock_r_id",
135 "alarm_name": "my_alarm",
136 "severity": "critical"}
137 self
.alarming
.list_alarms(alarm_endpoint
, auth_token
, list_details
)
139 perf_req
.assert_called_with(
140 "alarm_endpoint/v2/alarms/", auth_token
, req_type
="get")
142 @mock.patch
.object(Common
, "perform_request")
143 def test_ack_alarm_req(self
, perf_req
):
144 """Test update alarm state for acknowledge alarm request."""
145 self
.alarming
.update_alarm_state(alarm_endpoint
, auth_token
, "my_alarm_id")
147 perf_req
.assert_called_with(
148 "alarm_endpoint/v2/alarms/my_alarm_id/state", auth_token
, req_type
="put",
149 payload
=json
.dumps("ok"))
151 @mock.patch
.object(alarm_req
.Alarming
, "check_payload")
152 @mock.patch
.object(Common
, "perform_request")
153 def test_update_alarm_invalid(self
, perf_req
, check_pay
):
154 """Test update alarm with invalid get response."""
155 values
= {"alarm_uuid": "my_alarm_id"}
157 perf_req
.return_value
= type('obj', (object,), {'invalid_prop': 'Invalid response'})
159 self
.alarming
.update_alarm(alarm_endpoint
, auth_token
, values
, {})
161 perf_req
.assert_called_with(mock
.ANY
, auth_token
, req_type
="get")
162 check_pay
.assert_not_called()
164 @mock.patch
.object(alarm_req
.Alarming
, "check_payload")
165 @mock.patch
.object(Common
, "perform_request")
166 def test_update_alarm_invalid_payload(self
, perf_req
, check_pay
):
167 """Test update alarm with invalid payload."""
168 resp
= Response({"name": "my_alarm",
170 "gnocchi_resources_threshold_rule":
171 {"resource_id": "my_resource_id",
172 "metric": "my_metric"}})
173 perf_req
.return_value
= resp
174 check_pay
.return_value
= None
175 values
= {"alarm_uuid": "my_alarm_id"}
177 self
.alarming
.update_alarm(alarm_endpoint
, auth_token
, values
, {})
179 perf_req
.assert_called_with(mock
.ANY
, auth_token
, req_type
="get")
180 self
.assertEqual(perf_req
.call_count
, 1)
182 @mock.patch
.object(alarm_req
.Alarming
, "check_payload")
183 @mock.patch
.object(Common
, "perform_request")
184 def test_update_alarm_valid(self
, perf_req
, check_pay
):
185 """Test valid update alarm request."""
186 resp
= Response({"alarm_id": "1",
189 "gnocchi_resources_threshold_rule":
190 {"resource_id": "my_resource_id",
191 "metric": "disk.write.requests"}})
192 perf_req
.return_value
= resp
193 values
= {"alarm_uuid": "my_alarm_id"}
195 self
.alarming
.update_alarm(alarm_endpoint
, auth_token
, values
, {})
197 check_pay
.assert_called_with(values
, "disk_write_ops", "my_resource_id",
198 "my_alarm", alarm_state
="alarm")
200 self
.assertEqual(perf_req
.call_count
, 2)
201 # Second call is the update request
202 perf_req
.assert_called_with(
203 'alarm_endpoint/v2/alarms/my_alarm_id', auth_token
,
204 req_type
="put", payload
=check_pay
.return_value
)
206 @mock.patch
.object(Config
, "instance")
207 def test_check_valid_payload(self
, cfg
):
208 """Test the check payload function for a valid payload."""
209 values
= {"severity": "warning",
210 "statistic": "COUNT",
211 "threshold_value": 12,
214 "resource_type": "generic"}
215 cfg
.return_value
.OS_NOTIFIER_URI
= "http://localhost:8662"
216 payload
= self
.alarming
.check_payload(
217 values
, "disk_write_ops", "r_id", "alarm_name")
219 self
.assertDictEqual(
220 json
.loads(payload
), {"name": "alarm_name",
221 "gnocchi_resources_threshold_rule":
222 {"resource_id": "r_id",
223 "metric": "disk.write.requests",
224 "comparison_operator": "gt",
225 "aggregation_method": "count",
228 "resource_type": "generic"},
231 "type": "gnocchi_resources_threshold",
232 "alarm_actions": ["http://localhost:8662"]})
234 @mock.patch
.object(Config
, "instance")
235 @mock.patch
.object(Common
, "perform_request")
236 def test_check_valid_state_payload(self
, perform_req
, cfg
):
237 """Test the check payload function for a valid payload with state."""
238 values
= {"severity": "warning",
239 "statistic": "COUNT",
240 "threshold_value": 12,
243 "resource_type": "generic"}
244 cfg
.return_value
.OS_NOTIFIER_URI
= "http://localhost:8662"
245 payload
= self
.alarming
.check_payload(
246 values
, "disk_write_ops", "r_id", "alarm_name", alarm_state
="alarm")
249 json
.loads(payload
), {"name": "alarm_name",
250 "gnocchi_resources_threshold_rule":
251 {"resource_id": "r_id",
252 "metric": "disk.write.requests",
253 "comparison_operator": "gt",
254 "aggregation_method": "count",
257 "resource_type": "generic"},
260 "type": "gnocchi_resources_threshold",
261 "alarm_actions": ["http://localhost:8662"]})
263 def test_check_invalid_payload(self
):
264 """Test the check payload function for an invalid payload."""
265 values
= {"alarm_values": "mock_invalid_details"}
266 payload
= self
.alarming
.check_payload(
267 values
, "my_metric", "r_id", "alarm_name")
269 self
.assertEqual(payload
, None)
271 @mock.patch
.object(Common
, "perform_request")
272 def test_get_alarm_state(self
, perf_req
):
273 """Test the get alarm state function."""
274 perf_req
.return_value
= type('obj', (object,), {'text': '{"alarm_id":"1"}'})
276 self
.alarming
.get_alarm_state(alarm_endpoint
, auth_token
, "alarm_id")
278 perf_req
.assert_called_with(
279 "alarm_endpoint/v2/alarms/alarm_id/state", auth_token
, req_type
="get")
281 @mock.patch
.object(Common
, "get_endpoint")
282 @mock.patch
.object(Common
, "perform_request")
283 def test_check_for_metric(self
, perf_req
, get_endpoint
):
284 """Test the check for metric function."""
285 get_endpoint
.return_value
= "gnocchi_endpoint"
287 self
.alarming
.check_for_metric(auth_token
, metric_endpoint
, "metric_name", "r_id")
289 perf_req
.assert_called_with(
290 "metric_endpoint/v1/metric?sort=name:asc", auth_token
, req_type
="get")