osm_mon/test/OpenStack/integration/test_notify_alarm.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 from __future__ import unicode_literals
  25 import json
  26 import logging
  27 import socket
  28 import unittest
  29 from threading import Thread
  30
  31 import mock
  32 import requests
  33 from kafka import KafkaProducer
  34 from six.moves.BaseHTTPServer import BaseHTTPRequestHandler
  35 from six.moves.BaseHTTPServer import HTTPServer
  36
  37 from osm_mon.core.settings import Config
  38 from osm_mon.plugins.OpenStack.Aodh.alarm_handler import OpenstackAlarmHandler
  39 from osm_mon.plugins.OpenStack.common import Common
  40 from osm_mon.plugins.OpenStack.response import OpenStackResponseBuilder
  41
  42 log = logging.getLogger(__name__)
  43
  44 # Create an instance of the common openstack class, producer and consumer
  45 openstack_auth = Common()
  46
  47 # Mock a valid get_response for alarm details
  48 valid_get_resp = '{"gnocchi_resources_threshold_rule":\
  49                   {"resource_id": "my_resource_id"}}'
  50
  51
  52 class MockResponse(object):
  53     """Mock a response class for generating responses."""
  54
  55     def __init__(self, text):
  56         """Initialise a mock response with a text attribute."""
  57         self.text = text
  58
  59
  60 class MockNotifierHandler(BaseHTTPRequestHandler):
  61     """Mock the NotifierHandler class for testing purposes."""
  62
  63     def _set_headers(self):
  64         """Set the headers for a request."""
  65         self.send_response(200)
  66         self.send_header('Content-type', 'text/html')
  67         self.end_headers()
  68
  69     def do_GET(self):
  70         """Mock functionality for GET request."""
  71         #        self.send_response(requests.codes.ok)
  72         self._set_headers()
  73         pass
  74
  75     def do_POST(self):
  76         """Mock functionality for a POST request."""
  77         self._set_headers()
  78         content_length = int(self.headers['Content-Length'])
  79         post_data = self.rfile.read(content_length)
  80         try:
  81             post_data = post_data.decode()
  82         except AttributeError:
  83             pass
  84         self.notify_alarm(json.loads(post_data))
  85
  86     def notify_alarm(self, values):
  87         """Mock the notify_alarm functionality to generate a valid response."""
  88         cfg = Config.instance()
  89         self._alarming = OpenstackAlarmHandler()
  90         self._common = Common()
  91         self._response = OpenStackResponseBuilder()
  92         alarm_id = values['alarm_id']
  93
  94         auth_token = Common.get_auth_token('test_id')
  95         endpoint = Common.get_endpoint('alarming', 'test_id')
  96
  97         # If authenticated generate and send response message
  98         if auth_token is not None and endpoint is not None:
  99             url = "{}/v2/alarms/%s".format(endpoint) % alarm_id
 100
 101             # Get the resource_id of the triggered alarm and the date
 102             result = Common.perform_request(
 103                 url, auth_token, req_type="get")
 104             alarm_details = json.loads(result.text)
 105             gnocchi_rule = alarm_details['gnocchi_resources_threshold_rule']
 106             resource_id = gnocchi_rule['resource_id']
 107             # Mock a date for testing purposes
 108             a_date = "dd-mm-yyyy 00:00"
 109
 110             # Process an alarm notification if resource_id is valid
 111             if resource_id is not None:
 112                 # Try generate and send response
 113                 try:
 114                     resp_message = self._response.generate_response(
 115                         'notify_alarm',
 116                         alarm_id=alarm_id,
 117                         resource_id=resource_id,
 118                         sev=values['severity'], date=a_date,
 119                         state=values['current'], vim_type="OpenStack")
 120                 except Exception:
 121                     log.exception("Error generating response")
 122
 123
 124 def get_free_port():
 125     """Function to get a free port to run the test webserver on."""
 126     s = socket.socket(socket.AF_INET, type=socket.SOCK_STREAM)
 127     s.bind(('localhost', 0))
 128     address, port = s.getsockname()
 129     s.close()
 130     return port
 131
 132
 133 # Create the webserver, port and run it on its own thread
 134 mock_server_port = get_free_port()
 135 mock_server = HTTPServer(('localhost', mock_server_port), MockNotifierHandler)
 136 mock_server_thread = Thread(target=mock_server.serve_forever)
 137 mock_server_thread.setDaemon(True)
 138 mock_server_thread.start()
 139
 140
 141 def test_do_get():
 142     """Integration test for get request on notifier webserver."""
 143     url = 'http://localhost:{port}/users'.format(port=mock_server_port)
 144
 145     # Send a request to the mock API server and store the response.
 146     response = requests.get(url)
 147
 148     # Confirm that the request-response cycle completed successfully.
 149     assert response.ok
 150
 151
 152 class AlarmNotificationTest(unittest.TestCase):
 153     @mock.patch.object(OpenStackResponseBuilder, "generate_response")
 154     @mock.patch.object(Common, "perform_request")
 155     @mock.patch.object(Common, "get_endpoint")
 156     @mock.patch.object(Common, "get_auth_token")
 157     def test_post_notify_alarm(self, auth, endpoint, perf_req, resp):
 158         """Integration test for notify_alarm."""
 159         url = 'http://localhost:{port}/users'.format(port=mock_server_port)
 160         payload = {"severity": "critical",
 161                    "alarm_name": "my_alarm",
 162                    "current": "current_state",
 163                    "alarm_id": "my_alarm_id",
 164                    "reason": "Threshold has been broken",
 165                    "reason_data": {"count": 1,
 166                                    "most_recent": "null",
 167                                    "type": "threshold",
 168                                    "disposition": "unknown"},
 169                    "previous": "previous_state"}
 170
 171         # Mock authenticate and request response for testing
 172         auth.return_value = "my_auth_token"
 173         endpoint.return_value = "my_endpoint"
 174         perf_req.return_value = MockResponse(valid_get_resp)
 175
 176         # Generate a post request for testing
 177         response = requests.post(url, json.dumps(payload))
 178         self.assertEqual(response.status_code, 200)
 179         # A response message is generated with the following details
 180         resp.assert_called_with(
 181             "notify_alarm", alarm_id="my_alarm_id", resource_id="my_resource_id",
 182             sev="critical", date='dd-mm-yyyy 00:00', state="current_state",
 183             vim_type="OpenStack")