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 metric request message keys."""
30 from osm_mon
.core
.auth
import AuthManager
31 from osm_mon
.plugins
.OpenStack
.Gnocchi
import metric_handler
as metric_req
32 from osm_mon
.plugins
.OpenStack
.common
import Common
34 log
= logging
.getLogger(__name__
)
36 # Mock auth_token and endpoint
40 # Mock a valid metric list for some tests, and a resultant list
41 metric_list
= [{"name": "disk.write.requests",
44 "resource_id": "r_id"}]
45 result_list
= ["metric_id", "r_id", "units", "disk_write_ops"]
48 class Response(object):
49 """Mock a response object for requests."""
52 """Initialise test and status code values."""
53 self
.text
= json
.dumps([{"id": "test_id"}])
54 self
.status_code
= "STATUS_CODE"
57 def perform_request_side_effect(*args
, **kwargs
):
59 if 'marker' in args
[0]:
60 resp
.text
= json
.dumps([])
61 if 'resource/generic' in args
[0]:
62 resp
.text
= json
.dumps({'metrics': {'cpu_util': 'test_id'}})
66 class TestMetricCalls(unittest
.TestCase
):
67 """Integration test for metric request keys."""
70 """Setup the tests for metric request keys."""
71 super(TestMetricCalls
, self
).setUp()
72 self
.metrics
= metric_req
.OpenstackMetricHandler()
73 self
.metrics
._common
= Common()
75 @mock.patch
.object(metric_req
.OpenstackMetricHandler
, "get_metric_id")
76 @mock.patch
.object(Common
, "perform_request")
77 def test_invalid_config_metric_req(
78 self
, perf_req
, get_metric
):
79 """Test the configure metric function, for an invalid metric."""
80 # Test invalid configuration for creating a metric
81 values
= {"metric_details": "invalid_metric"}
83 with self
.assertRaises(ValueError):
84 self
.metrics
.configure_metric(endpoint
, auth_token
, values
, verify_ssl
=False)
86 perf_req
.assert_not_called()
88 # Test with an invalid metric name, will not perform request
89 values
= {"resource_uuid": "r_id"}
91 with self
.assertRaises(ValueError):
92 self
.metrics
.configure_metric(endpoint
, auth_token
, values
, verify_ssl
=False)
94 perf_req
.assert_not_called()
96 # If metric exists, it won't be recreated
97 get_metric
.return_value
= "metric_id"
99 with self
.assertRaises(ValueError):
100 self
.metrics
.configure_metric(endpoint
, auth_token
, values
, verify_ssl
=False)
102 perf_req
.assert_not_called()
104 @mock.patch
.object(metric_req
.OpenstackMetricHandler
, "get_metric_id")
105 @mock.patch
.object(Common
, "perform_request")
106 @mock.patch
.object(AuthManager
, "get_credentials")
107 def test_valid_config_metric_req(
108 self
, get_creds
, perf_req
, get_metric
):
109 """Test the configure metric function, for a valid metric."""
110 # Test valid configuration and payload for creating a metric
111 get_creds
.return_value
= type('obj', (object,), {'config': '{"insecure":true}'})
112 values
= {"resource_uuid": "r_id",
113 "metric_unit": "units",
114 "metric_name": "cpu_util"}
115 get_metric
.return_value
= None
116 payload
= {"id": "r_id",
117 "metrics": {"cpu_util":
118 {"archive_policy_name": "high",
122 perf_req
.return_value
= type('obj', (object,), {'text': '{"metrics":{"cpu_util":1}, "id":1}'})
124 self
.metrics
.configure_metric(endpoint
, auth_token
, values
, verify_ssl
=False)
126 perf_req
.assert_called_with(
127 "<ANY>/v1/resource/generic", auth_token
, req_type
="post", verify_ssl
=False,
128 payload
=json
.dumps(payload
, sort_keys
=True))
130 @mock.patch
.object(Common
, "perform_request")
131 def test_delete_metric_req(self
, perf_req
):
132 """Test the delete metric function."""
133 mock_response
= Response()
134 mock_response
.status_code
= 200
135 perf_req
.return_value
= mock_response
137 self
.metrics
.delete_metric(endpoint
, auth_token
, "metric_id", verify_ssl
=False)
139 perf_req
.assert_called_with(
140 "<ANY>/v1/metric/metric_id", auth_token
, req_type
="delete", verify_ssl
=False)
142 @mock.patch
.object(Common
, "perform_request")
143 def test_delete_metric_invalid_status(self
, perf_req
):
144 """Test invalid response for delete request."""
145 perf_req
.return_value
= type('obj', (object,), {"status_code": "404"})
147 with self
.assertRaises(ValueError):
148 self
.metrics
.delete_metric(endpoint
, auth_token
, "metric_id", verify_ssl
=False)
150 @mock.patch
.object(metric_req
.OpenstackMetricHandler
, "response_list")
151 @mock.patch
.object(Common
, "perform_request")
152 def test_complete_list_metric_req(self
, perf_req
, resp_list
):
153 """Test the complete list metric function."""
154 # Test listing metrics without any configuration options
156 perf_req
.side_effect
= perform_request_side_effect
157 self
.metrics
.list_metrics(endpoint
, auth_token
, values
, verify_ssl
=False)
159 perf_req
.assert_any_call(
160 "<ANY>/v1/metric?sort=name:asc", auth_token
, req_type
="get", verify_ssl
=False)
161 resp_list
.assert_called_with([{u
'id': u
'test_id'}])
163 @mock.patch
.object(metric_req
.OpenstackMetricHandler
, "response_list")
164 @mock.patch
.object(Common
, "perform_request")
165 def test_resource_list_metric_req(self
, perf_req
, resp_list
):
166 """Test the resource list metric function."""
167 # Test listing metrics with a resource id specified
168 values
= {"resource_uuid": "resource_id"}
169 perf_req
.side_effect
= perform_request_side_effect
170 self
.metrics
.list_metrics(endpoint
, auth_token
, values
, verify_ssl
=False)
172 perf_req
.assert_any_call(
173 "<ANY>/v1/metric/test_id", auth_token
, req_type
="get", verify_ssl
=False)
175 @mock.patch
.object(metric_req
.OpenstackMetricHandler
, "response_list")
176 @mock.patch
.object(Common
, "perform_request")
177 def test_name_list_metric_req(self
, perf_req
, resp_list
):
178 """Test the metric_name list metric function."""
179 # Test listing metrics with a metric_name specified
180 values
= {"metric_name": "disk_write_bytes"}
181 perf_req
.side_effect
= perform_request_side_effect
182 self
.metrics
.list_metrics(endpoint
, auth_token
, values
, verify_ssl
=False)
184 perf_req
.assert_any_call(
185 "<ANY>/v1/metric?sort=name:asc", auth_token
, req_type
="get", verify_ssl
=False)
186 resp_list
.assert_called_with(
187 [{u
'id': u
'test_id'}], metric_name
="disk_write_bytes")
189 @mock.patch
.object(metric_req
.OpenstackMetricHandler
, "response_list")
190 @mock.patch
.object(Common
, "perform_request")
191 def test_combined_list_metric_req(self
, perf_req
, resp_list
):
192 """Test the combined resource and metric list metric function."""
193 # Test listing metrics with a resource id and metric name specified
195 values
= {"resource_uuid": "resource_id",
196 "metric_name": "cpu_utilization"}
197 perf_req
.side_effect
= perform_request_side_effect
198 self
.metrics
.list_metrics(endpoint
, auth_token
, values
, verify_ssl
=False)
200 perf_req
.assert_any_call(
201 "<ANY>/v1/metric/test_id", auth_token
, req_type
="get", verify_ssl
=False)
203 @mock.patch
.object(Common
, "perform_request")
204 def test_get_metric_id(self
, perf_req
):
205 """Test get_metric_id function."""
206 mock_response
= Response()
207 mock_response
.text
= json
.dumps({'metrics': {'my_metric': 'id'}})
208 perf_req
.return_value
= mock_response
209 self
.metrics
.get_metric_id(endpoint
, auth_token
, "my_metric", "r_id", verify_ssl
=False)
211 perf_req
.assert_called_with(
212 "<ANY>/v1/resource/generic/r_id", auth_token
, req_type
="get", verify_ssl
=False)
214 @mock.patch
.object(metric_req
.OpenstackMetricHandler
, "get_metric_id")
215 @mock.patch
.object(Common
, "perform_request")
216 def test_valid_read_data_req(self
, perf_req
, get_metric
):
217 """Test the read metric data function, for a valid call."""
218 values
= {"metric_name": "disk_write_ops",
219 "resource_uuid": "resource_id",
220 "collection_unit": "DAY",
221 "collection_period": 1}
223 perf_req
.return_value
= type('obj', (object,), {'text': '{"metric_data":"[]"}'})
225 get_metric
.return_value
= "metric_id"
226 self
.metrics
.read_metric_data(endpoint
, auth_token
, values
, verify_ssl
=False)
228 perf_req
.assert_called_once()
230 @mock.patch
.object(Common
, "perform_request")
231 def test_invalid_read_data_req(self
, perf_req
):
232 """Test the read metric data function for an invalid call."""
235 with self
.assertRaises(KeyError):
236 self
.metrics
.read_metric_data(endpoint
, auth_token
, values
, verify_ssl
=False)
238 def test_complete_response_list(self
):
239 """Test the response list function for formatting metric lists."""
240 # Mock a list for testing purposes, with valid OSM metric
241 resp_list
= self
.metrics
.response_list(metric_list
)
243 # Check for the expected values in the resulting list
244 for l
in result_list
:
245 self
.assertIn(l
, resp_list
[0].values())
247 def test_name_response_list(self
):
248 """Test the response list with metric name configured."""
249 # Mock the metric name to test a metric name list
250 # Test with a name that is not in the list
251 invalid_name
= "my_metric"
252 resp_list
= self
.metrics
.response_list(
253 metric_list
, metric_name
=invalid_name
)
255 self
.assertEqual(resp_list
, [])
257 # Test with a name on the list
258 valid_name
= "disk_write_ops"
259 resp_list
= self
.metrics
.response_list(
260 metric_list
, metric_name
=valid_name
)
262 # Check for the expected values in the resulting list
263 for l
in result_list
:
264 self
.assertIn(l
, resp_list
[0].values())
266 def test_resource_response_list(self
):
267 """Test the response list with resource_id configured."""
268 # Mock a resource_id to test a resource list
269 # Test with resource not on the list
270 invalid_id
= "mock_resource"
271 resp_list
= self
.metrics
.response_list(metric_list
, resource
=invalid_id
)
273 self
.assertEqual(resp_list
, [])
275 # Test with a resource on the list
277 resp_list
= self
.metrics
.response_list(metric_list
, resource
=valid_id
)
279 # Check for the expected values in the resulting list
280 for l
in result_list
:
281 self
.assertIn(l
, resp_list
[0].values())
283 def test_combined_response_list(self
):
284 """Test the response list function with resource_id and metric_name."""
285 # Test for a combined resource and name list
286 # resource and name are on the list
287 valid_name
= "disk_write_ops"
289 resp_list
= self
.metrics
.response_list(
290 metric_list
, metric_name
=valid_name
, resource
=valid_id
)
292 # Check for the expected values in the resulting list
293 for l
in result_list
:
294 self
.assertIn(l
, resp_list
[0].values())
296 # resource not on list
297 invalid_id
= "mock_resource"
298 resp_list
= self
.metrics
.response_list(
299 metric_list
, metric_name
=valid_name
, resource
=invalid_id
)
301 self
.assertEqual(resp_list
, [])
303 # metric name not on list
304 invalid_name
= "mock_metric"
305 resp_list
= self
.metrics
.response_list(
306 metric_list
, metric_name
=invalid_name
, resource
=valid_id
)
308 self
.assertEqual(resp_list
, [])