blob: e73488d21343600d67d096843ec22d02f79537cf [file] [log] [blame]
# 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
"""Tests for notify.py."""
from __future__ import print_function
from __future__ import division
from __future__ import absolute_import
import json
import mock
import unittest
import flask
from google.appengine.ext import testbed
from features import notify
from features import notify_reasons
from framework import emailfmt
from framework import urls
from proto import tracker_pb2
from services import service_manager
from testing import fake
from testing import testing_helpers
from tracker import attachment_helpers
from tracker import tracker_bizobj
def MakeTestIssue(project_id, local_id, owner_id, reporter_id, is_spam=False):
issue = tracker_pb2.Issue()
issue.project_id = project_id
issue.local_id = local_id
issue.issue_id = 1000 * project_id + local_id
issue.owner_id = owner_id
issue.reporter_id = reporter_id
issue.is_spam = is_spam
return issue
class NotifyTaskHandleRequestTest(unittest.TestCase):
def setUp(self):
self.services = service_manager.Services(
user=fake.UserService(),
usergroup=fake.UserGroupService(),
project=fake.ProjectService(),
config=fake.ConfigService(),
issue=fake.IssueService(),
issue_star=fake.IssueStarService(),
features=fake.FeaturesService())
self.requester = self.services.user.TestAddUser('requester@example.com', 1)
self.nonmember = self.services.user.TestAddUser('user@example.com', 2)
self.member = self.services.user.TestAddUser('member@example.com', 3)
self.project = self.services.project.TestAddProject(
'test-project', owner_ids=[1, 3], project_id=12345)
self.issue1 = MakeTestIssue(
project_id=12345, local_id=1, owner_id=2, reporter_id=1)
self.issue2 = MakeTestIssue(
project_id=12345, local_id=2, owner_id=2, reporter_id=1)
self.services.issue.TestAddIssue(self.issue1)
self.orig_sign_attachment_id = attachment_helpers.SignAttachmentID
attachment_helpers.SignAttachmentID = (
lambda aid: 'signed_%d' % aid)
self.servlet = notify.OutboundEmailTask(services=self.services)
self.app = flask.Flask('test_app')
self.app.config['TESTING'] = True
self.app.add_url_rule(
'/_task/outboundEmail.do',
view_func=self.servlet.PostOutboundEmailTask,
methods=['POST'])
self.testbed = testbed.Testbed()
self.testbed.activate()
self.testbed.init_memcache_stub()
self.testbed.init_datastore_v3_stub()
def tearDown(self):
attachment_helpers.SignAttachmentID = self.orig_sign_attachment_id
def get_filtered_task_call_args(self, create_task_mock, relative_uri):
return [
(args, kwargs)
for (args, kwargs) in create_task_mock.call_args_list
if args[0]['app_engine_http_request']['relative_uri'] == relative_uri
]
def VerifyParams(self, result, params):
self.assertEqual(
bool(params['send_email']), result['params']['send_email'])
if 'issue_id' in params:
self.assertEqual(params['issue_id'], result['params']['issue_id'])
if 'issue_ids' in params:
self.assertEqual([int(p) for p in params['issue_ids'].split(',')],
result['params']['issue_ids'])
def testNotifyIssueChangeTask_Normal(self):
task = notify.NotifyIssueChangeTask(services=self.services)
params = {'send_email': 1, 'issue_id': 12345001, 'seq': 0,
'commenter_id': 2}
mr = testing_helpers.MakeMonorailRequest(
user_info={'user_id': 1},
params=params,
method='POST',
services=self.services)
result = task.HandleRequest(mr)
self.VerifyParams(result, params)
@mock.patch('framework.cloud_tasks_helpers.create_task')
def testNotifyIssueChangeTask_Spam(self, _create_task_mock):
issue = MakeTestIssue(
project_id=12345, local_id=1, owner_id=1, reporter_id=1,
is_spam=True)
self.services.issue.TestAddIssue(issue)
task = notify.NotifyIssueChangeTask(services=self.services)
params = {'send_email': 0, 'issue_id': issue.issue_id, 'seq': 0,
'commenter_id': 2}
mr = testing_helpers.MakeMonorailRequest(
user_info={'user_id': 1},
params=params,
method='POST',
services=self.services)
result = task.HandleRequest(mr)
self.assertEqual(0, len(result['notified']))
@mock.patch('framework.cloud_tasks_helpers.create_task')
def testNotifyBlockingChangeTask_Normal(self, _create_task_mock):
issue2 = MakeTestIssue(
project_id=12345, local_id=2, owner_id=2, reporter_id=1)
self.services.issue.TestAddIssue(issue2)
task = notify.NotifyBlockingChangeTask(services=self.services)
params = {
'send_email': 1, 'issue_id': issue2.issue_id, 'seq': 0,
'delta_blocker_iids': self.issue1.issue_id, 'commenter_id': 1,
'hostport': 'bugs.chromium.org'}
mr = testing_helpers.MakeMonorailRequest(
user_info={'user_id': 1},
params=params,
method='POST',
services=self.services)
result = task.HandleRequest(mr)
self.VerifyParams(result, params)
def testNotifyBlockingChangeTask_Spam(self):
issue2 = MakeTestIssue(
project_id=12345, local_id=2, owner_id=2, reporter_id=1,
is_spam=True)
self.services.issue.TestAddIssue(issue2)
task = notify.NotifyBlockingChangeTask(services=self.services)
params = {
'send_email': 1, 'issue_id': issue2.issue_id, 'seq': 0,
'delta_blocker_iids': self.issue1.issue_id, 'commenter_id': 1}
mr = testing_helpers.MakeMonorailRequest(
user_info={'user_id': 1},
params=params,
method='POST',
services=self.services)
result = task.HandleRequest(mr)
self.assertEqual(0, len(result['notified']))
@mock.patch('framework.cloud_tasks_helpers.create_task')
def testNotifyBulkChangeTask_Normal(self, create_task_mock):
"""We generate email tasks for each user involved in the issues."""
issue2 = MakeTestIssue(
project_id=12345, local_id=2, owner_id=2, reporter_id=1)
issue2.cc_ids = [3]
self.services.issue.TestAddIssue(issue2)
task = notify.NotifyBulkChangeTask(services=self.services)
params = {
'send_email': 1, 'seq': 0,
'issue_ids': '%d,%d' % (self.issue1.issue_id, issue2.issue_id),
'old_owner_ids': '1,1', 'commenter_id': 1}
mr = testing_helpers.MakeMonorailRequest(
user_info={'user_id': 1},
params=params,
method='POST',
services=self.services)
result = task.HandleRequest(mr)
self.VerifyParams(result, params)
call_args_list = self.get_filtered_task_call_args(
create_task_mock, urls.OUTBOUND_EMAIL_TASK + '.do')
self.assertEqual(2, len(call_args_list))
for (args, _kwargs) in call_args_list:
task = args[0]
body = json.loads(task['app_engine_http_request']['body'].decode())
if 'user' in body['to']:
self.assertIn(u'\u2026', body['from_addr'])
# Full email for members
if 'member' in body['to']:
self.assertNotIn(u'\u2026', body['from_addr'])
@mock.patch('framework.cloud_tasks_helpers.create_task')
def testNotifyBulkChangeTask_AlsoNotify(self, create_task_mock):
"""We generate email tasks for also-notify addresses."""
self.issue1.derived_notify_addrs = [
'mailing-list@example.com', 'member@example.com']
task = notify.NotifyBulkChangeTask(services=self.services)
params = {
'send_email': 1, 'seq': 0,
'issue_ids': '%d' % (self.issue1.issue_id),
'old_owner_ids': '1', 'commenter_id': 1}
mr = testing_helpers.MakeMonorailRequest(
user_info={'user_id': 1},
params=params,
method='POST',
services=self.services)
result = task.HandleRequest(mr)
self.VerifyParams(result, params)
call_args_list = self.get_filtered_task_call_args(
create_task_mock, urls.OUTBOUND_EMAIL_TASK + '.do')
self.assertEqual(3, len(call_args_list))
self.assertItemsEqual(
['user@example.com', 'mailing-list@example.com', 'member@example.com'],
result['notified'])
for (args, _kwargs) in call_args_list:
task = args[0]
body = json.loads(task['app_engine_http_request']['body'].decode())
# obfuscated email for non-members
if 'user' in body['to']:
self.assertIn(u'\u2026', body['from_addr'])
# Full email for members
if 'member' in body['to']:
self.assertNotIn(u'\u2026', body['from_addr'])
@mock.patch('framework.cloud_tasks_helpers.create_task')
def testNotifyBulkChangeTask_ProjectNotify(self, create_task_mock):
"""We generate email tasks for project.issue_notify_address."""
self.project.issue_notify_address = 'mailing-list@example.com'
task = notify.NotifyBulkChangeTask(services=self.services)
params = {
'send_email': 1, 'seq': 0,
'issue_ids': '%d' % (self.issue1.issue_id),
'old_owner_ids': '1', 'commenter_id': 1}
mr = testing_helpers.MakeMonorailRequest(
user_info={'user_id': 1},
params=params,
method='POST',
services=self.services)
result = task.HandleRequest(mr)
self.VerifyParams(result, params)
call_args_list = self.get_filtered_task_call_args(
create_task_mock, urls.OUTBOUND_EMAIL_TASK + '.do')
self.assertEqual(2, len(call_args_list))
self.assertItemsEqual(
['user@example.com', 'mailing-list@example.com'],
result['notified'])
for (args, _kwargs) in call_args_list:
task = args[0]
body = json.loads(task['app_engine_http_request']['body'].decode())
# obfuscated email for non-members
if 'user' in body['to']:
self.assertIn(u'\u2026', body['from_addr'])
# Full email for members
if 'member' in body['to']:
self.assertNotIn(u'\u2026', body['from_addr'])
@mock.patch('framework.cloud_tasks_helpers.create_task')
def testNotifyBulkChangeTask_SubscriberGetsEmail(self, create_task_mock):
"""If a user subscription matches the issue, notify that user."""
task = notify.NotifyBulkChangeTask(services=self.services)
params = {
'send_email': 1,
'issue_ids': '%d' % (self.issue1.issue_id),
'seq': 0,
'old_owner_ids': '1', 'commenter_id': 1}
mr = testing_helpers.MakeMonorailRequest(
user_info={'user_id': 1},
params=params,
method='POST',
services=self.services)
self.services.user.TestAddUser('subscriber@example.com', 4)
sq = tracker_bizobj.MakeSavedQuery(
1, 'all open issues', 2, '', subscription_mode='immediate',
executes_in_project_ids=[self.issue1.project_id])
self.services.features.UpdateUserSavedQueries('cnxn', 4, [sq])
result = task.HandleRequest(mr)
self.VerifyParams(result, params)
call_args_list = self.get_filtered_task_call_args(
create_task_mock, urls.OUTBOUND_EMAIL_TASK + '.do')
self.assertEqual(2, len(call_args_list))
@mock.patch('framework.cloud_tasks_helpers.create_task')
def testNotifyBulkChangeTask_CCAndSubscriberListsIssueOnce(
self, create_task_mock):
"""If a user both CCs and subscribes, include issue only once."""
task = notify.NotifyBulkChangeTask(services=self.services)
params = {
'send_email': 1,
'issue_ids': '%d' % (self.issue1.issue_id),
'seq': 0,
'old_owner_ids': '1', 'commenter_id': 1}
mr = testing_helpers.MakeMonorailRequest(
user_info={'user_id': 1},
params=params,
method='POST',
services=self.services)
self.services.user.TestAddUser('subscriber@example.com', 4)
self.issue1.cc_ids = [4]
sq = tracker_bizobj.MakeSavedQuery(
1, 'all open issues', 2, '', subscription_mode='immediate',
executes_in_project_ids=[self.issue1.project_id])
self.services.features.UpdateUserSavedQueries('cnxn', 4, [sq])
result = task.HandleRequest(mr)
self.VerifyParams(result, params)
call_args_list = self.get_filtered_task_call_args(
create_task_mock, urls.OUTBOUND_EMAIL_TASK + '.do')
self.assertEqual(2, len(call_args_list))
found = False
for (args, _kwargs) in call_args_list:
task = args[0]
body = json.loads(task['app_engine_http_request']['body'].decode())
if body['to'] == 'subscriber@example.com':
found = True
task_body = body['body']
self.assertEqual(1, task_body.count('Issue %d' % self.issue1.local_id))
self.assertTrue(found)
@mock.patch('framework.cloud_tasks_helpers.create_task')
def testNotifyBulkChangeTask_Spam(self, _create_task_mock):
"""A spam issue is excluded from notification emails."""
issue2 = MakeTestIssue(
project_id=12345, local_id=2, owner_id=2, reporter_id=1,
is_spam=True)
self.services.issue.TestAddIssue(issue2)
task = notify.NotifyBulkChangeTask(services=self.services)
params = {
'send_email': 1,
'issue_ids': '%d,%d' % (self.issue1.issue_id, issue2.issue_id),
'seq': 0,
'old_owner_ids': '1,1', 'commenter_id': 1}
mr = testing_helpers.MakeMonorailRequest(
user_info={'user_id': 1},
params=params,
method='POST',
services=self.services)
result = task.HandleRequest(mr)
self.assertEqual(1, len(result['notified']))
def testFormatBulkIssues_Normal_Single(self):
"""A user may see full notification details for all changed issues."""
self.issue1.summary = 'one summary'
task = notify.NotifyBulkChangeTask(services=self.services)
users_by_id = {}
commenter_view = None
config = self.services.config.GetProjectConfig('cnxn', 12345)
addrperm = notify_reasons.AddrPerm(
False, 'nonmember@example.com', self.nonmember,
notify_reasons.REPLY_NOT_ALLOWED, None)
subject, body = task._FormatBulkIssues(
[self.issue1], users_by_id, commenter_view, 'localhost:8080',
'test comment', [], config, addrperm)
self.assertIn('one summary', subject)
self.assertIn('one summary', body)
self.assertIn('test comment', body)
def testFormatBulkIssues_Normal_Multiple(self):
"""A user may see full notification details for all changed issues."""
self.issue1.summary = 'one summary'
self.issue2.summary = 'two summary'
task = notify.NotifyBulkChangeTask(services=self.services)
users_by_id = {}
commenter_view = None
config = self.services.config.GetProjectConfig('cnxn', 12345)
addrperm = notify_reasons.AddrPerm(
False, 'nonmember@example.com', self.nonmember,
notify_reasons.REPLY_NOT_ALLOWED, None)
subject, body = task._FormatBulkIssues(
[self.issue1, self.issue2], users_by_id, commenter_view, 'localhost:8080',
'test comment', [], config, addrperm)
self.assertIn('2 issues changed', subject)
self.assertIn('one summary', body)
self.assertIn('two summary', body)
self.assertIn('test comment', body)
def testFormatBulkIssues_LinkOnly_Single(self):
"""A user may not see full notification details for some changed issue."""
self.issue1.summary = 'one summary'
self.issue1.labels = ['Restrict-View-Google']
task = notify.NotifyBulkChangeTask(services=self.services)
users_by_id = {}
commenter_view = None
config = self.services.config.GetProjectConfig('cnxn', 12345)
addrperm = notify_reasons.AddrPerm(
False, 'nonmember@example.com', self.nonmember,
notify_reasons.REPLY_NOT_ALLOWED, None)
subject, body = task._FormatBulkIssues(
[self.issue1], users_by_id, commenter_view, 'localhost:8080',
'test comment', [], config, addrperm)
self.assertIn('issue 1', subject)
self.assertNotIn('one summary', subject)
self.assertNotIn('one summary', body)
self.assertNotIn('test comment', body)
def testFormatBulkIssues_LinkOnly_Multiple(self):
"""A user may not see full notification details for some changed issue."""
self.issue1.summary = 'one summary'
self.issue1.labels = ['Restrict-View-Google']
self.issue2.summary = 'two summary'
task = notify.NotifyBulkChangeTask(services=self.services)
users_by_id = {}
commenter_view = None
config = self.services.config.GetProjectConfig('cnxn', 12345)
addrperm = notify_reasons.AddrPerm(
False, 'nonmember@example.com', self.nonmember,
notify_reasons.REPLY_NOT_ALLOWED, None)
subject, body = task._FormatBulkIssues(
[self.issue1, self.issue2], users_by_id, commenter_view, 'localhost:8080',
'test comment', [], config, addrperm)
self.assertIn('2 issues', subject)
self.assertNotIn('summary', subject)
self.assertNotIn('one summary', body)
self.assertIn('two summary', body)
self.assertNotIn('test comment', body)
@mock.patch('framework.cloud_tasks_helpers.create_task')
def testNotifyApprovalChangeTask_Normal(self, _create_task_mock):
config = self.services.config.GetProjectConfig('cnxn', 12345)
config.field_defs = [
# issue's User field with any_comment is notified.
tracker_bizobj.MakeFieldDef(
121, 12345, 'TL', tracker_pb2.FieldTypes.USER_TYPE,
'', '', False, False, False, None, None, None, False, '',
None, tracker_pb2.NotifyTriggers.ANY_COMMENT, 'no_action',
'TL, notified on everything', False),
# issue's User field with never is not notified.
tracker_bizobj.MakeFieldDef(
122, 12345, 'silentTL', tracker_pb2.FieldTypes.USER_TYPE,
'', '', False, False, False, None, None, None, False, '',
None, tracker_pb2.NotifyTriggers.NEVER, 'no_action',
'TL, notified on nothing', False),
# approval's User field with any_comment is notified.
tracker_bizobj.MakeFieldDef(
123, 12345, 'otherapprovalTL', tracker_pb2.FieldTypes.USER_TYPE,
'', '', False, False, False, None, None, None, False, '',
None, tracker_pb2.NotifyTriggers.ANY_COMMENT, 'no_action',
'TL on the approvers team', False, approval_id=3),
# another approval's User field with any_comment is not notified.
tracker_bizobj.MakeFieldDef(
124, 12345, 'otherapprovalTL', tracker_pb2.FieldTypes.USER_TYPE,
'', '', False, False, False, None, None, None, False, '',
None, tracker_pb2.NotifyTriggers.ANY_COMMENT, 'no_action',
'TL on another approvers team', False, approval_id=4),
tracker_bizobj.MakeFieldDef(
3, 12345, 'Goat-Approval', tracker_pb2.FieldTypes.APPROVAL_TYPE,
'', '', False, False, False, None, None, None, False, '',
None, tracker_pb2.NotifyTriggers.NEVER, 'no_action',
'Get Approval from Goats', False)
]
self.services.config.StoreConfig('cnxn', config)
# Custom user_type field TLs
self.services.user.TestAddUser('TL@example.com', 111)
self.services.user.TestAddUser('silentTL@example.com', 222)
self.services.user.TestAddUser('approvalTL@example.com', 333)
self.services.user.TestAddUser('otherapprovalTL@example.com', 444)
# Approvers
self.services.user.TestAddUser('approver_old@example.com', 777)
self.services.user.TestAddUser('approver_new@example.com', 888)
self.services.user.TestAddUser('approver_still@example.com', 999)
self.services.user.TestAddUser('approver_group@example.com', 666)
self.services.user.TestAddUser('group_mem1@example.com', 661)
self.services.user.TestAddUser('group_mem2@example.com', 662)
self.services.user.TestAddUser('group_mem3@example.com', 663)
self.services.usergroup.TestAddGroupSettings(
666, 'approver_group@example.com')
self.services.usergroup.TestAddMembers(666, [661, 662, 663])
canary_phase = tracker_pb2.Phase(
name='Canary', phase_id=1, rank=1)
approval_values = [
tracker_pb2.ApprovalValue(approval_id=3,
approver_ids=[888, 999, 666, 661])]
approval_issue = MakeTestIssue(
project_id=12345, local_id=2, owner_id=2, reporter_id=1,
is_spam=True)
approval_issue.phases = [canary_phase]
approval_issue.approval_values = approval_values
approval_issue.field_values = [
tracker_bizobj.MakeFieldValue(121, None, None, 111, None, None, False),
tracker_bizobj.MakeFieldValue(122, None, None, 222, None, None, False),
tracker_bizobj.MakeFieldValue(123, None, None, 333, None, None, False),
tracker_bizobj.MakeFieldValue(124, None, None, 444, None, None, False),
]
self.services.issue.TestAddIssue(approval_issue)
amend = tracker_bizobj.MakeApprovalApproversAmendment([888], [777])
comment = tracker_pb2.IssueComment(
project_id=12345, user_id=999, issue_id=approval_issue.issue_id,
amendments=[amend], timestamp=1234567890, content='just a comment.')
attach = tracker_pb2.Attachment(
attachment_id=4567, filename='sploot.jpg', mimetype='image/png',
gcs_object_id='/pid/attachments/abcd', filesize=(1024 * 1023))
comment.attachments.append(attach)
self.services.issue.TestAddComment(comment, approval_issue.local_id)
self.services.issue.TestAddAttachment(
attach, comment.id, approval_issue.issue_id)
task = notify.NotifyApprovalChangeTask(services=self.services)
params = {
'send_email': 1,
'issue_id': approval_issue.issue_id,
'approval_id': 3,
'comment_id': comment.id,
}
mr = testing_helpers.MakeMonorailRequest(
user_info={'user_id': 1},
params=params,
method='POST',
services=self.services)
result = task.HandleRequest(mr)
self.assertTrue('just a comment' in result['tasks'][0]['body'])
self.assertTrue('Approvers: -appro...' in result['tasks'][0]['body'])
self.assertTrue('sploot.jpg' in result['tasks'][0]['body'])
self.assertTrue(
'/issues/attachment?aid=4567' in result['tasks'][0]['body'])
self.assertItemsEqual(
['user@example.com', 'approver_old@example.com',
'approver_new@example.com', 'TL@example.com',
'approvalTL@example.com', 'group_mem1@example.com',
'group_mem2@example.com', 'group_mem3@example.com'],
result['notified'])
# Test no approvers/groups notified
# Status change to NEED_INFO does not email approvers.
amend2 = tracker_bizobj.MakeApprovalStatusAmendment(
tracker_pb2.ApprovalStatus.NEED_INFO)
comment2 = tracker_pb2.IssueComment(
project_id=12345, user_id=999, issue_id=approval_issue.issue_id,
amendments=[amend2], timestamp=1234567891, content='')
self.services.issue.TestAddComment(comment2, approval_issue.local_id)
task = notify.NotifyApprovalChangeTask(services=self.services)
params = {
'send_email': 1,
'issue_id': approval_issue.issue_id,
'approval_id': 3,
'comment_id': comment2.id,
}
mr = testing_helpers.MakeMonorailRequest(
user_info={'user_id': 1},
params=params,
method='POST',
services=self.services)
result = task.HandleRequest(mr)
self.assertIsNotNone(result['tasks'][0].get('references'))
self.assertEqual(result['tasks'][0]['reply_to'], emailfmt.NoReplyAddress())
self.assertTrue('Status: need_info' in result['tasks'][0]['body'])
self.assertItemsEqual(
['user@example.com', 'TL@example.com', 'approvalTL@example.com'],
result['notified'])
def testNotifyApprovalChangeTask_GetApprovalEmailRecipients(self):
task = notify.NotifyApprovalChangeTask(services=self.services)
issue = fake.MakeTestIssue(789, 1, 'summary', 'New', 111)
approval_value = tracker_pb2.ApprovalValue(
approver_ids=[222, 333],
status=tracker_pb2.ApprovalStatus.APPROVED)
comment = tracker_pb2.IssueComment(
project_id=789, user_id=1, issue_id=78901)
# Comment with not amendments notifies everyone.
rids = task._GetApprovalEmailRecipients(
approval_value, comment, issue, [777, 888])
self.assertItemsEqual(rids, [111, 222, 333, 777, 888])
# New APPROVED status notifies owners and any_comment users.
amendment = tracker_bizobj.MakeApprovalStatusAmendment(
tracker_pb2.ApprovalStatus.APPROVED)
comment.amendments = [amendment]
rids = task._GetApprovalEmailRecipients(
approval_value, comment, issue, [777, 888])
self.assertItemsEqual(rids, [111, 777, 888])
# New REVIEW_REQUESTED status notifies approvers.
approval_value.status = tracker_pb2.ApprovalStatus.REVIEW_REQUESTED
amendment = tracker_bizobj.MakeApprovalStatusAmendment(
tracker_pb2.ApprovalStatus.REVIEW_REQUESTED)
comment.amendments = [amendment]
rids = task._GetApprovalEmailRecipients(
approval_value, comment, issue, [777, 888])
self.assertItemsEqual(rids, [222, 333])
# Approvers change notifies everyone.
amendment = tracker_bizobj.MakeApprovalApproversAmendment(
[222], [555])
comment.amendments = [amendment]
approval_value.approver_ids = [222]
rids = task._GetApprovalEmailRecipients(
approval_value, comment, issue, [777], omit_ids=[444, 333])
self.assertItemsEqual(rids, [111, 222, 555, 777])
@mock.patch('framework.cloud_tasks_helpers.create_task')
def testNotifyRulesDeletedTask(self, _create_task_mock):
self.services.project.TestAddProject(
'proj', owner_ids=[777, 888], project_id=789)
self.services.user.TestAddUser('owner1@test.com', 777)
self.services.user.TestAddUser('cow@test.com', 888)
task = notify.NotifyRulesDeletedTask(services=self.services)
params = {'project_id': 789,
'filter_rules': 'if green make yellow,if orange make blue'}
mr = testing_helpers.MakeMonorailRequest(
params=params,
method='POST',
services=self.services)
result = task.HandleRequest(mr)
self.assertEqual(len(result['tasks']), 2)
body = result['tasks'][0]['body']
self.assertTrue('if green make yellow' in body)
self.assertTrue('if green make yellow' in body)
self.assertTrue('/p/proj/adminRules' in body)
self.assertItemsEqual(
['cow@test.com', 'owner1@test.com'], result['notified'])
def testOutboundEmailTask_Normal(self):
"""We can send an email."""
params = {
'from_addr': 'requester@example.com',
'reply_to': 'user@example.com',
'to': 'user@example.com',
'subject': 'Test subject'}
data = json.dumps(params)
res = self.app.test_client().post('/_task/outboundEmail.do', data=data)
res_string = res.get_data()[5:]
res_json = json.loads(res_string)
self.assertEqual(params['from_addr'], res_json['sender'])
self.assertEqual(params['subject'], res_json['subject'])
def testOutboundEmailTask_MissingTo(self):
"""We skip emails that don't specify the To-line."""
params = {
'from_addr': 'requester@example.com',
'reply_to': 'user@example.com',
'subject': 'Test subject'}
data = json.dumps(params)
res = self.app.test_client().post('/_task/outboundEmail.do', data=data)
res_string = res.get_data()[5:]
res_json = json.loads(res_string)
self.assertEqual(
'Skipping because no "to" address found.', res_json['note'])
self.assertNotIn('from_addr', res_string)
def testOutboundEmailTask_BannedUser(self):
"""We don't send emails to banned users.."""
self.servlet.services.user.TestAddUser(
'banned@example.com', 404, banned=True)
params = {
'from_addr': 'requester@example.com',
'reply_to': 'user@example.com',
'to': 'banned@example.com',
'subject': 'Test subject'}
data = json.dumps(params)
res = self.app.test_client().post('/_task/outboundEmail.do', data=data)
res_string = res.get_data()[5:]
res_json = json.loads(res_string)
self.assertEqual('Skipping because user is banned.', res_json['note'])
self.assertNotIn('from_addr', res_string)