X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm-mon%2Ftest%2FOpenStack%2Ftest_alarming.py;fp=osm-mon%2Ftest%2FOpenStack%2Ftest_alarming.py;h=557a93dd7d0abb0eacb14c0d8db16f656da86f72;hb=71ce7eca516321aff84332df56702e718968735b;hp=0000000000000000000000000000000000000000;hpb=7d5e61f406eab8d151125001a09f742dbadc3d70;p=osm%2FMON.git diff --git a/osm-mon/test/OpenStack/test_alarming.py b/osm-mon/test/OpenStack/test_alarming.py new file mode 100644 index 0000000..557a93d --- /dev/null +++ b/osm-mon/test/OpenStack/test_alarming.py @@ -0,0 +1,271 @@ +# 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( + "/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( + "/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( + "/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( + "/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( + "/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( + '/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( + "/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")