# 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 from oslotest import base import six from monasca_notification import notification as m_notification from monasca_notification.plugins import jira_notifier 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() 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.BaseTestCase): 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() 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) self._jira_config = {'user': 'username', 'password': 'password'} @mock.patch('monasca_notification.plugins.jira_notifier.jira') def _notify(self, transitions_value, issue_status, jira_config, address, mock_jira): self._jr.config(jira_config) 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, self._jira_config, 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, self._jira_config, 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, self._jira_config, 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, self._jira_config, 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, self._jira_config, 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, self._jira_config, 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, self._jira_config, 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): jira_config = self._jira_config jira_config.update({'proxy': 'http://yourid:password@proxyserver:8080'}) mock_jira, mock_jira_obj, result = self._notify(TestJira.default_transitions_value, TestJira.issue_status_resolved, jira_config, 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 jira_config = self._jira_config jira_config.update( {'custom_formatter': 'tests/resources/test_jiraformat.yml'}) mock_jira, mock_jira_obj, result = self._notify(TestJira.default_transitions_value, issue_status, jira_config, 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): jira_config = self._jira_config jira_config.update( {'custom_formatter': 'tests/resources/test_jiraformat_without_summary.yml'}) mock_jira, mock_jira_obj, result = self._notify(TestJira.default_transitions_value, TestJira.issue_status_resolved, jira_config, TestJira.default_address) self.assertFalse(result) def test_send_notification_custom_config_without_comments(self): jira_config = self._jira_config jira_config.update( {'custom_formatter': 'tests/resources/test_jiraformat_without_comments.yml'}) mock_jira, mock_jira_obj, result = self._notify(TestJira.default_transitions_value, TestJira.issue_status_resolved, jira_config, TestJira.default_address) self.assertTrue(result) mock_jira_obj.add_comment.assert_not_called() def test_send_notification_custom_config_exception(self): jira_config = self._jira_config jira_config.update({'custom_formatter': 'tests/resources/not_exist_file.yml'}) self.assertRaises(Exception, self._notify, TestJira.default_transitions_value, TestJira.issue_status_resolved, jira_config, 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) def test_config_exception(self): self.assertRaises(Exception, self._jr.config, {})