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 import json
  25 import unittest
  26
  27 import mock
  28 from kafka import KafkaProducer
  29
  30 from osm_mon.core.database import DatabaseManager, Alarm
  31 from osm_mon.plugins.OpenStack.Aodh.notifier import NotifierHandler
  32
  33 post_data = {"severity": "critical",
  34              "alarm_name": "my_alarm",
  35              "current": "current_state",
  36              "alarm_id": "my_alarm_id",
  37              "reason": "Threshold has been broken",
  38              "reason_data": {"count": 1,
  39                              "most_recent": "null",
  40                              "type": "threshold",
  41                              "disposition": "unknown"},
  42              "previous": "previous_state"}
  43
  44
  45 class Response(object):
  46     """Mock a response class for generating responses."""
  47
  48     def __init__(self, text):
  49         """Initialise a mock response with a text attribute."""
  50         self.text = text
  51
  52
  53 class RFile():
  54     def read(self, content_length):
  55         return json.dumps(post_data, sort_keys=True)
  56
  57
  58 class MockNotifierHandler(NotifierHandler):
  59     """Mock the NotifierHandler class for testing purposes."""
  60
  61     def __init__(self):
  62         """Initialise mock NotifierHandler."""
  63         self.headers = {'Content-Length': '20'}
  64         self.rfile = RFile()
  65
  66     def setup(self):
  67         """Mock setup function."""
  68         pass
  69
  70     def handle(self):
  71         """Mock handle function."""
  72         pass
  73
  74     def finish(self):
  75         """Mock finish function."""
  76         pass
  77
  78
  79 @mock.patch.object(KafkaProducer, "__init__", lambda *args, **kwargs: None)
  80 @mock.patch.object(KafkaProducer, "flush", mock.Mock())
  81 class TestNotifier(unittest.TestCase):
  82     """Test the NotifierHandler class for requests from aodh."""
  83
  84     def setUp(self):
  85         """Setup tests."""
  86         super(TestNotifier, self).setUp()
  87         self.handler = MockNotifierHandler()
  88
  89     @mock.patch.object(NotifierHandler, "_set_headers")
  90     def test_do_GET(self, set_head):
  91         """Tests do_GET. Validates _set_headers has been called."""
  92         self.handler.do_GET()
  93
  94         set_head.assert_called_once()
  95
  96     @mock.patch.object(NotifierHandler, "notify_alarm")
  97     @mock.patch.object(NotifierHandler, "_set_headers")
  98     def test_do_POST(self, set_head, notify):
  99         """Tests do_POST. Validates notify_alarm has been called."""
 100         self.handler.do_POST()
 101
 102         set_head.assert_called_once()
 103         notify.assert_called_with(post_data)
 104
 105     @mock.patch.object(NotifierHandler, "_publish_response")
 106     @mock.patch.object(DatabaseManager, "get_alarm")
 107     def test_notify_alarm_valid_alarm(
 108             self, get_alarm, notify):
 109         """
 110         Tests notify_alarm when request from OpenStack references an existing alarm in the DB.
 111         Validates KafkaProducer.notify_alarm has been called.
 112         """
 113         # Generate return values for valid notify_alarm operation
 114         mock_alarm = Alarm()
 115         get_alarm.return_value = mock_alarm
 116
 117         self.handler.notify_alarm(post_data)
 118         notify.assert_called_with('notify_alarm', mock.ANY)
 119
 120     @mock.patch.object(NotifierHandler, "_publish_response")
 121     @mock.patch.object(DatabaseManager, "get_alarm")
 122     def test_notify_alarm_invalid_alarm(
 123             self, get_alarm, notify):
 124         """
 125         Tests notify_alarm when request from OpenStack references a non existing alarm in the DB.
 126         Validates Exception is thrown and KafkaProducer.notify_alarm has not been called.
 127         """
 128         # Generate return values for valid notify_alarm operation
 129         get_alarm.return_value = None
 130
 131         with self.assertRaises(Exception):
 132             self.handler.notify_alarm(post_data)
 133         notify.assert_not_called()