1 # Copyright 2017 Intel Research and Development Ireland Limited
2 # *************************************************************
4 # This file is part of OSM Monitoring module
5 # All Rights Reserved to Intel Corporation
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
11 # http://www.apache.org/licenses/LICENSE-2.0
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
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
22 """Tests for all common OpenStack methods."""
24 from __future__
import unicode_literals
29 from threading
import Thread
33 from kafka
import KafkaProducer
34 from six
.moves
.BaseHTTPServer
import BaseHTTPRequestHandler
35 from six
.moves
.BaseHTTPServer
import HTTPServer
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
42 log
= logging
.getLogger(__name__
)
44 # Create an instance of the common openstack class, producer and consumer
45 openstack_auth
= Common()
47 # Mock a valid get_response for alarm details
48 valid_get_resp
= '{"gnocchi_resources_threshold_rule":\
49 {"resource_id": "my_resource_id"}}'
52 class MockResponse(object):
53 """Mock a response class for generating responses."""
55 def __init__(self
, text
):
56 """Initialise a mock response with a text attribute."""
60 class MockNotifierHandler(BaseHTTPRequestHandler
):
61 """Mock the NotifierHandler class for testing purposes."""
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')
70 """Mock functionality for GET request."""
71 # self.send_response(requests.codes.ok)
76 """Mock functionality for a POST request."""
78 content_length
= int(self
.headers
['Content-Length'])
79 post_data
= self
.rfile
.read(content_length
)
81 post_data
= post_data
.decode()
82 except AttributeError:
84 self
.notify_alarm(json
.loads(post_data
))
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']
94 auth_token
= Common
.get_auth_token('test_id')
95 endpoint
= Common
.get_endpoint('alarming', 'test_id')
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
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"
110 # Process an alarm notification if resource_id is valid
111 if resource_id
is not None:
112 # Try generate and send response
114 resp_message
= self
._response
.generate_response(
117 resource_id
=resource_id
,
118 sev
=values
['severity'], date
=a_date
,
119 state
=values
['current'], vim_type
="OpenStack")
121 log
.exception("Error generating response")
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()
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()
142 """Integration test for get request on notifier webserver."""
143 url
= 'http://localhost:{port}/users'.format(port
=mock_server_port
)
145 # Send a request to the mock API server and store the response.
146 response
= requests
.get(url
)
148 # Confirm that the request-response cycle completed successfully.
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",
168 "disposition": "unknown"},
169 "previous": "previous_state"}
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
)
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")