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."""
32 from osm_mon
.plugins
.OpenStack
.Aodh
import alarming
as alarm_req
33 from osm_mon
.plugins
.OpenStack
.common
import Common
35 log
= logging
.getLogger(__name__
)
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()
59 self
.alarming
.common
= Common()
61 @mock.patch
.object(alarm_req
.Alarming
, "check_payload")
62 @mock.patch
.object(alarm_req
.Alarming
, "check_for_metric")
63 @mock.patch
.object(Common
, "_perform_request")
64 def test_config_invalid_alarm_req(self
, perf_req
, check_metric
, check_pay
):
65 """Test configure an invalid alarm request."""
66 # Configuring with invalid alarm name results in failure
67 values
= {"alarm_name": "my_alarm",
68 "metric_name": "my_metric",
69 "resource_uuid": "my_r_id"}
70 self
.alarming
.configure_alarm(endpoint
, auth_token
, values
)
71 perf_req
.assert_not_called
74 # Correct alarm_name will check for metric in Gnocchi
75 # If there isn't one an alarm won;t be created
76 values
= {"alarm_name": "disk_write_ops",
77 "metric_name": "disk_write_ops",
78 "resource_uuid": "my_r_id"}
80 check_metric
.return_value
= None
82 self
.alarming
.configure_alarm(endpoint
, auth_token
, values
)
83 perf_req
.assert_not_called
85 @mock.patch
.object(alarm_req
.Alarming
, "check_payload")
86 @mock.patch
.object(alarm_req
.Alarming
, "check_for_metric")
87 @mock.patch
.object(Common
, "_perform_request")
88 def test_config_valid_alarm_req(self
, perf_req
, check_metric
, check_pay
):
89 """Test config a valid alarm."""
90 # Correct alarm_name will check for metric in Gnocchi
91 # And conform that the payload is configured correctly
92 values
= {"alarm_name": "disk_write_ops",
93 "metric_name": "disk_write_ops",
94 "resource_uuid": "my_r_id"}
96 check_metric
.return_value
= "my_metric_id"
97 check_pay
.return_value
= "my_payload"
99 self
.alarming
.configure_alarm(endpoint
, auth_token
, values
)
100 perf_req
.assert_called_with(
101 "<ANY>/v2/alarms/", auth_token
,
102 req_type
="post", payload
="my_payload")
104 @mock.patch
.object(Common
, "_perform_request")
105 def test_delete_alarm_req(self
, perf_req
):
106 """Test delete alarm request."""
107 self
.alarming
.delete_alarm(endpoint
, auth_token
, "my_alarm_id")
109 perf_req
.assert_called_with(
110 "<ANY>/v2/alarms/my_alarm_id", auth_token
, req_type
="delete")
112 @mock.patch
.object(Common
, "_perform_request")
113 def test_invalid_list_alarm_req(self
, perf_req
):
114 """Test invalid list alarm_req."""
115 # Request will not be performed with out a resoure_id
116 list_details
= {"mock_details": "invalid_details"}
117 self
.alarming
.list_alarms(endpoint
, auth_token
, list_details
)
119 perf_req
.assert_not_called
121 @mock.patch
.object(Common
, "_perform_request")
122 def test_valid_list_alarm_req(self
, perf_req
):
123 """Test valid list alarm request."""
124 # Minimum requirement for an alarm list is resource_id
125 list_details
= {"resource_uuid": "mock_r_id"}
126 self
.alarming
.list_alarms(endpoint
, auth_token
, list_details
)
128 perf_req
.assert_called_with(
129 "<ANY>/v2/alarms/", auth_token
, req_type
="get")
130 perf_req
.reset_mock()
132 # Check list with alarm_name defined
133 list_details
= {"resource_uuid": "mock_r_id",
134 "alarm_name": "my_alarm",
135 "severity": "critical"}
136 self
.alarming
.list_alarms(endpoint
, auth_token
, list_details
)
138 perf_req
.assert_called_with(
139 "<ANY>/v2/alarms/", auth_token
, req_type
="get")
141 @mock.patch
.object(Common
, "_perform_request")
142 def test_ack_alarm_req(self
, perf_req
):
143 """Test update alarm state for acknowledge alarm request."""
144 self
.alarming
.update_alarm_state(endpoint
, auth_token
, "my_alarm_id")
146 perf_req
.assert_called_with(
147 "<ANY>/v2/alarms/my_alarm_id/state", auth_token
, req_type
="put",
148 payload
=json
.dumps("ok"))
150 @mock.patch
.object(alarm_req
.Alarming
, "check_payload")
151 @mock.patch
.object(Common
, "_perform_request")
152 def test_update_alarm_invalid(self
, perf_req
, check_pay
):
153 """Test update alarm with invalid get response."""
154 values
= {"alarm_uuid": "my_alarm_id"}
156 self
.alarming
.update_alarm(endpoint
, auth_token
, values
)
158 perf_req
.assert_called_with(mock
.ANY
, auth_token
, req_type
="get")
159 check_pay
.assert_not_called
161 @mock.patch
.object(alarm_req
.Alarming
, "check_payload")
162 @mock.patch
.object(Common
, "_perform_request")
163 def test_update_alarm_invalid_payload(self
, perf_req
, check_pay
):
164 """Test update alarm with invalid payload."""
165 resp
= Response({"name": "my_alarm",
167 "gnocchi_resources_threshold_rule":
168 {"resource_id": "my_resource_id",
169 "metric": "my_metric"}})
170 perf_req
.return_value
= resp
171 check_pay
.return_value
= None
172 values
= {"alarm_uuid": "my_alarm_id"}
174 self
.alarming
.update_alarm(endpoint
, auth_token
, values
)
176 perf_req
.assert_called_with(mock
.ANY
, auth_token
, req_type
="get")
177 self
.assertEqual(perf_req
.call_count
, 1)
179 @mock.patch
.object(alarm_req
.Alarming
, "check_payload")
180 @mock.patch
.object(Common
, "_perform_request")
181 def test_update_alarm_valid(self
, perf_req
, check_pay
):
182 """Test valid update alarm request."""
183 resp
= Response({"name": "my_alarm",
185 "gnocchi_resources_threshold_rule":
186 {"resource_id": "my_resource_id",
187 "metric": "my_metric"}})
188 perf_req
.return_value
= resp
189 values
= {"alarm_uuid": "my_alarm_id"}
191 self
.alarming
.update_alarm(endpoint
, auth_token
, values
)
193 check_pay
.assert_called_with(values
, "my_metric", "my_resource_id",
194 "my_alarm", alarm_state
="alarm")
196 self
.assertEqual(perf_req
.call_count
, 2)
197 # Second call is the update request
198 perf_req
.assert_called_with(
199 '<ANY>/v2/alarms/my_alarm_id', auth_token
,
200 req_type
="put", payload
=check_pay
.return_value
)
202 def test_check_valid_payload(self
):
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 payload
= self
.alarming
.check_payload(
211 values
, "disk_write_ops", "r_id", "alarm_name")
213 self
.assertDictEqual(
214 json
.loads(payload
), {"name": "alarm_name",
215 "gnocchi_resources_threshold_rule":
216 {"resource_id": "r_id",
217 "metric": "disk.disk_ops",
218 "comparison_operator": "gt",
219 "aggregation_method": "count",
222 "resource_type": "generic"},
225 "type": "gnocchi_resources_threshold",
226 "alarm_actions": ["http://localhost:8662"]})
228 def test_check_valid_state_payload(self
):
229 """Test the check payload function for a valid payload with state."""
230 values
= {"severity": "warning",
231 "statistic": "COUNT",
232 "threshold_value": 12,
235 "resource_type": "generic"}
236 payload
= self
.alarming
.check_payload(
237 values
, "disk_write_ops", "r_id", "alarm_name", alarm_state
="alarm")
240 json
.loads(payload
), {"name": "alarm_name",
241 "gnocchi_resources_threshold_rule":
242 {"resource_id": "r_id",
243 "metric": "disk.disk_ops",
244 "comparison_operator": "gt",
245 "aggregation_method": "count",
248 "resource_type": "generic"},
251 "type": "gnocchi_resources_threshold",
252 "alarm_actions": ["http://localhost:8662"]})
254 def test_check_invalid_payload(self
):
255 """Test the check payload function for an invalid payload."""
256 values
= {"alarm_values": "mock_invalid_details"}
257 payload
= self
.alarming
.check_payload(
258 values
, "my_metric", "r_id", "alarm_name")
260 self
.assertEqual(payload
, None)
262 @mock.patch
.object(Common
, "_perform_request")
263 def test_get_alarm_state(self
, perf_req
):
264 """Test the get alarm state function."""
265 self
.alarming
.get_alarm_state(endpoint
, auth_token
, "alarm_id")
267 perf_req
.assert_called_with(
268 "<ANY>/v2/alarms/alarm_id/state", auth_token
, req_type
="get")
270 @mock.patch
.object(Common
, "get_endpoint")
271 @mock.patch
.object(Common
, "_perform_request")
272 def test_check_for_metric(self
, perf_req
, get_endpoint
):
273 """Test the check for metric function."""
274 get_endpoint
.return_value
= "gnocchi_endpoint"
276 self
.alarming
.check_for_metric(auth_token
, "metric_name", "r_id")
278 perf_req
.assert_called_with(
279 "gnocchi_endpoint/v1/metric?sort=name:asc", auth_token
, req_type
="get")