| # Copyright 2016 The Chromium Authors. All rights reserved. |
| # Use of this source code is govered by a BSD-style |
| # license that can be found in the LICENSE file or at |
| # https://developers.google.com/open-source/licenses/bsd |
| |
| """Unittests for monorail.feature.inboundemail.""" |
| |
| import unittest |
| |
| import mox |
| |
| from features import commitlogcommands |
| from features import inboundemail |
| from framework import emailfmt |
| from framework import monorailrequest |
| from framework import permissions |
| from proto import project_pb2 |
| from proto import tracker_pb2 |
| from proto import user_pb2 |
| from services import service_manager |
| from testing import fake |
| from testing import testing_helpers |
| |
| |
| class InboundEmailTest(unittest.TestCase): |
| |
| def setUp(self): |
| self.cnxn = 'fake cnxn' |
| self.services = service_manager.Services( |
| issue=fake.IssueService(), |
| user=fake.UserService(), |
| project=fake.ProjectService()) |
| self.project = self.services.project.TestAddProject( |
| 'proj', project_id=987, process_inbound_email=True) |
| self.project_addr = 'proj@monorail.example.com' |
| |
| self.issue = tracker_pb2.Issue() |
| self.issue.project_id = 987 |
| self.issue.local_id = 100 |
| self.services.issue.TestAddIssue(self.issue) |
| |
| self.msg = testing_helpers.MakeMessage( |
| testing_helpers.HEADER_LINES, 'awesome!') |
| |
| request, _ = testing_helpers.GetRequestObjects() |
| self.inbound = inboundemail.InboundEmail(request, None, self.services) |
| self.mox = mox.Mox() |
| |
| def tearDown(self): |
| self.mox.UnsetStubs() |
| self.mox.ResetAll() |
| |
| def testTemplates(self): |
| for name, template_path in self.inbound._templates.iteritems(): |
| assert(name in inboundemail.MSG_TEMPLATES) |
| assert( |
| template_path.GetTemplatePath().endswith( |
| inboundemail.MSG_TEMPLATES[name])) |
| |
| def testProcessMail_MsgTooBig(self): |
| self.mox.StubOutWithMock(emailfmt, 'IsBodyTooBigToParse') |
| emailfmt.IsBodyTooBigToParse(mox.IgnoreArg()).AndReturn(True) |
| self.mox.ReplayAll() |
| |
| email_tasks = self.inbound.ProcessMail(self.msg, self.project_addr) |
| self.mox.VerifyAll() |
| self.assertEquals(1, len(email_tasks)) |
| email_task = email_tasks[0] |
| self.assertEquals('user@example.com', email_task['to']) |
| self.assertEquals('Email body too long', email_task['subject']) |
| |
| def testProcessMail_NoProjectOnToLine(self): |
| self.mox.StubOutWithMock(emailfmt, 'IsProjectAddressOnToLine') |
| emailfmt.IsProjectAddressOnToLine( |
| self.project_addr, [self.project_addr]).AndReturn(False) |
| self.mox.ReplayAll() |
| |
| ret = self.inbound.ProcessMail(self.msg, self.project_addr) |
| self.mox.VerifyAll() |
| self.assertIsNone(ret) |
| |
| def testProcessMail_ProjectUnidentified(self): |
| self.mox.StubOutWithMock(emailfmt, 'IdentifyProjectAndIssue') |
| emailfmt.IdentifyProjectAndIssue( |
| self.project_addr, mox.IgnoreArg()).AndReturn((None, None)) |
| self.mox.ReplayAll() |
| |
| ret = self.inbound.ProcessMail(self.msg, self.project_addr) |
| self.mox.VerifyAll() |
| self.assertIsNone(ret) |
| |
| def testProcessMail_ProjectNotLive(self): |
| self.project.state = project_pb2.ProjectState.DELETABLE |
| email_tasks = self.inbound.ProcessMail(self.msg, self.project_addr) |
| email_task = email_tasks[0] |
| self.assertEquals('user@example.com', email_task['to']) |
| self.assertEquals('Project not found', email_task['subject']) |
| |
| def testProcessMail_ProjectInboundEmailDisabled(self): |
| self.project.process_inbound_email = False |
| email_tasks = self.inbound.ProcessMail(self.msg, self.project_addr) |
| email_task = email_tasks[0] |
| self.assertEquals('user@example.com', email_task['to']) |
| self.assertEquals('Email replies are not enabled in project proj', |
| email_task['subject']) |
| |
| def testProcessMail_NoRefHeader(self): |
| self.mox.StubOutWithMock(emailfmt, 'ValidateReferencesHeader') |
| emailfmt.ValidateReferencesHeader( |
| mox.IgnoreArg(), self.project, mox.IgnoreArg(), |
| mox.IgnoreArg()).AndReturn(False) |
| self.mox.ReplayAll() |
| |
| email_tasks = self.inbound.ProcessMail(self.msg, self.project_addr) |
| self.mox.VerifyAll() |
| self.assertEquals(1, len(email_tasks)) |
| email_task = email_tasks[0] |
| self.assertEquals('user@example.com', email_task['to']) |
| self.assertEquals('Your message is not a reply to a notification email', |
| email_task['subject']) |
| |
| def testProcessMail_NoAccount(self): |
| self.mox.StubOutWithMock(emailfmt, 'ValidateReferencesHeader') |
| emailfmt.ValidateReferencesHeader( |
| mox.IgnoreArg(), self.project, mox.IgnoreArg(), |
| mox.IgnoreArg()).AndReturn(True) |
| self.mox.ReplayAll() |
| |
| email_tasks = self.inbound.ProcessMail(self.msg, self.project_addr) |
| self.mox.VerifyAll() |
| self.assertEquals(1, len(email_tasks)) |
| email_task = email_tasks[0] |
| self.assertEquals('user@example.com', email_task['to']) |
| self.assertEquals('Could not determine account of sender', |
| email_task['subject']) |
| |
| def testProcessMail_BannedAccount(self): |
| self.services.user.TestAddUser('user@example.com', 111L) |
| class MockAuthData: |
| def __init__(self): |
| self.user_pb = user_pb2.MakeUser() |
| self.effective_ids = set([1, 2, 3]) |
| self.user_id = 111L |
| mock_auth_data = MockAuthData() |
| mock_auth_data.user_pb.banned = 'banned' |
| |
| self.mox.StubOutWithMock(emailfmt, 'ValidateReferencesHeader') |
| emailfmt.ValidateReferencesHeader( |
| mox.IgnoreArg(), self.project, mox.IgnoreArg(), |
| mox.IgnoreArg()).AndReturn(True) |
| self.mox.StubOutWithMock(monorailrequest.AuthData, 'FromEmail') |
| monorailrequest.AuthData.FromEmail( |
| mox.IgnoreArg(), 'user@example.com', self.services).AndReturn( |
| mock_auth_data) |
| self.mox.ReplayAll() |
| |
| email_tasks = self.inbound.ProcessMail(self.msg, self.project_addr) |
| self.mox.VerifyAll() |
| self.assertEquals(1, len(email_tasks)) |
| email_task = email_tasks[0] |
| self.assertEquals('user@example.com', email_task['to']) |
| self.assertEquals('You are banned from using this issue tracker', |
| email_task['subject']) |
| |
| def testProcessMail_Success(self): |
| self.services.user.TestAddUser('user@example.com', 111L) |
| class MockAuthData: |
| def __init__(self): |
| self.user_pb = user_pb2.MakeUser() |
| self.effective_ids = set([1, 2, 3]) |
| self.user_id = 111L |
| mock_auth_data = MockAuthData() |
| |
| self.mox.StubOutWithMock(emailfmt, 'ValidateReferencesHeader') |
| emailfmt.ValidateReferencesHeader( |
| mox.IgnoreArg(), self.project, mox.IgnoreArg(), |
| mox.IgnoreArg()).AndReturn(True) |
| |
| self.mox.StubOutWithMock(monorailrequest.AuthData, 'FromEmail') |
| monorailrequest.AuthData.FromEmail( |
| mox.IgnoreArg(), 'user@example.com', self.services).AndReturn( |
| mock_auth_data) |
| |
| self.mox.StubOutWithMock(permissions, 'GetPermissions') |
| permissions.GetPermissions( |
| mock_auth_data.user_pb, mock_auth_data.effective_ids, |
| self.project).AndReturn('test permissions') |
| |
| self.mox.StubOutWithMock(self.inbound, 'ProcessIssueReply') |
| self.inbound.ProcessIssueReply( |
| mox.IgnoreArg(), self.project, 123, self.project_addr, |
| 'user@example.com', 111L, mock_auth_data.effective_ids, |
| 'test permissions', 'awesome!') |
| |
| self.mox.ReplayAll() |
| |
| ret = self.inbound.ProcessMail(self.msg, self.project_addr) |
| self.mox.VerifyAll() |
| self.assertIsNone(ret) |
| |
| def testProcessIssueReply_NoIssue(self): |
| nonexistant_local_id = 200 |
| email_tasks = self.inbound.ProcessIssueReply( |
| self.cnxn, self.project, nonexistant_local_id, self.project_addr, |
| 'user@example.com', 111L, [1, 2, 3], permissions.USER_PERMISSIONSET, |
| 'awesome!') |
| self.assertEquals(1, len(email_tasks)) |
| email_task = email_tasks[0] |
| self.assertEquals('user@example.com', email_task['to']) |
| self.assertEquals('Could not find issue %d in project %s' % ( |
| nonexistant_local_id, self.project.project_name), |
| email_task['subject']) |
| |
| def testProcessIssueReply_DeletedIssue(self): |
| self.issue.deleted = True |
| email_tasks = self.inbound.ProcessIssueReply( |
| self.cnxn, self.project, self.issue.local_id, self.project_addr, |
| 'user@example.com', 111L, [1, 2, 3], permissions.USER_PERMISSIONSET, |
| 'awesome!') |
| self.assertEquals(1, len(email_tasks)) |
| email_task = email_tasks[0] |
| self.assertEquals('user@example.com', email_task['to']) |
| self.assertEquals('Could not find issue %d in project %s' % ( |
| self.issue.local_id, self.project.project_name), |
| email_task['subject']) |
| |
| def testProcessIssueReply_NoAddIssuePerm(self): |
| perms = permissions.READ_ONLY_PERMISSIONSET |
| email_tasks = self.inbound.ProcessIssueReply( |
| self.cnxn, self.project, self.issue.local_id, self.project_addr, |
| 'user@example.com', 111L, [1, 2, 3], perms, 'awesome!') |
| self.assertEquals(1, len(email_tasks)) |
| email_task = email_tasks[0] |
| self.assertEquals('user@example.com', email_task['to']) |
| self.assertEquals('User does not have permission to add a comment', |
| email_task['subject']) |
| |
| def testProcessIssueReply_NoEditIssuePerm(self): |
| perms = permissions.USER_PERMISSIONSET |
| mock_uia = commitlogcommands.UpdateIssueAction(self.issue.local_id) |
| |
| self.mox.StubOutWithMock(commitlogcommands, 'UpdateIssueAction') |
| commitlogcommands.UpdateIssueAction(self.issue.local_id).AndReturn(mock_uia) |
| |
| self.mox.StubOutWithMock(mock_uia, 'Parse') |
| mock_uia.Parse( |
| self.cnxn, self.project.project_name, 111L, ['awesome!'], self.services, |
| strip_quoted_lines=True) |
| self.mox.StubOutWithMock(mock_uia, 'Run') |
| # Allow edit is false here because the permission set does not contain |
| # EDIT_ISSUE. |
| mock_uia.Run(self.cnxn, self.services, allow_edit=False) |
| |
| self.mox.ReplayAll() |
| ret = self.inbound.ProcessIssueReply( |
| self.cnxn, self.project, self.issue.local_id, self.project_addr, |
| 'from_addr', 111L, [1, 2, 3], perms, 'awesome!') |
| self.mox.VerifyAll() |
| self.assertIsNone(ret) |
| |
| def testProcessIssueReply_Success(self): |
| perms = permissions.COMMITTER_ACTIVE_PERMISSIONSET |
| mock_uia = commitlogcommands.UpdateIssueAction(self.issue.local_id) |
| |
| self.mox.StubOutWithMock(commitlogcommands, 'UpdateIssueAction') |
| commitlogcommands.UpdateIssueAction(self.issue.local_id).AndReturn(mock_uia) |
| |
| self.mox.StubOutWithMock(mock_uia, 'Parse') |
| mock_uia.Parse( |
| self.cnxn, self.project.project_name, 111L, ['awesome!'], self.services, |
| strip_quoted_lines=True) |
| self.mox.StubOutWithMock(mock_uia, 'Run') |
| mock_uia.Run(self.cnxn, self.services, allow_edit=True) |
| |
| self.mox.ReplayAll() |
| ret = self.inbound.ProcessIssueReply( |
| self.cnxn, self.project, self.issue.local_id, self.project_addr, |
| 'from_addr', 111L, [1, 2, 3], perms, 'awesome!') |
| self.mox.VerifyAll() |
| self.assertIsNone(ret) |
