X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_mon%2Ftest%2FOpenStack%2Funit%2Ftest_alarming.py;fp=osm_mon%2Ftest%2FOpenStack%2Funit%2Ftest_alarming.py;h=0000000000000000000000000000000000000000;hb=4da146638bc3838270fa41c9f9fb91961f726c97;hp=67486e7bb6e6e1118900ac661c7a30d38adc8ba9;hpb=2aec92e1eb52d5512c2acae9ce9878f2f3c8f782;p=osm%2FMON.git diff --git a/osm_mon/test/OpenStack/unit/test_alarming.py b/osm_mon/test/OpenStack/unit/test_alarming.py deleted file mode 100644 index 67486e7..0000000 --- a/osm_mon/test/OpenStack/unit/test_alarming.py +++ /dev/null @@ -1,299 +0,0 @@ -# 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 osm_mon.core.settings import Config -from osm_mon.plugins.OpenStack.Aodh import alarm_handler as alarm_req -from osm_mon.plugins.OpenStack.common import Common - -log = logging.getLogger(__name__) - -auth_token = mock.ANY -alarm_endpoint = "alarm_endpoint" -metric_endpoint = "metric_endpoint" - - -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.""" - - maxDiff = None - - def setUp(self): - """Setup for tests.""" - super(TestAlarming, self).setUp() - self.alarming = alarm_req.OpenstackAlarmHandler() - - @mock.patch.object(Common, "perform_request") - def test_config_invalid_alarm_req(self, perf_req): - """Test configure an invalid alarm request.""" - # Configuring with invalid metric name results in failure - values = {"alarm_name": "my_alarm", - "metric_name": "my_metric", - "resource_uuid": "my_r_id"} - with self.assertRaises(KeyError): - self.alarming.configure_alarm(alarm_endpoint, auth_token, values, {}, True) - perf_req.assert_not_called() - perf_req.reset_mock() - - # Configuring with missing metric name results in failure - values = {"alarm_name": "disk_write_ops", - "resource_uuid": "my_r_id"} - - with self.assertRaises(KeyError): - self.alarming.configure_alarm(alarm_endpoint, auth_token, values, {}, True) - perf_req.assert_not_called() - - @mock.patch.object(Common, "perform_request") - def test_config_valid_alarm_req(self, perf_req): - """Test config a valid alarm.""" - values = {"alarm_name": "disk_write_ops", - "metric_name": "disk_write_ops", - "resource_uuid": "my_r_id", - "statistic": "AVERAGE", - "threshold_value": 60, - "operation": "GT"} - - perf_req.return_value = type('obj', (object,), {'text': '{"alarm_id":"1"}'}) - - self.alarming.configure_alarm(alarm_endpoint, auth_token, values, {}, True) - payload = {"name": "disk_write_ops", - "gnocchi_resources_threshold_rule": {"resource_type": "generic", "comparison_operator": "gt", - "granularity": "300", "metric": "disk.write.requests", - "aggregation_method": "mean", "threshold": 60, - "resource_id": "my_r_id"}, - "alarm_actions": ["http://localhost:8662"], "repeat_actions": True, "state": "ok", "type": "gnocchi_resources_threshold", - "severity": "critical"} - perf_req.assert_called_with( - "alarm_endpoint/v2/alarms/", auth_token, - req_type="post", payload=json.dumps(payload, sort_keys=True), verify_ssl=True) - - @mock.patch.object(Common, "perform_request") - def test_delete_alarm_req(self, perf_req): - """Test delete alarm request.""" - self.alarming.delete_alarm(alarm_endpoint, auth_token, "my_alarm_id", True) - - perf_req.assert_called_with( - "alarm_endpoint/v2/alarms/my_alarm_id", auth_token, req_type="delete", verify_ssl=True) - - @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 without a resource_id - list_details = {"mock_details": "invalid_details"} - with self.assertRaises(KeyError): - self.alarming.list_alarms(alarm_endpoint, auth_token, list_details, True) - 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", "alarm_name": "mock_alarm", "severity": "critical"} - - mock_perf_req_return_value = [ - {"alarm_id": "1", "name": "mock_alarm", "severity": "critical", - "gnocchi_resources_threshold_rule": {"resource_id": "mock_r_id"}}] - perf_req.return_value = type('obj', (object,), - {'text': json.dumps(mock_perf_req_return_value)}) - - alarm_list = self.alarming.list_alarms(alarm_endpoint, auth_token, list_details, True) - - self.assertDictEqual(alarm_list[0], mock_perf_req_return_value[0]) - - perf_req.assert_called_with( - "alarm_endpoint/v2/alarms/", auth_token, req_type="get", verify_ssl=True) - perf_req.reset_mock() - - # Check list with alarm_name defined - list_details = {"resource_uuid": "mock_r_id", - "alarm_name": "mock_alarm", - "severity": "critical"} - alarm_list = self.alarming.list_alarms(alarm_endpoint, auth_token, list_details, True) - - self.assertDictEqual(alarm_list[0], mock_perf_req_return_value[0]) - - perf_req.assert_called_with( - "alarm_endpoint/v2/alarms/", auth_token, req_type="get", verify_ssl=True) - - @mock.patch.object(Common, "perform_request") - def test_ack_alarm_req(self, perf_req): - """Test update alarm state for acknowledge alarm request.""" - resp = Response({}) - perf_req.return_value = resp - - self.alarming.update_alarm_state(alarm_endpoint, auth_token, "my_alarm_id", True) - - perf_req.assert_called_with( - "alarm_endpoint/v2/alarms/my_alarm_id/state", auth_token, req_type="put", - payload=json.dumps("ok"), verify_ssl=True) - - @mock.patch.object(Common, "perform_request") - def test_update_alarm_invalid(self, perf_req): - """Test update alarm with invalid get response.""" - values = {"alarm_uuid": "my_alarm_id"} - - perf_req.return_value = type('obj', (object,), {'invalid_prop': 'Invalid response'}) - - with self.assertRaises(Exception): - self.alarming.update_alarm(alarm_endpoint, auth_token, values, {}, True) - perf_req.assert_called_with(mock.ANY, auth_token, req_type="get") - - @mock.patch.object(Common, "perform_request") - def test_update_alarm_invalid_payload(self, perf_req): - """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 - values = {"alarm_uuid": "my_alarm_id"} - - with self.assertRaises(Exception): - self.alarming.update_alarm(alarm_endpoint, auth_token, values, {}, True) - perf_req.assert_called_with(mock.ANY, auth_token, req_type="get") - self.assertEqual(perf_req.call_count, 1) - - @mock.patch.object(alarm_req.OpenstackAlarmHandler, "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({"alarm_id": "1", - "name": "my_alarm", - "state": "alarm", - "gnocchi_resources_threshold_rule": - {"resource_id": "my_resource_id", - "metric": "disk.write.requests"}}) - perf_req.return_value = resp - values = {"alarm_uuid": "my_alarm_id"} - - self.alarming.update_alarm(alarm_endpoint, auth_token, values, {}, True) - - check_pay.assert_called_with(values, "disk_write_ops", "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( - 'alarm_endpoint/v2/alarms/my_alarm_id', auth_token, - req_type="put", payload=check_pay.return_value, verify_ssl=True) - - @mock.patch.object(Config, "instance") - def test_check_valid_payload(self, cfg): - """Test the check payload function for a valid payload.""" - values = {"severity": "warning", - "statistic": "COUNT", - "threshold_value": 12, - "operation": "GT", - "granularity": 300, - "resource_type": "generic"} - cfg.return_value.OS_NOTIFIER_URI = "http://localhost:8662" - payload = self.alarming.check_payload( - values, "disk_write_ops", "r_id", "alarm_name") - - self.assertDictEqual( - json.loads(payload), {"name": "alarm_name", - "gnocchi_resources_threshold_rule": - {"resource_id": "r_id", - "metric": "disk.write.requests", - "comparison_operator": "gt", - "aggregation_method": "count", - "threshold": 12, - "granularity": 300, - "resource_type": "generic"}, - "severity": "low", - "state": "ok", - "type": "gnocchi_resources_threshold", - "alarm_actions": ["http://localhost:8662"], - "repeat_actions": True}) - - @mock.patch.object(Config, "instance") - @mock.patch.object(Common, "perform_request") - def test_check_valid_state_payload(self, perform_req, cfg): - """Test the check payload function for a valid payload with state.""" - values = {"severity": "warning", - "statistic": "COUNT", - "threshold_value": 12, - "operation": "GT", - "granularity": 300, - "resource_type": "generic"} - cfg.return_value.OS_NOTIFIER_URI = "http://localhost:8662" - payload = self.alarming.check_payload( - values, "disk_write_ops", "r_id", "alarm_name", alarm_state="alarm") - - self.assertEqual( - json.loads(payload), {"name": "alarm_name", - "gnocchi_resources_threshold_rule": - {"resource_id": "r_id", - "metric": "disk.write.requests", - "comparison_operator": "gt", - "aggregation_method": "count", - "threshold": 12, - "granularity": 300, - "resource_type": "generic"}, - "severity": "low", - "state": "alarm", - "type": "gnocchi_resources_threshold", - "alarm_actions": ["http://localhost:8662"], - "repeat_actions": True}) - - def test_check_invalid_payload(self): - """Test the check payload function for an invalid payload.""" - values = {"alarm_values": "mock_invalid_details"} - with self.assertRaises(Exception): - self.alarming.check_payload(values, "my_metric", "r_id", "alarm_name") - - @mock.patch.object(Common, "perform_request") - def test_get_alarm_state(self, perf_req): - """Test the get alarm state function.""" - perf_req.return_value = type('obj', (object,), {'text': '{"alarm_id":"1"}'}) - - self.alarming.get_alarm_state(alarm_endpoint, auth_token, "alarm_id") - - perf_req.assert_called_with( - "alarm_endpoint/v2/alarms/alarm_id/state", auth_token, req_type="get") - - @mock.patch.object(Common, "perform_request") - def test_check_for_metric(self, perf_req): - """Test the check for metric function.""" - mock_perf_req_return_value = {"metrics": {"cpu_util": 123}} - perf_req.return_value = type('obj', (object,), {'text': json.dumps(mock_perf_req_return_value)}) - - self.alarming.check_for_metric(auth_token, metric_endpoint, "cpu_utilization", "r_id", True) - - perf_req.assert_called_with( - "metric_endpoint/v1/resource/generic/r_id", auth_token, req_type="get", verify_ssl=True)