| # Copyright 2016 The Chromium Authors. All rights reserved. |
| # Use of this source code is governed 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.""" |
| from __future__ import print_function |
| from __future__ import division |
| from __future__ import absolute_import |
| |
| import unittest |
| |
| try: |
| from mox3 import mox |
| except ImportError: |
| import mox |
| import time |
| |
| from google.appengine.api import mail |
| |
| import settings |
| from businesslogic import work_env |
| from features import alert2issue |
| from features import commitlogcommands |
| from features import inboundemail |
| from framework import authdata |
| from framework import emailfmt |
| from framework import monorailcontext |
| 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 |
| from tracker import tracker_helpers |
| |
| |
| class InboundEmailTest(unittest.TestCase): |
| def setUp(self): |
| self.cnxn = 'fake cnxn' |
| self.services = service_manager.Services( |
| config=fake.ConfigService(), |
| issue=fake.IssueService(), |
| user=fake.UserService(), |
| usergroup=fake.UserGroupService(), |
| project=fake.ProjectService()) |
| self.project = self.services.project.TestAddProject( |
| 'proj', project_id=987, process_inbound_email=True, |
| contrib_ids=[111]) |
| 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!') |
| |
| self.inbound = inboundemail.InboundEmail(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.items(): |
| 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.assertEqual(1, len(email_tasks)) |
| email_task = email_tasks[0] |
| self.assertEqual('user@example.com', email_task['to']) |
| self.assertEqual('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_IssueUnidentified(self): |
| self.mox.StubOutWithMock(emailfmt, 'IdentifyProjectVerbAndLabel') |
| emailfmt.IdentifyProjectVerbAndLabel(self.project_addr).AndReturn(('proj', |
| None, None)) |
| |
| self.mox.StubOutWithMock(emailfmt, 'IdentifyIssue') |
| emailfmt.IdentifyIssue('proj', mox.IgnoreArg()).AndReturn((None)) |
| |
| self.mox.ReplayAll() |
| |
| ret = self.inbound.ProcessMail(self.msg, self.project_addr) |
| self.mox.VerifyAll() |
| self.assertIsNone(ret) |
| |
| def testProcessMail_ProjectNotLive(self): |
| self.services.user.TestAddUser('user@example.com', 111) |
| self.project.state = project_pb2.ProjectState.DELETABLE |
| email_tasks = self.inbound.ProcessMail(self.msg, self.project_addr) |
| email_task = email_tasks[0] |
| self.assertEqual('user@example.com', email_task['to']) |
| self.assertEqual('Project not found', email_task['subject']) |
| |
| def testProcessMail_ProjectInboundEmailDisabled(self): |
| self.services.user.TestAddUser('user@example.com', 111) |
| self.project.process_inbound_email = False |
| email_tasks = self.inbound.ProcessMail(self.msg, self.project_addr) |
| email_task = email_tasks[0] |
| self.assertEqual('user@example.com', email_task['to']) |
| self.assertEqual( |
| 'Email replies are not enabled in project proj', email_task['subject']) |
| |
| def testProcessMail_NoRefHeader(self): |
| self.services.user.TestAddUser('user@example.com', 111) |
| self.mox.StubOutWithMock(emailfmt, 'ValidateReferencesHeader') |
| emailfmt.ValidateReferencesHeader( |
| mox.IgnoreArg(), self.project, mox.IgnoreArg(), |
| mox.IgnoreArg()).AndReturn(False) |
| 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.assertEqual(1, len(email_tasks)) |
| email_task = email_tasks[0] |
| self.assertEqual('user@example.com', email_task['to']) |
| self.assertEqual( |
| 'Your message is not a reply to a notification email', |
| email_task['subject']) |
| |
| def testProcessMail_NoAccount(self): |
| # Note: not calling TestAddUser(). |
| email_tasks = self.inbound.ProcessMail(self.msg, self.project_addr) |
| self.mox.VerifyAll() |
| self.assertEqual(1, len(email_tasks)) |
| email_task = email_tasks[0] |
| self.assertEqual('user@example.com', email_task['to']) |
| self.assertEqual( |
| 'Could not determine account of sender', email_task['subject']) |
| |
| def testProcessMail_BannedAccount(self): |
| user_pb = self.services.user.TestAddUser('user@example.com', 111) |
| user_pb.banned = 'banned' |
| |
| 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.assertEqual(1, len(email_tasks)) |
| email_task = email_tasks[0] |
| self.assertEqual('user@example.com', email_task['to']) |
| self.assertEqual( |
| 'You are banned from using this issue tracker', email_task['subject']) |
| |
| def testProcessMail_Success(self): |
| self.services.user.TestAddUser('user@example.com', 111) |
| |
| self.mox.StubOutWithMock(emailfmt, 'ValidateReferencesHeader') |
| emailfmt.ValidateReferencesHeader( |
| mox.IgnoreArg(), self.project, mox.IgnoreArg(), |
| mox.IgnoreArg()).AndReturn(True) |
| |
| self.mox.StubOutWithMock(self.inbound, 'ProcessIssueReply') |
| self.inbound.ProcessIssueReply( |
| mox.IgnoreArg(), self.project, 123, self.project_addr, |
| 'awesome!') |
| |
| self.mox.ReplayAll() |
| |
| ret = self.inbound.ProcessMail(self.msg, self.project_addr) |
| self.mox.VerifyAll() |
| self.assertIsNone(ret) |
| |
| def testProcessMail_Success_with_AlertNotification(self): |
| """Test ProcessMail with an alert notification message. |
| |
| This is a sanity check for alert2issue.ProcessEmailNotification to ensure |
| that it can be successfully invoked in ProcessMail. Each function of |
| alert2issue module should be tested in aler2issue_test. |
| """ |
| project_name = self.project.project_name |
| verb = 'alert' |
| trooper_queue = 'my-trooper' |
| project_addr = '%s+%s+%s@example.com' % (project_name, verb, trooper_queue) |
| |
| self.mox.StubOutWithMock(emailfmt, 'IsProjectAddressOnToLine') |
| emailfmt.IsProjectAddressOnToLine( |
| project_addr, mox.IgnoreArg()).AndReturn(True) |
| |
| class MockAuthData(object): |
| def __init__(self): |
| self.user_pb = user_pb2.MakeUser(111) |
| self.effective_ids = set([1, 2, 3]) |
| self.user_id = 111 |
| self.email = 'user@example.com' |
| |
| mock_auth_data = MockAuthData() |
| self.mox.StubOutWithMock(authdata.AuthData, 'FromEmail') |
| authdata.AuthData.FromEmail( |
| mox.IgnoreArg(), settings.alert_service_account, self.services, |
| autocreate=True).AndReturn(mock_auth_data) |
| |
| self.mox.StubOutWithMock(alert2issue, 'ProcessEmailNotification') |
| alert2issue.ProcessEmailNotification( |
| self.services, mox.IgnoreArg(), self.project, project_addr, |
| mox.IgnoreArg(), mock_auth_data, mox.IgnoreArg(), 'awesome!', '', |
| self.msg, trooper_queue) |
| |
| self.mox.ReplayAll() |
| ret = self.inbound.ProcessMail(self.msg, project_addr) |
| self.mox.VerifyAll() |
| self.assertIsNone(ret) |
| |
| def testProcessIssueReply_NoIssue(self): |
| nonexistant_local_id = 200 |
| mc = monorailcontext.MonorailContext( |
| self.services, cnxn=self.cnxn, requester='user@example.com') |
| mc.LookupLoggedInUserPerms(self.project) |
| |
| email_tasks = self.inbound.ProcessIssueReply( |
| mc, self.project, nonexistant_local_id, self.project_addr, |
| 'awesome!') |
| self.assertEqual(1, len(email_tasks)) |
| email_task = email_tasks[0] |
| self.assertEqual('user@example.com', email_task['to']) |
| self.assertEqual( |
| '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 |
| mc = monorailcontext.MonorailContext( |
| self.services, cnxn=self.cnxn, requester='user@example.com') |
| mc.LookupLoggedInUserPerms(self.project) |
| |
| email_tasks = self.inbound.ProcessIssueReply( |
| mc, self.project, self.issue.local_id, self.project_addr, |
| 'awesome!') |
| self.assertEqual(1, len(email_tasks)) |
| email_task = email_tasks[0] |
| self.assertEqual('user@example.com', email_task['to']) |
| self.assertEqual( |
| 'Could not find issue %d in project %s' % |
| (self.issue.local_id, self.project.project_name), email_task['subject']) |
| |
| def VerifyUserHasNoPerm(self, perms): |
| mc = monorailcontext.MonorailContext( |
| self.services, cnxn=self.cnxn, requester='user@example.com') |
| mc.perms = perms |
| |
| email_tasks = self.inbound.ProcessIssueReply( |
| mc, self.project, self.issue.local_id, self.project_addr, |
| 'awesome!') |
| self.assertEqual(1, len(email_tasks)) |
| email_task = email_tasks[0] |
| self.assertEqual('user@example.com', email_task['to']) |
| self.assertEqual( |
| 'User does not have permission to add a comment', email_task['subject']) |
| |
| def testProcessIssueReply_NoViewPerm(self): |
| self.VerifyUserHasNoPerm(permissions.EMPTY_PERMISSIONSET) |
| |
| def testProcessIssueReply_CantViewRestrictedIssue(self): |
| self.issue.labels.append('Restrict-View-CoreTeam') |
| self.VerifyUserHasNoPerm(permissions.USER_PERMISSIONSET) |
| |
| def testProcessIssueReply_NoAddIssuePerm(self): |
| self.VerifyUserHasNoPerm(permissions.READ_ONLY_PERMISSIONSET) |
| |
| def testProcessIssueReply_NoEditIssuePerm(self): |
| self.services.user.TestAddUser('user@example.com', 111) |
| mc = monorailcontext.MonorailContext( |
| self.services, cnxn=self.cnxn, requester='user@example.com') |
| mc.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, 111, ['awesome!'], self.services, |
| strip_quoted_lines=True) |
| self.mox.StubOutWithMock(mock_uia, 'Run') |
| # mc.perms does not contain permission EDIT_ISSUE. |
| mock_uia.Run(mc, self.services) |
| |
| self.mox.ReplayAll() |
| ret = self.inbound.ProcessIssueReply( |
| mc, self.project, self.issue.local_id, self.project_addr, |
| 'awesome!') |
| self.mox.VerifyAll() |
| self.assertIsNone(ret) |
| |
| def testProcessIssueReply_Success(self): |
| self.services.user.TestAddUser('user@example.com', 111) |
| mc = monorailcontext.MonorailContext( |
| self.services, cnxn=self.cnxn, requester='user@example.com') |
| mc.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, 111, ['awesome!'], self.services, |
| strip_quoted_lines=True) |
| self.mox.StubOutWithMock(mock_uia, 'Run') |
| mock_uia.Run(mc, self.services) |
| |
| self.mox.ReplayAll() |
| ret = self.inbound.ProcessIssueReply( |
| mc, self.project, self.issue.local_id, self.project_addr, |
| 'awesome!') |
| self.mox.VerifyAll() |
| self.assertIsNone(ret) |
| |
| |
| class BouncedEmailTest(unittest.TestCase): |
| |
| def setUp(self): |
| self.cnxn = 'fake cnxn' |
| self.services = service_manager.Services( |
| user=fake.UserService()) |
| self.user = self.services.user.TestAddUser('user@example.com', 111) |
| |
| self.servlet = inboundemail.BouncedEmail(self.services) |
| self.mox = mox.Mox() |
| |
| def tearDown(self): |
| self.mox.UnsetStubs() |
| self.mox.ResetAll() |
| |
| def testReceive_Normal(self): |
| """Find the user that bounced and set email_bounce_timestamp.""" |
| self.assertEqual(0, self.user.email_bounce_timestamp) |
| |
| bounce_message = testing_helpers.Blank(original={'to': 'user@example.com'}) |
| self.servlet.receive(bounce_message) |
| |
| self.assertNotEqual(0, self.user.email_bounce_timestamp) |
| |
| def testReceive_NoSuchUser(self): |
| """When not found, log it and ignore without creating a user record.""" |
| bounce_message = testing_helpers.Blank( |
| original={'to': 'nope@example.com'}, |
| notification='notification') |
| self.servlet.receive(bounce_message) |
| self.assertEqual(1, len(self.services.user.users_by_id)) |