blob: cab9d0abdab735aa453952d24f638c52f0dea60b [file] [log] [blame]
# Copyright 2016 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""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 mrproto import project_pb2
from mrproto import tracker_pb2
from mrproto 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))