"""Unit tests for the evergreen_task_timeout script.""" import logging import unittest from http import HTTPStatus from http.client import HTTPConnection import requests.exceptions from jira import JIRAError from mock import MagicMock, patch, PropertyMock from buildscripts.validate_commit_message import Status, jira_client, \ configure_logging, validate_public_ticket, get_full_message, GIT_SHOW_COMMAND, \ validate_message, handle_violations, Violation, validate_ticket, find_ticket, main # pylint: disable=missing-docstring,no-self-use from buildscripts.validate_commit_message import find_matching_pattern, VALID_PATTERNS AUTHOR = 'jim' INVALID_MESSAGES = [ [""], # You must provide a message ["RevertEVG-1"], # revert and ticket must be formatted ["revert EVG-1"], # revert must be capitalized ["This is not a valid message"], # message must be valid ["Fix lint plus extras is not a valid message"], # Fix lint is strict ] NS = "buildscripts.validate_commit_message" def ns(relative_name): # pylint: disable=invalid-name """Return a full name from a name relative to the tested module's name space.""" return NS + "." + relative_name class GetFullMessageTest(unittest.TestCase): def test_no_message(self): with patch(ns('subprocess')) as mock_subprocess: expected = "output" mock_subprocess.check_output.return_value = bytearray(expected, "utf-8") self.assertEqual(expected, get_full_message([])) mock_subprocess.check_output.assert_called_once_with(GIT_SHOW_COMMAND) def test_empty_message(self): with patch(ns('subprocess')) as mock_subprocess: # blank / empty string is false self.assertFalse(get_full_message([""])) mock_subprocess.check_output.assert_not_called() def test_message(self): with patch(ns('subprocess')) as mock_subprocess: expected = "this is a test" self.assertEqual(expected, get_full_message(expected.split())) mock_subprocess.check_output.assert_not_called() class FindTicketTest(unittest.TestCase): def test_last_git_commit_success(self): messages = [ _make_expected("Fix lint", lint='Fix lint', public=True), _make_expected( 'SERVER-44612 ' 'this is the message\n\n' '(cherry picked from commit beef)', revert=None, ticket='SERVER-44612', backport='(cherry picked from commit beef)', public=True), _make_expected('SERVER-45074 this is a test (cherry picked from', revert=None, ticket='SERVER-45074', backport='(cherry picked from', public=True), _make_expected('SERVER-45074 this is a test (cherry picked from commit beef)', revert=None, ticket='SERVER-45074', backport='(cherry picked from commit beef)', public=True), _make_expected('SERVER-20', revert=None, ticket='SERVER-20', backport=None, public=True), _make_expected('WT-300', revert=None, ticket='WT-300', backport=None, public=True), _make_expected('SERVER-44338', revert=None, ticket='SERVER-44338', backport=None, public=True), _make_expected('Revert EVG-5', revert='Revert', ticket='EVG-5', backport=None, public=True), _make_expected('Revert SERVER-60', revert='Revert', ticket='SERVER-60', backport=None, public=True), _make_expected('Revert WT-700', revert='Revert', ticket='WT-700', backport=None, public=True), _make_expected('Revert \'SERVER-8000', revert='Revert', ticket='SERVER-8000', backport=None, public=True), _make_expected('Revert \'SERVER-8000\'', revert='Revert', ticket='SERVER-8000', backport=None, public=True), _make_expected('Revert "SERVER-90000"', revert='Revert', ticket='SERVER-90000', backport=None, public=True), _make_expected('Revert "SERVER-90000"', revert='Revert', ticket='SERVER-90000', backport=None, public=True), _make_expected( 'Import wiredtiger: 58115abb6fbb3c1cc7bfd087d41a47347bce9a69 from ' 'branch mongodb-4.4', imported='Import wiredtiger: 58115abb6fbb3c1cc7bfd087d41a47347bce9a69' ' from branch mongodb-4.4', public=True), _make_expected( 'Import tools: 58115abb6fbb3c1cc7bfd087d41a47347bce9a69 from ' 'branch mongodb-4.4', imported='Import tools: 58115abb6fbb3c1cc7bfd087d41a47347bce9a69' ' from branch mongodb-4.4', public=True), _make_expected('PRIVATE-1 this is a test', ticket='PRIVATE-1', public=False), _make_expected('this is a test'), ] self.assertTrue(all(find_ticket(message) == expected for message, expected in messages)) class FindMatchingPatternTest(unittest.TestCase): def test_last_git_commit_success(self): messages = [ _make_expected("Fix lint", lint='Fix lint'), _make_expected( 'SERVER-44612 ' 'this is the message\n\n' '(cherry picked from commit beef)', revert=None, ticket='SERVER-44612', backport='(cherry picked from commit beef)'), _make_expected('SERVER-45074 this is a test (cherry picked from', revert=None, ticket='SERVER-45074', backport='(cherry picked from'), _make_expected('SERVER-45074 this is a test (cherry picked from commit beef)', revert=None, ticket='SERVER-45074', backport='(cherry picked from commit beef)'), _make_expected('SERVER-20', revert=None, ticket='SERVER-20', backport=None), _make_expected('WT-300', revert=None, ticket='WT-300', backport=None), _make_expected('SERVER-44338', revert=None, ticket='SERVER-44338', backport=None), _make_expected('Revert EVG-5', revert='Revert', ticket='EVG-5', backport=None), _make_expected('Revert SERVER-60', revert='Revert', ticket='SERVER-60', backport=None), _make_expected('Revert WT-700', revert='Revert', ticket='WT-700', backport=None), _make_expected('Revert \'SERVER-8000', revert='Revert', ticket='SERVER-8000', backport=None), _make_expected('Revert \'SERVER-8000\'', revert='Revert', ticket='SERVER-8000', backport=None), _make_expected('Revert "SERVER-90000"', revert='Revert', ticket='SERVER-90000', backport=None), _make_expected('Revert "SERVER-90000"', revert='Revert', ticket='SERVER-90000', backport=None), _make_expected( 'Import wiredtiger: 58115abb6fbb3c1cc7bfd087d41a47347bce9a69 from ' 'branch mongodb-4.4', imported='Import wiredtiger: 58115abb6fbb3c1cc7bfd087d41a47347bce9a69' ' from branch mongodb-4.4'), _make_expected( 'Import tools: 58115abb6fbb3c1cc7bfd087d41a47347bce9a69 from ' 'branch mongodb-4.4', imported='Import tools: 58115abb6fbb3c1cc7bfd087d41a47347bce9a69' ' from branch mongodb-4.4'), ] self.assertTrue( all( find_matching_pattern(message, VALID_PATTERNS) == expected for message, expected in messages)) def _mock_issue(author, status='open', display="Jim O'Leary"): mock_assignee = MagicMock(name='assignee', displayName=display) type(mock_assignee).name = PropertyMock(return_value=author) mock_fields = MagicMock(name='fields', status=status, assignee=mock_assignee) mock_issue = MagicMock(name='issue', fields=mock_fields) return mock_issue class ValidateMessageTest(unittest.TestCase): def _test_empty_message(self, message=''): mock_jira = MagicMock(name='jira') ticket, validations = validate_message(message, AUTHOR, mock_jira) self.assertFalse(ticket) self.assertEqual(1, len(validations)) self.assertEqual(Status.ERROR, validations[0].status) self.assertIn('empty commit', validations[0].message) def test_empty_message(self): self._test_empty_message('') def test_blank_message(self): self._test_empty_message(' ') def test_validate(self): mock_jira = MagicMock(name='jira') message = 'message' ticket = {} validations = [] with patch(ns('find_ticket'), return_value=ticket) as mock_find_ticket, \ patch(ns('validate_ticket'), return_value=validations) as mock_validate_ticket: actual_ticket, actual_validations = validate_message(message, AUTHOR, mock_jira) self.assertEqual(actual_ticket, ticket) self.assertEqual(actual_validations, validations) mock_find_ticket.assert_called_once_with(message) mock_validate_ticket.assert_called_once_with(ticket, AUTHOR, mock_jira) class ValidateTicketTest(unittest.TestCase): def test_empty_ticket(self): mock_jira = MagicMock(name='jira') ticket = {} validations = validate_ticket(ticket, AUTHOR, mock_jira) self.assertEqual(1, len(validations)) self.assertEqual(Status.WARNING, validations[0].status) self.assertIn('commit without a ticket', validations[0].message) def test_public_ticket(self): mock_jira = MagicMock(name='jira') ticket = dict(public=True) expected = [] with patch(ns('validate_public_ticket'), return_value=expected) as mock_validate_public_ticket: self.assertEqual(expected, validate_ticket(ticket, AUTHOR, mock_jira)) mock_validate_public_ticket.assert_called_once_with(ticket, AUTHOR, mock_jira) def test_private_ticket(self): mock_jira = MagicMock(name='jira') ticket = dict(public=False, ticket='PRIVATE-1') expected = [] with patch(ns('validate_public_ticket'), return_value=expected): validations = validate_ticket(ticket, AUTHOR, mock_jira) self.assertEqual(1, len(validations)) self.assertEqual(Status.ERROR, validations[0].status) self.assertIn('private project', validations[0].message) class ValidatePublicTicketTest(unittest.TestCase): def test_no_issue(self): mock_jira = MagicMock(name='jira') mock_jira.issue.return_value = None ticket = dict(ticket='SERVER-45074') faults = validate_public_ticket(ticket, AUTHOR, mock_jira) self.assertEqual(len(faults), 1) self.assertEqual(faults[0].status, Status.WARNING) def test_closed(self): mock_jira = MagicMock(name='jira') mock_jira.issue.return_value = _mock_issue(AUTHOR, status='closed') ticket = dict(ticket='SERVER-45074') faults = validate_public_ticket(ticket, AUTHOR, mock_jira) self.assertEqual(len(faults), 1) self.assertEqual(faults[0].status, Status.ERROR) self.assertIn('status cannot be', faults[0].message) def test_different_author(self): mock_jira = MagicMock(name='jira') mock_jira.issue.return_value = _mock_issue('not ' + AUTHOR) ticket = dict(ticket='SERVER-45074') faults = validate_public_ticket(ticket, AUTHOR, mock_jira) self.assertEqual(len(faults), 1) self.assertEqual(faults[0].status, Status.WARNING) self.assertIn('assignee is not author', faults[0].message) def test_multiple(self): mock_jira = MagicMock(name='jira') mock_jira.issue.return_value = _mock_issue('not ' + AUTHOR, status='closed') ticket = dict(ticket='SERVER-45074') faults = validate_public_ticket(ticket, AUTHOR, mock_jira) self.assertEqual(len(faults), 2) self.assertEqual(len([fault for fault in faults if fault.status == Status.WARNING]), 1) self.assertEqual(len([fault for fault in faults if fault.status == Status.ERROR]), 1) self.assertTrue(any(['assignee is not author' in fault.message for fault in faults])) self.assertTrue(any(['status cannot be' in fault.message for fault in faults])) def test_backport_not_author(self): mock_jira = MagicMock(name='jira') mock_jira.issue.return_value = _mock_issue('not ' + AUTHOR) ticket = dict(ticket='SERVER-45074', backport='(cherry picled from X)') faults = validate_public_ticket(ticket, AUTHOR, mock_jira) self.assertFalse(faults) def test_backport_author(self): mock_jira = MagicMock(name='jira') mock_jira.issue.return_value = _mock_issue(AUTHOR) ticket = dict(ticket='SERVER-45074', backport='(cherry picled from X)') faults = validate_public_ticket(ticket, AUTHOR, mock_jira) self.assertFalse(faults) def test_connection_exception(self): mock_jira = MagicMock(name='jira') mock_jira.issue.side_effect = requests.exceptions.ConnectionError('boom') ticket = dict(ticket='SERVER-45074') faults = validate_public_ticket(ticket, AUTHOR, mock_jira) self.assertEqual(1, len(faults)) self.assertTrue(faults[0].status == Status.WARNING) self.assertIn('unexpected connection exception', faults[0].message) def test_not_found(self): mock_jira = MagicMock(name='jira') mock_jira.issue.side_effect = JIRAError(status_code=HTTPStatus.NOT_FOUND) ticket = dict(ticket='SERVER-45074') faults = validate_public_ticket(ticket, AUTHOR, mock_jira) self.assertEqual(1, len(faults)) self.assertTrue(faults[0].status == Status.ERROR) self.assertIn('not found', faults[0].message) def test_permission(self): mock_jira = MagicMock(name='jira') mock_jira.issue.side_effect = JIRAError(status_code=HTTPStatus.UNAUTHORIZED) ticket = dict(ticket='SERVER-45074') faults = validate_public_ticket(ticket, AUTHOR, mock_jira) self.assertEqual(1, len(faults)) self.assertTrue(faults[0].status == Status.ERROR) self.assertIn('private', faults[0].message) def test_other_jira_exception(self): mock_jira = MagicMock(name='jira') mock_jira.issue.side_effect = JIRAError() ticket = dict(ticket='SERVER-45074') faults = validate_public_ticket(ticket, AUTHOR, mock_jira) self.assertEqual(1, len(faults)) self.assertTrue(faults[0].status == Status.WARNING) self.assertIn('unexpected jira error', faults[0].message) class HandleValidationTest(unittest.TestCase): def test_empty(self): ticket = {} message = 'this is a test' validations = [] warnings_as_errors = False self.assertEqual(Status.OK, handle_violations(ticket, message, validations, warnings_as_errors)) def test_error(self): ticket = {} message = 'this is a test' validations = [Violation(Status.ERROR, 'private project: PRIVATE-1')] warnings_as_errors = False self.assertEqual(Status.ERROR, handle_violations(ticket, message, validations, warnings_as_errors)) def _test_warnings(self, validations=None, warnings_as_errors=False): ticket = {} message = 'this is a test' if validations is None: validations = [ Violation(Status.WARNING, 'found a commit without a ticket'), Violation(Status.WARNING, 'another warning') ] self.assertEqual(Status.OK if not warnings_as_errors else Status.ERROR, handle_violations(ticket, message, validations, warnings_as_errors)) def test_single_warning(self): self._test_warnings([Violation(Status.WARNING, 'found a commit without a ticket')]) def test_multiple_warnings(self): self._test_warnings() def test_warning_as_error(self): self._test_warnings(warnings_as_errors=True) def test_warnings_and_errors(self): ticket = {} message = 'this is a test' validations = [ Violation(Status.ERROR, 'private project: PRIVATE-1'), Violation(Status.WARNING, 'found a commit without a ticket') ] warnings_as_errors = False self.assertEqual(Status.ERROR, handle_violations(ticket, message, validations, warnings_as_errors)) def _make_expected(message, **kwargs): return message, dict(**kwargs) class JiraClientTest(unittest.TestCase): def _test_exception(self, exception): with patch(ns("JIRA"), side_effect=exception("boom")): self.assertIsNone(jira_client('server', verbose=False)) def test_connection_exceptions(self): """test connection exceptions.""" self._test_exception(requests.exceptions.ConnectionError) def test_jira_exception(self): """test jira exceptions.""" self._test_exception(JIRAError) def test_value_exception(self): """test ValueError exceptions.""" self._test_exception(ValueError) def test_exception(self): """test Exception exceptions.""" self._test_exception(Exception) def test_issue(self): mock_jira = MagicMock() # mock_jira.issue.return_value = 'issue' with patch(ns("JIRA"), return_value=mock_jira) as mock_ctor: jira = jira_client('server') mock_ctor.assert_called_once_with('server', logging=False) mock_jira.issue.assert_called_once_with('SERVER-1') self.assertEqual(mock_jira, jira) class ConfigureLoggingTest(unittest.TestCase): def _test_configure_formatter(self, formatter=None): mock_root_logger = MagicMock(name="root_logger") with patch.object(logging, 'basicConfig') as mock_basic_config, \ patch.object(logging, 'getLogger') as mock_get_logger: mock_get_logger.return_value = mock_root_logger if formatter is None: configure_logging(0) else: configure_logging(0, formatter=formatter) if formatter is None: expected = '%(levelname)s: %(message)s' else: expected = formatter mock_basic_config.assert_called_once_with(format=expected) mock_root_logger.setLevel.assert_called_once_with(logging.WARNING) def test_configure_default_formatter(self): self._test_configure_formatter() def test_configure_formatter(self): self._test_configure_formatter(formatter='format') def _test_configure_level(self, level, expected, debuglevel=0): old_debuglevel = HTTPConnection.debuglevel try: mock_root_logger = MagicMock(name="root_logger") with patch.object(logging, 'basicConfig'), \ patch.object(logging, 'getLogger') as mock_get_logger: mock_get_logger.return_value = mock_root_logger configure_logging(level) mock_root_logger.setLevel.assert_called_once_with(expected) self.assertEqual(HTTPConnection.debuglevel, debuglevel) finally: HTTPConnection.debuglevel = old_debuglevel def test_configure_level_0(self): self._test_configure_level(0, logging.WARNING) def test_configure_level_1(self): self._test_configure_level(1, logging.INFO) def test_configure_level_2(self): self._test_configure_level(2, logging.DEBUG, debuglevel=1) def test_configure_level_3(self): self._test_configure_level(3, logging.DEBUG, debuglevel=1) class MainTest(unittest.TestCase): def _test_helper(self, args=None, violations=None, expected=Status.OK): if violations is None: violations = [] if args is None: args = [] args = ['SERVER-45074 this is a test', '-a', AUTHOR] + args with patch(ns('jira_client')), patch(ns('validate_public_ticket'), return_value=violations): self.assertEqual(expected, main(args)) def test_no_violations(self): self._test_helper() def test_warning(self): violations = [Violation(Status.WARNING, 'just a warning')] self._test_helper(violations=violations) def test_warning_as_error(self): violations = [Violation(Status.WARNING, 'just a warning')] args = ['-W'] self._test_helper(args=args, violations=violations, expected=Status.ERROR) def test_error(self): violations = [Violation(Status.ERROR, 'just an error')] self._test_helper(violations=violations, expected=Status.ERROR) def test_no_author(self): args = ['SERVER-45074 this is a test'] with self.assertRaises(SystemExit) as cm: main(args) exception = cm.exception self.assertEqual(exception.code, 2) def test_author(self): args = ['SERVER-45074 this is a test', '-a', AUTHOR] with patch(ns('JIRA'), side_effect=requests.exceptions.ConnectionError): self.assertEqual(main(args), 0)