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
.plugins
.OpenStack
.Aodh
import alarming
as alarm_req
31 from osm_mon
.plugins
.OpenStack
.common
import Common
32 from osm_mon
.plugins
.OpenStack
.settings
import Config
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 self
.alarming
.configure_alarm(alarm_endpoint
, metric_endpoint
, auth_token
, values
)
99 perf_req
.assert_called_with(
100 "alarm_endpoint/v2/alarms/", auth_token
,
101 req_type
="post", payload
="my_payload")
103 @mock.patch
.object(Common
, "perform_request")
104 def test_delete_alarm_req(self
, perf_req
):
105 """Test delete alarm request."""
106 self
.alarming
.delete_alarm(alarm_endpoint
, auth_token
, "my_alarm_id")
108 perf_req
.assert_called_with(
109 "alarm_endpoint/v2/alarms/my_alarm_id", auth_token
, req_type
="delete")
111 @mock.patch
.object(Common
, "perform_request")
112 def test_invalid_list_alarm_req(self
, perf_req
):
113 """Test invalid list alarm_req."""
114 # Request will not be performed with out a resoure_id
115 list_details
= {"mock_details": "invalid_details"}
116 self
.alarming
.list_alarms(alarm_endpoint
, auth_token
, list_details
)
118 perf_req
.assert_not_called()
120 @mock.patch
.object(Common
, "perform_request")
121 def test_valid_list_alarm_req(self
, perf_req
):
122 """Test valid list alarm request."""
123 # Minimum requirement for an alarm list is resource_id
124 list_details
= {"resource_uuid": "mock_r_id"}
125 self
.alarming
.list_alarms(alarm_endpoint
, auth_token
, list_details
)
127 perf_req
.assert_called_with(
128 "alarm_endpoint/v2/alarms/", auth_token
, req_type
="get")
129 perf_req
.reset_mock()
131 # Check list with alarm_name defined
132 list_details
= {"resource_uuid": "mock_r_id",
133 "alarm_name": "my_alarm",
134 "severity": "critical"}
135 self
.alarming
.list_alarms(alarm_endpoint
, auth_token
, list_details
)
137 perf_req
.assert_called_with(
138 "alarm_endpoint/v2/alarms/", auth_token
, req_type
="get")
140 @mock.patch
.object(Common
, "perform_request")
141 def test_ack_alarm_req(self
, perf_req
):
142 """Test update alarm state for acknowledge alarm request."""
143 self
.alarming
.update_alarm_state(alarm_endpoint
, auth_token
, "my_alarm_id")
145 perf_req
.assert_called_with(
146 "alarm_endpoint/v2/alarms/my_alarm_id/state", auth_token
, req_type
="put",
147 payload
=json
.dumps("ok"))
149 @mock.patch
.object(alarm_req
.Alarming
, "check_payload")
150 @mock.patch
.object(Common
, "perform_request")
151 def test_update_alarm_invalid(self
, perf_req
, check_pay
):
152 """Test update alarm with invalid get response."""
153 values
= {"alarm_uuid": "my_alarm_id"}
155 self
.alarming
.update_alarm(alarm_endpoint
, auth_token
, values
)
157 perf_req
.assert_called_with(mock
.ANY
, auth_token
, req_type
="get")
158 check_pay
.assert_not_called()
160 @mock.patch
.object(alarm_req
.Alarming
, "check_payload")
161 @mock.patch
.object(Common
, "perform_request")
162 def test_update_alarm_invalid_payload(self
, perf_req
, check_pay
):
163 """Test update alarm with invalid payload."""
164 resp
= Response({"name": "my_alarm",
166 "gnocchi_resources_threshold_rule":
167 {"resource_id": "my_resource_id",
168 "metric": "my_metric"}})
169 perf_req
.return_value
= resp
170 check_pay
.return_value
= None
171 values
= {"alarm_uuid": "my_alarm_id"}
173 self
.alarming
.update_alarm(alarm_endpoint
, auth_token
, values
)
175 perf_req
.assert_called_with(mock
.ANY
, auth_token
, req_type
="get")
176 self
.assertEqual(perf_req
.call_count
, 1)
178 @mock.patch
.object(alarm_req
.Alarming
, "check_payload")
179 @mock.patch
.object(Common
, "perform_request")
180 def test_update_alarm_valid(self
, perf_req
, check_pay
):
181 """Test valid update alarm request."""
182 resp
= Response({"name": "my_alarm",
184 "gnocchi_resources_threshold_rule":
185 {"resource_id": "my_resource_id",
186 "metric": "disk.write.requests"}})
187 perf_req
.return_value
= resp
188 values
= {"alarm_uuid": "my_alarm_id"}
190 self
.alarming
.update_alarm(alarm_endpoint
, auth_token
, values
)
192 check_pay
.assert_called_with(values
, "disk_write_ops", "my_resource_id",
193 "my_alarm", alarm_state
="alarm")
195 self
.assertEqual(perf_req
.call_count
, 2)
196 # Second call is the update request
197 perf_req
.assert_called_with(
198 'alarm_endpoint/v2/alarms/my_alarm_id', auth_token
,
199 req_type
="put", payload
=check_pay
.return_value
)
201 @mock.patch
.object(Config
, "instance")
202 def test_check_valid_payload(self
, cfg
):
203 """Test the check payload function for a valid payload."""
204 values
= {"severity": "warning",
205 "statistic": "COUNT",
206 "threshold_value": 12,
209 "resource_type": "generic"}
210 cfg
.return_value
.OS_NOTIFIER_URI
= "http://localhost:8662"
211 payload
= self
.alarming
.check_payload(
212 values
, "disk_write_ops", "r_id", "alarm_name")
214 self
.assertDictEqual(
215 json
.loads(payload
), {"name": "alarm_name",
216 "gnocchi_resources_threshold_rule":
217 {"resource_id": "r_id",
218 "metric": "disk.write.requests",
219 "comparison_operator": "gt",
220 "aggregation_method": "count",
223 "resource_type": "generic"},
226 "type": "gnocchi_resources_threshold",
227 "alarm_actions": ["http://localhost:8662"]})
229 @mock.patch
.object(Config
, "instance")
230 @mock.patch
.object(Common
, "perform_request")
231 def test_check_valid_state_payload(self
, perform_req
, cfg
):
232 """Test the check payload function for a valid payload with state."""
233 values
= {"severity": "warning",
234 "statistic": "COUNT",
235 "threshold_value": 12,
238 "resource_type": "generic"}
239 cfg
.return_value
.OS_NOTIFIER_URI
= "http://localhost:8662"
240 payload
= self
.alarming
.check_payload(
241 values
, "disk_write_ops", "r_id", "alarm_name", alarm_state
="alarm")
244 json
.loads(payload
), {"name": "alarm_name",
245 "gnocchi_resources_threshold_rule":
246 {"resource_id": "r_id",
247 "metric": "disk.write.requests",
248 "comparison_operator": "gt",
249 "aggregation_method": "count",
252 "resource_type": "generic"},
255 "type": "gnocchi_resources_threshold",
256 "alarm_actions": ["http://localhost:8662"]})
258 def test_check_invalid_payload(self
):
259 """Test the check payload function for an invalid payload."""
260 values
= {"alarm_values": "mock_invalid_details"}
261 payload
= self
.alarming
.check_payload(
262 values
, "my_metric", "r_id", "alarm_name")
264 self
.assertEqual(payload
, None)
266 @mock.patch
.object(Common
, "perform_request")
267 def test_get_alarm_state(self
, perf_req
):
268 """Test the get alarm state function."""
269 self
.alarming
.get_alarm_state(alarm_endpoint
, auth_token
, "alarm_id")
271 perf_req
.assert_called_with(
272 "alarm_endpoint/v2/alarms/alarm_id/state", auth_token
, req_type
="get")
274 @mock.patch
.object(Common
, "get_endpoint")
275 @mock.patch
.object(Common
, "perform_request")
276 def test_check_for_metric(self
, perf_req
, get_endpoint
):
277 """Test the check for metric function."""
278 get_endpoint
.return_value
= "gnocchi_endpoint"
280 self
.alarming
.check_for_metric(auth_token
, metric_endpoint
, "metric_name", "r_id")
282 perf_req
.assert_called_with(
283 "metric_endpoint/v1/metric?sort=name:asc", auth_token
, req_type
="get")