features/test/send_notifications_test.py

# 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.

"""Tests for prepareandsend.py"""
from __future__ import print_function
from __future__ import division
from __future__ import absolute_import

import mock
import unittest
from six.moves.urllib.parse import parse_qs

from features import send_notifications
from framework import urls
from tracker import tracker_bizobj


class SendNotificationTest(unittest.TestCase):

  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'].startswith(
            relative_uri)
    ]

  def _get_create_task_path_and_params(self, call):
    (args, _kwargs) = call
    path = args[0]['app_engine_http_request']['relative_uri']
    encoded_params = args[0]['app_engine_http_request']['body']
    params = {k: v[0] for k, v in parse_qs(encoded_params, True).items()}
    return path, params

  @mock.patch('framework.cloud_tasks_helpers.create_task')
  def testPrepareAndSendIssueChangeNotification(self, create_task_mock):
    send_notifications.PrepareAndSendIssueChangeNotification(
        issue_id=78901,
        hostport='testbed-test.appspotmail.com',
        commenter_id=1,
        old_owner_id=2,
        send_email=True)

    call_args_list = self._get_filtered_task_call_args(
        create_task_mock, urls.NOTIFY_ISSUE_CHANGE_TASK + '.do')
    self.assertEqual(1, len(call_args_list))

  @mock.patch('framework.cloud_tasks_helpers.create_task')
  def testPrepareAndSendIssueBlockingNotification(self, create_task_mock):
    send_notifications.PrepareAndSendIssueBlockingNotification(
        issue_id=78901,
        hostport='testbed-test.appspotmail.com',
        delta_blocker_iids=[],
        commenter_id=1,
        send_email=True)

    call_args_list = self._get_filtered_task_call_args(
        create_task_mock, urls.NOTIFY_BLOCKING_CHANGE_TASK + '.do')
    self.assertEqual(0, len(call_args_list))

    send_notifications.PrepareAndSendIssueBlockingNotification(
        issue_id=78901,
        hostport='testbed-test.appspotmail.com',
        delta_blocker_iids=[2],
        commenter_id=1,
        send_email=True)

    call_args_list = self._get_filtered_task_call_args(
        create_task_mock, urls.NOTIFY_BLOCKING_CHANGE_TASK + '.do')
    self.assertEqual(1, len(call_args_list))

  @mock.patch('framework.cloud_tasks_helpers.create_task')
  def testPrepareAndSendApprovalChangeNotification(self, create_task_mock):
    send_notifications.PrepareAndSendApprovalChangeNotification(
        78901, 3, 'testbed-test.appspotmail.com', 55)

    call_args_list = self._get_filtered_task_call_args(
        create_task_mock, urls.NOTIFY_APPROVAL_CHANGE_TASK + '.do')
    self.assertEqual(1, len(call_args_list))

  @mock.patch('framework.cloud_tasks_helpers.create_task')
  def testSendIssueBulkChangeNotification_CommentOnly(self, create_task_mock):
    send_notifications.SendIssueBulkChangeNotification(
        issue_ids=[78901],
        hostport='testbed-test.appspotmail.com',
        old_owner_ids=[2],
        comment_text='comment',
        commenter_id=1,
        amendments=[],
        send_email=True,
        users_by_id=2)

    call_args_list = self._get_filtered_task_call_args(
        create_task_mock, urls.NOTIFY_BULK_CHANGE_TASK + '.do')
    self.assertEqual(1, len(call_args_list))
    _path, params = self._get_create_task_path_and_params(call_args_list[0])
    self.assertEqual(params[b'comment_text'], b'comment')
    self.assertEqual(params[b'amendments'], b'')

  @mock.patch('framework.cloud_tasks_helpers.create_task')
  def testSendIssueBulkChangeNotification_Normal(self, create_task_mock):
    send_notifications.SendIssueBulkChangeNotification(
        issue_ids=[78901],
        hostport='testbed-test.appspotmail.com',
        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)

    call_args_list = self._get_filtered_task_call_args(
        create_task_mock, urls.NOTIFY_BULK_CHANGE_TASK + '.do')
    self.assertEqual(1, len(call_args_list))
    _path, params = self._get_create_task_path_and_params(call_args_list[0])
    self.assertEqual(params[b'comment_text'], b'comment')
    self.assertEqual(
        params[b'amendments'].split(b'\n'),
        [b'    Status: New', b'    Labels: -Removed Added'])

  @mock.patch('framework.cloud_tasks_helpers.create_task')
  def testPrepareAndSendDeletedFilterRulesNotifications(self, create_task_mock):
    filter_rule_strs = ['if yellow make orange', 'if orange make blue']
    send_notifications.PrepareAndSendDeletedFilterRulesNotification(
        789, 'testbed-test.appspotmail.com', filter_rule_strs)

    call_args_list = self._get_filtered_task_call_args(
        create_task_mock, urls.NOTIFY_RULES_DELETED_TASK + '.do')
    self.assertEqual(1, len(call_args_list))
    _path, params = self._get_create_task_path_and_params(call_args_list[0])
    self.assertEqual(params[b'project_id'], b'789')
    self.assertEqual(
        params[b'filter_rules'], b'if yellow make orange,if orange make blue')