X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_mon%2Ftests%2Fplugins%2FOpenStack%2Funit%2Ftest_notifier.py;fp=osm_mon%2Ftests%2Fplugins%2FOpenStack%2Funit%2Ftest_notifier.py;h=a420c708467755cbec010ff37d37c34bdcdcbb94;hb=6eda48bf7489c00fee4e1f276404ab82db23fd1d;hp=0000000000000000000000000000000000000000;hpb=058d51d5f8ba1666242a12f52d99bdd74ec7f94f;p=osm%2FMON.git

diff --git a/osm_mon/tests/plugins/OpenStack/unit/test_notifier.py b/osm_mon/tests/plugins/OpenStack/unit/test_notifier.py
new file mode 100644
index 0000000..a420c70
--- /dev/null
+++ b/osm_mon/tests/plugins/OpenStack/unit/test_notifier.py
@@ -0,0 +1,133 @@
+# Copyright 2017 Intel 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 common OpenStack methods."""
+
+import json
+import unittest
+
+import mock
+from kafka import KafkaProducer
+
+from osm_mon.core.database import DatabaseManager, Alarm
+from osm_mon.plugins.OpenStack.Aodh.notifier import NotifierHandler
+
+post_data = {"severity": "critical",
+             "alarm_name": "my_alarm",
+             "current": "current_state",
+             "alarm_id": "my_alarm_id",
+             "reason": "Threshold has been broken",
+             "reason_data": {"count": 1,
+                             "most_recent": "null",
+                             "type": "threshold",
+                             "disposition": "unknown"},
+             "previous": "previous_state"}
+
+
+class Response(object):
+    """Mock a response class for generating responses."""
+
+    def __init__(self, text):
+        """Initialise a mock response with a text attribute."""
+        self.text = text
+
+
+class RFile():
+    def read(self, content_length):
+        return json.dumps(post_data, sort_keys=True)
+
+
+class MockNotifierHandler(NotifierHandler):
+    """Mock the NotifierHandler class for testing purposes."""
+
+    def __init__(self):
+        """Initialise mock NotifierHandler."""
+        self.headers = {'Content-Length': '20'}
+        self.rfile = RFile()
+
+    def setup(self):
+        """Mock setup function."""
+        pass
+
+    def handle(self):
+        """Mock handle function."""
+        pass
+
+    def finish(self):
+        """Mock finish function."""
+        pass
+
+
+@mock.patch.object(KafkaProducer, "__init__", lambda *args, **kwargs: None)
+@mock.patch.object(KafkaProducer, "flush", mock.Mock())
+class TestNotifier(unittest.TestCase):
+    """Test the NotifierHandler class for requests from aodh."""
+
+    def setUp(self):
+        """Setup tests."""
+        super(TestNotifier, self).setUp()
+        self.handler = MockNotifierHandler()
+
+    @mock.patch.object(NotifierHandler, "_set_headers")
+    def test_do_GET(self, set_head):
+        """Tests do_GET. Validates _set_headers has been called."""
+        self.handler.do_GET()
+
+        set_head.assert_called_once()
+
+    @mock.patch.object(NotifierHandler, "notify_alarm")
+    @mock.patch.object(NotifierHandler, "_set_headers")
+    def test_do_POST(self, set_head, notify):
+        """Tests do_POST. Validates notify_alarm has been called."""
+        self.handler.do_POST()
+
+        set_head.assert_called_once()
+        notify.assert_called_with(post_data)
+
+    @mock.patch.object(NotifierHandler, "_publish_response")
+    @mock.patch.object(DatabaseManager, "get_alarm")
+    def test_notify_alarm_valid_alarm(
+            self, get_alarm, notify):
+        """
+        Tests notify_alarm when request from OpenStack references an existing alarm in the DB.
+        Validates KafkaProducer.notify_alarm has been called.
+        """
+        # Generate return values for valid notify_alarm operation
+        mock_alarm = Alarm()
+        get_alarm.return_value = mock_alarm
+
+        self.handler.notify_alarm(post_data)
+        notify.assert_called_with('notify_alarm', mock.ANY)
+
+    @mock.patch.object(NotifierHandler, "_publish_response")
+    @mock.patch.object(DatabaseManager, "get_alarm")
+    def test_notify_alarm_invalid_alarm(
+            self, get_alarm, notify):
+        """
+        Tests notify_alarm when request from OpenStack references a non existing alarm in the DB.
+        Validates Exception is thrown and KafkaProducer.notify_alarm has not been called.
+        """
+        # Generate return values for valid notify_alarm operation
+        get_alarm.return_value = None
+
+        with self.assertRaises(Exception):
+            self.handler.notify_alarm(post_data)
+        notify.assert_not_called()