| # Copyright 2017 iIntel Research and Development Ireland Limited |
| # ************************************************************** |
| |
| # This file is part of OSM Monitoring module |
| # All Rights Reserved to Intel Corporation |
| |
| # Licensed under the Apache License, Version 2.0 (the "License"); you may |
| # not use this file except in compliance with the License. You may obtain |
| # a copy of the License at |
| |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT |
| # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the |
| # License for the specific language governing permissions and limitations |
| # under the License. |
| |
| # For those usages not covered by the Apache License, Version 2.0 please |
| # contact: helena.mcgough@intel.com or adrian.hoban@intel.com |
| ## |
| """Tests for all alarm request message keys.""" |
| |
| import json |
| |
| import logging |
| |
| import unittest |
| |
| import mock |
| |
| from plugins.OpenStack.Aodh import alarming as alarm_req |
| from plugins.OpenStack.common import Common |
| |
| __author__ = "Helena McGough" |
| |
| log = logging.getLogger(__name__) |
| |
| auth_token = mock.ANY |
| endpoint = mock.ANY |
| |
| |
| class Response(object): |
| """Mock a response message class.""" |
| |
| def __init__(self, result): |
| """Initialise the response text and status code.""" |
| self.text = json.dumps(result) |
| self.status_code = "MOCK_STATUS_CODE" |
| |
| |
| class TestAlarming(unittest.TestCase): |
| """Tests for alarming class functions.""" |
| |
| def setUp(self): |
| """Setup for tests.""" |
| super(TestAlarming, self).setUp() |
| self.alarming = alarm_req.Alarming() |
| self.alarming.common = Common() |
| |
| @mock.patch.object(alarm_req.Alarming, "check_payload") |
| @mock.patch.object(alarm_req.Alarming, "check_for_metric") |
| @mock.patch.object(Common, "_perform_request") |
| def test_config_invalid_alarm_req(self, perf_req, check_metric, check_pay): |
| """Test configure an invalid alarm request.""" |
| # Configuring with invalid alarm name results in failure |
| values = {"alarm_name": "my_alarm", |
| "metric_name": "my_metric", |
| "resource_uuid": "my_r_id"} |
| self.alarming.configure_alarm(endpoint, auth_token, values) |
| perf_req.assert_not_called |
| perf_req.reset_mock() |
| |
| # Correct alarm_name will check for metric in Gnocchi |
| # If there isn't one an alarm won;t be created |
| values = {"alarm_name": "disk_write_ops", |
| "metric_name": "disk_write_ops", |
| "resource_uuid": "my_r_id"} |
| |
| check_metric.return_value = None |
| |
| self.alarming.configure_alarm(endpoint, auth_token, values) |
| perf_req.assert_not_called |
| |
| @mock.patch.object(alarm_req.Alarming, "check_payload") |
| @mock.patch.object(alarm_req.Alarming, "check_for_metric") |
| @mock.patch.object(Common, "_perform_request") |
| def test_config_valid_alarm_req(self, perf_req, check_metric, check_pay): |
| """Test config a valid alarm.""" |
| # Correct alarm_name will check for metric in Gnocchi |
| # And conform that the payload is configured correctly |
| values = {"alarm_name": "disk_write_ops", |
| "metric_name": "disk_write_ops", |
| "resource_uuid": "my_r_id"} |
| |
| check_metric.return_value = "my_metric_id" |
| check_pay.return_value = "my_payload" |
| |
| self.alarming.configure_alarm(endpoint, auth_token, values) |
| perf_req.assert_called_with( |
| "<ANY>/v2/alarms/", auth_token, |
| req_type="post", payload="my_payload") |
| |
| @mock.patch.object(Common, "_perform_request") |
| def test_delete_alarm_req(self, perf_req): |
| """Test delete alarm request.""" |
| self.alarming.delete_alarm(endpoint, auth_token, "my_alarm_id") |
| |
| perf_req.assert_called_with( |
| "<ANY>/v2/alarms/my_alarm_id", auth_token, req_type="delete") |
| |
| @mock.patch.object(Common, "_perform_request") |
| def test_invalid_list_alarm_req(self, perf_req): |
| """Test invalid list alarm_req.""" |
| # Request will not be performed with out a resoure_id |
| list_details = {"mock_details": "invalid_details"} |
| self.alarming.list_alarms(endpoint, auth_token, list_details) |
| |
| perf_req.assert_not_called |
| |
| @mock.patch.object(Common, "_perform_request") |
| def test_valid_list_alarm_req(self, perf_req): |
| """Test valid list alarm request.""" |
| # Minimum requirement for an alarm list is resource_id |
| list_details = {"resource_uuid": "mock_r_id"} |
| self.alarming.list_alarms(endpoint, auth_token, list_details) |
| |
| perf_req.assert_called_with( |
| "<ANY>/v2/alarms/", auth_token, req_type="get") |
| perf_req.reset_mock() |
| |
| # Check list with alarm_name defined |
| list_details = {"resource_uuid": "mock_r_id", |
| "alarm_name": "my_alarm", |
| "severity": "critical"} |
| self.alarming.list_alarms(endpoint, auth_token, list_details) |
| |
| perf_req.assert_called_with( |
| "<ANY>/v2/alarms/", auth_token, req_type="get") |
| |
| @mock.patch.object(Common, "_perform_request") |
| def test_ack_alarm_req(self, perf_req): |
| """Test update alarm state for acknowledge alarm request.""" |
| self.alarming.update_alarm_state(endpoint, auth_token, "my_alarm_id") |
| |
| perf_req.assert_called_with( |
| "<ANY>/v2/alarms/my_alarm_id/state", auth_token, req_type="put", |
| payload=json.dumps("ok")) |
| |
| @mock.patch.object(alarm_req.Alarming, "check_payload") |
| @mock.patch.object(Common, "_perform_request") |
| def test_update_alarm_invalid(self, perf_req, check_pay): |
| """Test update alarm with invalid get response.""" |
| values = {"alarm_uuid": "my_alarm_id"} |
| |
| self.alarming.update_alarm(endpoint, auth_token, values) |
| |
| perf_req.assert_called_with(mock.ANY, auth_token, req_type="get") |
| check_pay.assert_not_called |
| |
| @mock.patch.object(alarm_req.Alarming, "check_payload") |
| @mock.patch.object(Common, "_perform_request") |
| def test_update_alarm_invalid_payload(self, perf_req, check_pay): |
| """Test update alarm with invalid payload.""" |
| resp = Response({"name": "my_alarm", |
| "state": "alarm", |
| "gnocchi_resources_threshold_rule": |
| {"resource_id": "my_resource_id", |
| "metric": "my_metric"}}) |
| perf_req.return_value = resp |
| check_pay.return_value = None |
| values = {"alarm_uuid": "my_alarm_id"} |
| |
| self.alarming.update_alarm(endpoint, auth_token, values) |
| |
| perf_req.assert_called_with(mock.ANY, auth_token, req_type="get") |
| self.assertEqual(perf_req.call_count, 1) |
| |
| @mock.patch.object(alarm_req.Alarming, "check_payload") |
| @mock.patch.object(Common, "_perform_request") |
| def test_update_alarm_valid(self, perf_req, check_pay): |
| """Test valid update alarm request.""" |
| resp = Response({"name": "my_alarm", |
| "state": "alarm", |
| "gnocchi_resources_threshold_rule": |
| {"resource_id": "my_resource_id", |
| "metric": "my_metric"}}) |
| perf_req.return_value = resp |
| values = {"alarm_uuid": "my_alarm_id"} |
| |
| self.alarming.update_alarm(endpoint, auth_token, values) |
| |
| check_pay.assert_called_with(values, "my_metric", "my_resource_id", |
| "my_alarm", alarm_state="alarm") |
| |
| self.assertEqual(perf_req.call_count, 2) |
| # Second call is the update request |
| perf_req.assert_called_with( |
| '<ANY>/v2/alarms/my_alarm_id', auth_token, |
| req_type="put", payload=check_pay.return_value) |
| |
| def test_check_valid_payload(self): |
| """Test the check payload function for a valid payload.""" |
| values = {"severity": "warning", |
| "statistic": "COUNT", |
| "threshold_value": 12, |
| "operation": "GT"} |
| payload = self.alarming.check_payload( |
| values, "my_metric", "r_id", "alarm_name") |
| |
| self.assertEqual( |
| json.loads(payload), {"name": "alarm_name", |
| "gnocchi_resources_threshold_rule": |
| {"resource_id": "r_id", |
| "metric": "my_metric", |
| "comparison_operator": "gt", |
| "aggregation_method": "count", |
| "threshold": 12, |
| "resource_type": "generic"}, |
| "severity": "low", |
| "state": "ok", |
| "type": "gnocchi_resources_threshold"}) |
| |
| def test_check_valid_state_payload(self): |
| """Test the check payload function for a valid payload with state.""" |
| values = {"severity": "warning", |
| "statistic": "COUNT", |
| "threshold_value": 12, |
| "operation": "GT"} |
| payload = self.alarming.check_payload( |
| values, "my_metric", "r_id", "alarm_name", alarm_state="alarm") |
| |
| self.assertEqual( |
| json.loads(payload), {"name": "alarm_name", |
| "gnocchi_resources_threshold_rule": |
| {"resource_id": "r_id", |
| "metric": "my_metric", |
| "comparison_operator": "gt", |
| "aggregation_method": "count", |
| "threshold": 12, |
| "resource_type": "generic"}, |
| "severity": "low", |
| "state": "alarm", |
| "type": "gnocchi_resources_threshold"}) |
| |
| def test_check_invalid_payload(self): |
| """Test the check payload function for an invalid payload.""" |
| values = {"alarm_values": "mock_invalid_details"} |
| payload = self.alarming.check_payload( |
| values, "my_metric", "r_id", "alarm_name") |
| |
| self.assertEqual(payload, None) |
| |
| @mock.patch.object(Common, "_perform_request") |
| def test_get_alarm_state(self, perf_req): |
| """Test the get alarm state function.""" |
| self.alarming.get_alarm_state(endpoint, auth_token, "alarm_id") |
| |
| perf_req.assert_called_with( |
| "<ANY>/v2/alarms/alarm_id/state", auth_token, req_type="get") |
| |
| @mock.patch.object(Common, "get_endpoint") |
| @mock.patch.object(Common, "_perform_request") |
| def test_check_for_metric(self, perf_req, get_endpoint): |
| """Test the check for metric function.""" |
| get_endpoint.return_value = "gnocchi_endpoint" |
| |
| self.alarming.check_for_metric(auth_token, "metric_name", "r_id") |
| |
| perf_req.assert_called_with( |
| "gnocchi_endpoint/v1/metric/", auth_token, req_type="get") |