osm_mon/test/OpenStack/unit/test_notifier.py

   1 # Copyright 2017 Intel Research and Development Ireland Limited
   2 # *************************************************************
   3
   4 # This file is part of OSM Monitoring module
   5 # All Rights Reserved to Intel Corporation
   6
   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
  10
  11 #         http://www.apache.org/licenses/LICENSE-2.0
  12
  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
  17 # under the License.
  18
  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
  21 ##
  22 """Tests for all common OpenStack methods."""
  23
  24 # TODO: Mock database calls. Improve assertions.
  25
  26 import json
  27 import unittest
  28
  29 import mock
  30
  31 from six.moves.BaseHTTPServer import HTTPServer
  32
  33 from osm_mon.core.database import DatabaseManager, Alarm
  34 from osm_mon.core.message_bus.producer import KafkaProducer
  35 from osm_mon.plugins.OpenStack.Aodh.notifier import NotifierHandler
  36 from osm_mon.plugins.OpenStack.common import Common
  37 from osm_mon.plugins.OpenStack.response import OpenStack_Response
  38
  39 # Mock data from post request
  40 post_data = json.dumps({"severity": "critical",
  41                         "alarm_name": "my_alarm",
  42                         "current": "current_state",
  43                         "alarm_id": "my_alarm_id",
  44                         "reason": "Threshold has been broken",
  45                         "reason_data": {"count": 1,
  46                                         "most_recent": "null",
  47                                         "type": "threshold",
  48                                         "disposition": "unknown"},
  49                         "previous": "previous_state"})
  50
  51 valid_get_resp = '{"gnocchi_resources_threshold_rule":\
  52                    {"resource_id": "my_resource_id"}}'
  53
  54 invalid_get_resp = '{"gnocchi_resources_threshold_rule":\
  55                      {"resource_id": null}}'
  56
  57 valid_notify_resp = '{"notify_details": {"status": "current_state",\
  58                                          "severity": "critical",\
  59                                          "resource_uuid": "my_resource_id",\
  60                                          "alarm_uuid": "my_alarm_id",\
  61                                          "vim_type": "OpenStack",\
  62                                          "start_date": "dd-mm-yyyy 00:00"},\
  63                       "schema_version": "1.0",\
  64                       "schema_type": "notify_alarm"}'
  65
  66 invalid_notify_resp = '{"notify_details": {"invalid":"mock_details"}'
  67
  68
  69 class Response(object):
  70     """Mock a response class for generating responses."""
  71
  72     def __init__(self, text):
  73         """Initialise a mock response with a text attribute."""
  74         self.text = text
  75
  76
  77 class RFile():
  78     def read(self, content_length):
  79         return post_data
  80
  81
  82 class MockNotifierHandler(NotifierHandler):
  83     """Mock the NotifierHandler class for testing purposes."""
  84
  85     def __init__(self):
  86         """Initialise mock NotifierHandler."""
  87         self.headers = {'Content-Length': '20'}
  88         self.rfile = RFile()
  89
  90     def setup(self):
  91         """Mock setup function."""
  92         pass
  93
  94     def handle(self):
  95         """Mock handle function."""
  96         pass
  97
  98     def finish(self):
  99         """Mock finish function."""
 100         pass
 101
 102
 103 class TestNotifier(unittest.TestCase):
 104     """Test the NotifierHandler class for requests from aodh."""
 105
 106     def setUp(self):
 107         """Setup tests."""
 108         super(TestNotifier, self).setUp()
 109         self.handler = MockNotifierHandler()
 110
 111     @mock.patch.object(NotifierHandler, "_set_headers")
 112     def test_do_GET(self, set_head):
 113         """Test do_GET, generates headers for get request."""
 114         self.handler.do_GET()
 115
 116         set_head.assert_called_once()
 117
 118     @mock.patch.object(NotifierHandler, "notify_alarm")
 119     @mock.patch.object(NotifierHandler, "_set_headers")
 120     def test_do_POST(self, set_head, notify):
 121         """Test do_POST functionality for a POST request."""
 122         self.handler.do_POST()
 123
 124         set_head.assert_called_once()
 125         notify.assert_called_with(json.loads(post_data))
 126
 127     @mock.patch.object(Common, "get_endpoint")
 128     @mock.patch.object(Common, "get_auth_token")
 129     @mock.patch.object(Common, "perform_request")
 130     def test_notify_alarm_unauth(self, perf_req, auth, endpoint):
 131         """Test notify alarm when not authenticated with keystone."""
 132         # Response request will not be performed unless there is a valid
 133         # auth_token and endpoint
 134         # Invalid auth_token and endpoint
 135         auth.return_value = None
 136         endpoint.return_value = None
 137         self.handler.notify_alarm(json.loads(post_data))
 138
 139         perf_req.assert_not_called()
 140
 141         # Valid endpoint
 142         auth.return_value = None
 143         endpoint.return_value = "my_endpoint"
 144         self.handler.notify_alarm(json.loads(post_data))
 145
 146         perf_req.assert_not_called()
 147
 148         # Valid auth_token
 149         auth.return_value = "my_auth_token"
 150         endpoint.return_value = None
 151         self.handler.notify_alarm(json.loads(post_data))
 152
 153         perf_req.assert_not_called()
 154
 155     @mock.patch.object(Common, "get_endpoint")
 156     @mock.patch.object(OpenStack_Response, "generate_response")
 157     @mock.patch.object(Common, "get_auth_token")
 158     @mock.patch.object(Common, "perform_request")
 159     def test_notify_alarm_invalid_alarm(self, perf_req, auth, resp, endpoint):
 160         """Test valid authentication, invalid alarm details."""
 161         # Mock valid auth_token and endpoint
 162         auth.return_value = "my_auth_token"
 163         endpoint.return_value = "my_endpoint"
 164         perf_req.return_value = Response(invalid_get_resp)
 165
 166         self.handler.notify_alarm(json.loads(post_data))
 167
 168         # Response is not generated
 169         resp.assert_not_called()
 170
 171     @mock.patch.object(KafkaProducer, "notify_alarm")
 172     @mock.patch.object(Common, "get_endpoint")
 173     @mock.patch.object(OpenStack_Response, "generate_response")
 174     @mock.patch.object(Common, "get_auth_token")
 175     @mock.patch.object(Common, "perform_request")
 176     @mock.patch.object(DatabaseManager, "get_alarm")
 177     def test_notify_alarm_resp_call(self, get_alarm, perf_req, auth, response, endpoint, notify):
 178         """Test notify_alarm tries to generate a response for SO."""
 179         # Mock valid auth token and endpoint, valid response from aodh
 180         auth.return_value = "my_auth_token"
 181         endpoint.returm_value = "my_endpoint"
 182         perf_req.return_value = Response(valid_get_resp)
 183         mock_alarm = Alarm()
 184         get_alarm.return_value = mock_alarm
 185         self.handler.notify_alarm(json.loads(post_data))
 186
 187         notify.assert_called()
 188         response.assert_called_with('notify_alarm', a_id='my_alarm_id', date=mock.ANY, metric_name=None,
 189                                     ns_id=None, operation=None, sev='critical', state='current_state',
 190                                     threshold_value=None, vdu_name=None, vnf_member_index=None)
 191
 192     @mock.patch.object(Common, "get_endpoint")
 193     @mock.patch.object(KafkaProducer, "notify_alarm")
 194     @mock.patch.object(OpenStack_Response, "generate_response")
 195     @mock.patch.object(Common, "get_auth_token")
 196     @mock.patch.object(Common, "perform_request")
 197     @unittest.skip("Schema validation not implemented yet.")
 198     def test_notify_alarm_invalid_resp(
 199             self, perf_req, auth, response, notify, endpoint):
 200         """Test the notify_alarm function, sends response to the producer."""
 201         # Generate return values for valid notify_alarm operation
 202         auth.return_value = "my_auth_token"
 203         endpoint.return_value = "my_endpoint"
 204         perf_req.return_value = Response(valid_get_resp)
 205         response.return_value = invalid_notify_resp
 206
 207         self.handler.notify_alarm(json.loads(post_data))
 208
 209         notify.assert_not_called()
 210
 211     @mock.patch.object(Common, "get_endpoint")
 212     @mock.patch.object(KafkaProducer, "notify_alarm")
 213     @mock.patch.object(OpenStack_Response, "generate_response")
 214     @mock.patch.object(Common, "get_auth_token")
 215     @mock.patch.object(Common, "perform_request")
 216     @mock.patch.object(DatabaseManager, "get_alarm")
 217     def test_notify_alarm_valid_resp(
 218             self, get_alarm, perf_req, auth, response, notify, endpoint):
 219         """Test the notify_alarm function, sends response to the producer."""
 220         # Generate return values for valid notify_alarm operation
 221         auth.return_value = "my_auth_token"
 222         endpoint.return_value = "my_endpoint"
 223         perf_req.return_value = Response(valid_get_resp)
 224         response.return_value = valid_notify_resp
 225         mock_alarm = Alarm()
 226         get_alarm.return_value = mock_alarm
 227         self.handler.notify_alarm(json.loads(post_data))
 228
 229         notify.assert_called_with(
 230             "notify_alarm", valid_notify_resp)