Open Source Monorail

appengine/monorail/features/test/notify_test.py

+# 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
+
+"""Tests for notify.py."""
+
+import os
+import unittest
+import urllib
+import webapp2
+import webtest
+
+from google.appengine.api import taskqueue
+from google.appengine.ext import testbed
+
+from features import notify
+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 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.owner_id = owner_id
+  issue.reporter_id = reporter_id
+  issue.is_spam = is_spam
+  return issue
+
+
+class SendNotificationTest(unittest.TestCase):
+
+  def setUp(self):
+    self.testbed = testbed.Testbed()
+    self.testbed.activate()
+    self.testbed.init_taskqueue_stub()
+    self.taskqueue_stub = self.testbed.get_stub(testbed.TASKQUEUE_SERVICE_NAME)
+    self.taskqueue_stub._root_path = os.path.dirname(
+        os.path.dirname(os.path.dirname( __file__ )))
+
+  def tearDown(self):
+    self.testbed.deactivate()
+
+  def testPrepareAndSendIssueChangeNotification(self):
+    notify.PrepareAndSendIssueChangeNotification(
+        project_id=789,
+        local_id=1,
+        hostport='testbed-test.appspotmail.com',
+        commenter_id=1,
+        seq_num=0,
+        old_owner_id=2,
+        send_email=True)
+
+    tasks = self.taskqueue_stub.get_filtered_tasks(
+        url=urls.NOTIFY_ISSUE_CHANGE_TASK + '.do')
+    self.assertEqual(1, len(tasks))
+
+  def testPrepareAndSendIssueBlockingNotification(self):
+    notify.PrepareAndSendIssueBlockingNotification(
+        project_id=789,
+        hostport='testbed-test.appspotmail.com',
+        local_id=1,
+        delta_blocker_iids=[],
+        commenter_id=1,
+        send_email=True)
+
+    tasks = self.taskqueue_stub.get_filtered_tasks(
+        url=urls.NOTIFY_BLOCKING_CHANGE_TASK + '.do')
+    self.assertEqual(0, len(tasks))
+
+    notify.PrepareAndSendIssueBlockingNotification(
+        project_id=789,
+        hostport='testbed-test.appspotmail.com',
+        local_id=1,
+        delta_blocker_iids=[2],
+        commenter_id=1,
+        send_email=True)
+
+    tasks = self.taskqueue_stub.get_filtered_tasks(
+        url=urls.NOTIFY_BLOCKING_CHANGE_TASK + '.do')
+    self.assertEqual(1, len(tasks))
+
+  def testSendIssueBulkChangeNotification_CommentOnly(self):
+    notify.SendIssueBulkChangeNotification(
+        hostport='testbed-test.appspotmail.com',
+        project_id=789,
+        local_ids=[1],
+        old_owner_ids=[2],
+        comment_text='comment',
+        commenter_id=1,
+        amendments=[],
+        send_email=True,
+        users_by_id=2)
+
+    tasks = self.taskqueue_stub.get_filtered_tasks(
+        url=urls.NOTIFY_BULK_CHANGE_TASK + '.do')
+    self.assertEqual(1, len(tasks))
+    params = dict(urllib.unquote_plus(item).split('=')
+                  for item in tasks[0].payload.split('&'))
+    self.assertEqual('comment', params['comment_text'])
+    self.assertEqual('', params['amendments'])
+
+  def testSendIssueBulkChangeNotification_Normal(self):
+    notify.SendIssueBulkChangeNotification(
+        hostport='testbed-test.appspotmail.com',
+        project_id=789,
+        local_ids=[1],
+        old_owner_ids=[2],
+        comment_text='comment',
+        commenter_id=1,
+        amendments=[
+            tracker_bizobj.MakeStatusAmendment('New', 'Old'),
+            tracker_bizobj.MakeLabelsAmendment(['Added'], ['Removed']),
+            tracker_bizobj.MakeStatusAmendment('New', 'Old'),
+            ],
+        send_email=True,
+        users_by_id=2)
+
+    tasks = self.taskqueue_stub.get_filtered_tasks(
+        url=urls.NOTIFY_BULK_CHANGE_TASK + '.do')
+    self.assertEqual(1, len(tasks))
+    params = dict(urllib.unquote_plus(item).split('=')
+                  for item in tasks[0].payload.split('&'))
+    self.assertEqual('comment', params['comment_text'])
+    self.assertEqual(
+        ['    Status: New',
+         '    Labels: -Removed Added'],
+        params['amendments'].split('\n'))
+
+  def testAddAllEmailTasks(self):
+    notify.AddAllEmailTasks(
+      tasks=[{'to': 'user'}, {'to': 'user2'}])
+
+    tasks = self.taskqueue_stub.get_filtered_tasks(
+        url=urls.OUTBOUND_EMAIL_TASK + '.do')
+    self.assertEqual(2, len(tasks))
+
+
+class NotifyTaskHandleRequestTest(unittest.TestCase):
+
+  def setUp(self):
+    self.testbed = testbed.Testbed()
+    self.testbed.activate()
+    self.testbed.init_taskqueue_stub()
+    self.taskqueue_stub = self.testbed.get_stub(testbed.TASKQUEUE_SERVICE_NAME)
+    self.taskqueue_stub._root_path = os.path.dirname(
+        os.path.dirname(os.path.dirname( __file__ )))
+    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.services.user.TestAddUser('requester@example.com', 1)
+    self.services.user.TestAddUser('user@example.com', 2)
+    self.services.project.TestAddProject(
+        'test-project', owner_ids=[1],
+        project_id=12345)
+    issue1 = MakeTestIssue(
+        project_id=12345, local_id=1, owner_id=2, reporter_id=1)
+    self.services.issue.TestAddIssue(issue1)
+
+  def VerifyParams(self, result, params):
+    self.assertEqual(
+        bool(params['send_email']), result['params']['send_email'])
+    if 'id' in params:
+      self.assertEqual(params['id'], result['params']['local_id'])
+    if 'ids' in params:
+      self.assertEqual([int(p) for p in params['ids'].split(',')],
+                       result['params']['local_ids'])
+    self.assertEqual(params['project_id'], result['params']['project_id'])
+
+  def testNotifyIssueChangeTask(self):
+    task = notify.NotifyIssueChangeTask(
+        request=None, response=None, services=self.services)
+    params = {'send_email': 1, 'project_id': 12345, 'id': 1, '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)
+
+  def testNotifyIssueChangeTask_spam(self):
+    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(
+        request=None, response=None, services=self.services)
+    params = {'send_email': 0, 'project_id': 12345, 'id': 1, '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.assertEquals(0, len(result['notified']))
+
+  def testNotifyBlockingChangeTask(self):
+    issue2 = MakeTestIssue(
+        project_id=12345, local_id=2, owner_id=2, reporter_id=1)
+    self.services.issue.TestAddIssue(issue2)
+    task = notify.NotifyBlockingChangeTask(
+        request=None, response=None, services=self.services)
+    params = {'send_email': 1, 'project_id': 12345, 'id': 1, 'seq': 0,
+              'delta_blocker_iids': 2, '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)
+
+  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(
+        request=None, response=None, services=self.services)
+    params = {'send_email': 1, 'project_id': 12345, 'id': 1, 'seq': 0,
+              'delta_blocker_iids': 2, 'commenter_id': 1}
+    mr = testing_helpers.MakeMonorailRequest(
+        user_info={'user_id': 1},
+        params=params,
+        method='POST',
+        services=self.services)
+    result = task.HandleRequest(mr)
+    self.assertEquals(0, len(result['notified']))
+
+  def testNotifyBulkChangeTask(self):
+    issue2 = MakeTestIssue(
+        project_id=12345, local_id=2, owner_id=2, reporter_id=1)
+    self.services.issue.TestAddIssue(issue2)
+    task = notify.NotifyBulkChangeTask(
+        request=None, response=None, services=self.services)
+    params = {'send_email': 1, 'project_id': 12345, 'ids': '1,2', '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.VerifyParams(result, params)
+
+  def testNotifyBulkChangeTask_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.NotifyBulkChangeTask(
+        request=None, response=None, services=self.services)
+    params = {'send_email': 1, 'project_id': 12345, 'ids': '1,2', '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.assertEquals(1, len(result['notified']))
+
+  def testOutboundEmailTask(self):
+    task = notify.OutboundEmailTask(
+        request=None, response=None, services=self.services)
+    params = {
+        'from_addr': 'requester@example.com',
+        'reply_to': 'user@example.com',
+        'to': 'user@example.com',
+        'subject': 'Test subject'}
+    mr = testing_helpers.MakeMonorailRequest(
+        user_info={'user_id': 1},
+        params=params,
+        method='POST',
+        services=self.services)
+    result = task.HandleRequest(mr)
+    self.assertEqual(params['from_addr'], result['sender'])
+    self.assertEqual(params['subject'], result['subject'])