
Change upgrades the monasca-notification to leverage the capabilities of both oslo.log and oslo.conf: - configuration of logging separated from application settings - ability to enforce data types for application settings - ability to use oslo.config-generator capabilities - automatic configuration parsing done by oslo.cfg That change will bring it closer to the rest of monasca components where such transition has happened already. However, in the rest of monasca, oslo.cfg was partially or fully implemented whereas monasca-notification has been relying on YAML based configuration file. Therefore backward compatybility for such format will be kept for now. Story: 2000959 Task: 4093 Task: 4092 Change-Id: Ia75c3b60d0fada854178f21ca5ccb9e6a880f37f
248 lines
11 KiB
Python
248 lines
11 KiB
Python
# Copyright 2017 FUJITSU LIMITED
|
|
#
|
|
# 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.
|
|
|
|
import mock
|
|
import six
|
|
|
|
from monasca_notification import notification as m_notification
|
|
from monasca_notification.plugins import jira_notifier
|
|
from tests import base
|
|
|
|
if six.PY2:
|
|
import Queue as queue
|
|
else:
|
|
import queue
|
|
|
|
|
|
def alarm():
|
|
return {'tenantId': '0',
|
|
'alarmId': '0',
|
|
'alarmDefinitionId': 0,
|
|
'alarmName': 'test Alarm',
|
|
'alarmDescription': 'test Alarm description',
|
|
'oldState': 'OK',
|
|
'newState': 'ALARM',
|
|
'severity': 'CRITICAL',
|
|
'link': 'some-link',
|
|
'lifecycleState': 'OPEN',
|
|
'stateChangeReason': 'I am alarming!',
|
|
'timestamp': 1429023453632,
|
|
'metrics': {'dimension': {'hostname': 'foo1', 'service': 'bar1'}}}
|
|
|
|
|
|
def issue(component=True, custom_config=False):
|
|
|
|
issue = {'fields': {'description': 'Monasca alarm',
|
|
'issuetype': {'name': 'Bug'},
|
|
'project': {'key': 'MyProject'},
|
|
'summary': 'Monasca alarm for alarm_defintion test Alarm status '
|
|
'changed to ALARM for the alarm_id 0'}}
|
|
if component:
|
|
issue['fields'].update({'components': [{'name': 'MyComponent'}]})
|
|
if custom_config:
|
|
alarm_value = alarm()
|
|
issue['fields'].update({'description': alarm_value.get('alarmName')})
|
|
summary_format_string = 'Alarm created for {0} with severity {1} for {2}'
|
|
summary = summary_format_string.format(alarm_value.get('alarmName'),
|
|
alarm_value.get('newState'),
|
|
alarm_value.get('alarmId'))
|
|
issue['fields'].update({'summary': summary})
|
|
return issue
|
|
|
|
|
|
class RequestsResponse(object):
|
|
def __init__(self, status):
|
|
self.status_code = status
|
|
|
|
|
|
class TestJira(base.PluginTestCase):
|
|
|
|
default_address = 'http://test.jira:3333/?project=MyProject' \
|
|
'&component=MyComponent'
|
|
default_transitions_value = [[{'id': 100, 'name': 'reopen'}]]
|
|
issue_status_resolved = 'resolved'
|
|
|
|
def setUp(self):
|
|
super(TestJira, self).setUp(
|
|
jira_notifier.register_opts
|
|
)
|
|
self.conf_override(
|
|
group='jira_notifier',
|
|
user='username',
|
|
password='password'
|
|
)
|
|
|
|
self._trap = queue.Queue()
|
|
|
|
mock_log = mock.Mock()
|
|
mock_log.info = self._trap.put
|
|
mock_log.debug = self._trap.put
|
|
mock_log.warn = self._trap.put
|
|
mock_log.error = self._trap.put
|
|
mock_log.exception = self._trap.put
|
|
|
|
self._jr = jira_notifier.JiraNotifier(mock_log)
|
|
|
|
@mock.patch('monasca_notification.plugins.jira_notifier.jira')
|
|
def _notify(self,
|
|
transitions_value,
|
|
issue_status,
|
|
address,
|
|
mock_jira):
|
|
alarm_dict = alarm()
|
|
|
|
mock_jira_obj = mock.Mock()
|
|
mock_jira.JIRA.return_value = mock_jira_obj
|
|
mock_jira_issue = mock.Mock()
|
|
if issue_status:
|
|
mock_jira_obj.search_issues.return_value = [mock_jira_issue]
|
|
mock_jira_issue.fields.status.name = issue_status
|
|
else:
|
|
mock_jira_obj.search_issues.return_value = []
|
|
mock_jira_obj.transitions.side_effect = transitions_value
|
|
|
|
notification = m_notification.Notification(0, 'jira',
|
|
'jira notification',
|
|
address, 0, 0, alarm_dict)
|
|
|
|
return mock_jira, mock_jira_obj, self._jr.send_notification(notification)
|
|
|
|
def test_send_notification_issue_status_resolved(self):
|
|
mock_jira, mock_jira_obj, result = self._notify(TestJira.default_transitions_value,
|
|
TestJira.issue_status_resolved,
|
|
TestJira.default_address)
|
|
self.assertTrue(result)
|
|
mock_jira_obj.create_issue.assert_not_called()
|
|
self.assertEqual(
|
|
"project=MyProject and reporter='username' and summary ~ '0'",
|
|
mock_jira_obj.search_issues.call_args[0][0])
|
|
self.assertEqual(100, mock_jira_obj.transition_issue.call_args[0][1])
|
|
|
|
def test_send_notification_issue_status_closed(self):
|
|
issue_status = 'closed'
|
|
mock_jira, mock_jira_obj, result = self._notify(TestJira.default_transitions_value,
|
|
issue_status,
|
|
TestJira.default_address)
|
|
self.assertTrue(result)
|
|
mock_jira_obj.create_issue.assert_not_called()
|
|
self.assertEqual(
|
|
"project=MyProject and reporter='username' and summary ~ '0'",
|
|
mock_jira_obj.search_issues.call_args[0][0])
|
|
self.assertEqual(100, mock_jira_obj.transition_issue.call_args[0][1])
|
|
|
|
def test_send_notification_issue_status_progress(self):
|
|
issue_status = 'progress'
|
|
mock_jira, mock_jira_obj, result = self._notify(TestJira.default_transitions_value,
|
|
issue_status,
|
|
TestJira.default_address)
|
|
self.assertTrue(result)
|
|
mock_jira_obj.create_issue.assert_not_called()
|
|
mock_jira_obj.transitions.assert_not_called()
|
|
|
|
def test_send_notification_not_allow_transitions(self):
|
|
transitions_value = [[{'id': 100, 'name': 'not open'}]]
|
|
mock_jira, mock_jira_obj, result = self._notify(transitions_value,
|
|
TestJira.issue_status_resolved,
|
|
TestJira.default_address)
|
|
self.assertTrue(result)
|
|
mock_jira_obj.create_issue.assert_not_called()
|
|
mock_jira_obj.transition_issue.assert_not_called()
|
|
|
|
def test_send_notification_without_issue_list(self):
|
|
issue_status = None
|
|
mock_jira, mock_jira_obj, result = self._notify(TestJira.default_transitions_value,
|
|
issue_status,
|
|
TestJira.default_address)
|
|
self.assertTrue(result)
|
|
mock_jira_obj.transitions.assert_not_called()
|
|
self.assertEqual(issue(), mock_jira_obj.create_issue.call_args[1])
|
|
|
|
def test_send_notification_without_componet(self):
|
|
issue_status = None
|
|
address = 'http://test.jira:3333/?project=MyProject'
|
|
mock_jira, mock_jira_obj, result = self._notify(TestJira.default_transitions_value,
|
|
issue_status,
|
|
address)
|
|
self.assertTrue(result)
|
|
self.assertEqual(issue(component=False), mock_jira_obj.create_issue.call_args[1])
|
|
mock_jira_obj.transitions.assert_not_called()
|
|
|
|
def test_send_notification_create_jira_object_args(self):
|
|
mock_jira, mock_jira_obj, result = self._notify(TestJira.default_transitions_value,
|
|
TestJira.issue_status_resolved,
|
|
TestJira.default_address)
|
|
self.assertEqual('http://test.jira:3333/', mock_jira.JIRA.call_args[0][0])
|
|
self.assertEqual(('username', 'password'), mock_jira.JIRA.call_args[1].get('basic_auth'))
|
|
self.assertEqual(None, mock_jira.JIRA.call_args[1].get('proxies'))
|
|
|
|
def test_send_notification_with_proxy(self):
|
|
self.conf_override(
|
|
proxy='http://yourid:password@proxyserver:8080',
|
|
group='jira_notifier'
|
|
)
|
|
mock_jira, mock_jira_obj, result = self._notify(TestJira.default_transitions_value,
|
|
TestJira.issue_status_resolved,
|
|
TestJira.default_address)
|
|
self.assertTrue(result)
|
|
self.assertEqual({'https': 'http://yourid:password@proxyserver:8080'},
|
|
mock_jira.JIRA.call_args[1].get('proxies'))
|
|
|
|
def test_send_notification_custom_config_success(self):
|
|
issue_status = None
|
|
self.conf_override(
|
|
custom_formatter='tests/resources/test_jiraformat.yml',
|
|
group='jira_notifier'
|
|
)
|
|
mock_jira, mock_jira_obj, result = self._notify(TestJira.default_transitions_value,
|
|
issue_status,
|
|
TestJira.default_address)
|
|
self.assertTrue(result)
|
|
mock_jira_obj.transitions.assert_not_called()
|
|
self.assertEqual(issue(custom_config=True), mock_jira_obj.create_issue.call_args[1])
|
|
|
|
def test_send_notification_custom_config_failed(self):
|
|
self.conf_override(
|
|
custom_formatter='tests/resources/test_jiraformat_without_summary.yml',
|
|
group='jira_notifier'
|
|
)
|
|
mock_jira, mock_jira_obj, result = self._notify(TestJira.default_transitions_value,
|
|
TestJira.issue_status_resolved,
|
|
TestJira.default_address)
|
|
self.assertFalse(result)
|
|
|
|
def test_send_notification_custom_config_without_comments(self):
|
|
self.conf_override(
|
|
custom_formatter='tests/resources/test_jiraformat_without_comments.yml',
|
|
group='jira_notifier'
|
|
)
|
|
mock_jira, mock_jira_obj, result = self._notify(TestJira.default_transitions_value,
|
|
TestJira.issue_status_resolved,
|
|
TestJira.default_address)
|
|
self.assertTrue(result)
|
|
mock_jira_obj.add_comment.assert_not_called()
|
|
|
|
def test_send_notification_custom_config_exception(self):
|
|
self.conf_override(
|
|
custom_formatter='tests/resources/not_exist_file.yml',
|
|
group='jira_notifier'
|
|
)
|
|
self.assertRaises(Exception, self._notify,
|
|
TestJira.default_transitions_value,
|
|
TestJira.issue_status_resolved,
|
|
TestJira.default_address)
|
|
|
|
def test_type(self):
|
|
self.assertEqual('jira', self._jr.type)
|
|
|
|
def test_statsd_name(self):
|
|
self.assertEqual('jira_notifier', self._jr.statsd_name)
|