api/v3/test/issues_servicer_test.py

# Copyright 2020 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 the issues servicer."""
from __future__ import print_function
from __future__ import division
from __future__ import absolute_import

import copy
import unittest
import mock

from api.v3 import converters
from api.v3 import issues_servicer
from api.v3.api_proto import issues_pb2
from api.v3.api_proto import issue_objects_pb2
from framework import exceptions
from framework import framework_helpers
from framework import monorailcontext
from framework import permissions
from proto import tracker_pb2
from testing import fake
from services import service_manager

from google.appengine.ext import testbed
from google.protobuf import timestamp_pb2
from google.protobuf import field_mask_pb2


def _Issue(project_id, local_id):
  issue = tracker_pb2.Issue(owner_id=0)
  issue.project_name = 'proj-%d' % project_id
  issue.project_id = project_id
  issue.local_id = local_id
  issue.issue_id = project_id * 100 + local_id
  return issue


CURRENT_TIME = 12346.78


class IssuesServicerTest(unittest.TestCase):

  def setUp(self):
    # memcache and datastore needed for generating page tokens.
    self.testbed = testbed.Testbed()
    self.testbed.activate()
    self.testbed.init_memcache_stub()
    self.testbed.init_datastore_v3_stub()

    self.cnxn = fake.MonorailConnection()
    self.services = service_manager.Services(
        config=fake.ConfigService(),
        issue=fake.IssueService(),
        issue_star=fake.IssueStarService(),
        project=fake.ProjectService(),
        features=fake.FeaturesService(),
        spam=fake.SpamService(),
        user=fake.UserService(),
        usergroup=fake.UserGroupService())
    self.issues_svcr = issues_servicer.IssuesServicer(
        self.services, make_rate_limiter=False)
    self.PAST_TIME = int(CURRENT_TIME - 1)

    self.owner = self.services.user.TestAddUser('owner@example.com', 111)
    self.user_2 = self.services.user.TestAddUser('user_2@example.com', 222)

    self.project_1 = self.services.project.TestAddProject(
        'chicken', project_id=789)
    self.issue_1_resource_name = 'projects/chicken/issues/1234'
    self.issue_1 = fake.MakeTestIssue(
        self.project_1.project_id,
        1234,
        'sum',
        'New',
        self.owner.user_id,
        labels=['find-me', 'pri-3'],
        project_name=self.project_1.project_name)
    self.services.issue.TestAddIssue(self.issue_1)

    self.project_2 = self.services.project.TestAddProject('cow', project_id=788)
    self.issue_2_resource_name = 'projects/cow/issues/1235'
    self.issue_2 = fake.MakeTestIssue(
        self.project_2.project_id,
        1235,
        'sum',
        'New',
        self.user_2.user_id,
        project_name=self.project_2.project_name)
    self.services.issue.TestAddIssue(self.issue_2)
    self.issue_3 = fake.MakeTestIssue(
        self.project_2.project_id,
        1236,
        'sum',
        'New',
        self.user_2.user_id,
        labels=['find-me', 'pri-1'],
        project_name=self.project_2.project_name)
    self.services.issue.TestAddIssue(self.issue_3)

  def CallWrapped(self, wrapped_handler, mc, *args, **kwargs):
    self.issues_svcr.converter = converters.Converter(mc, self.services)
    return wrapped_handler.wrapped(self.issues_svcr, mc, *args, **kwargs)

  def testGetIssue(self):
    """We can get an issue."""
    request = issues_pb2.GetIssueRequest(name=self.issue_1_resource_name)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    actual_response = self.CallWrapped(self.issues_svcr.GetIssue, mc, request)
    self.assertEqual(
        actual_response, self.issues_svcr.converter.ConvertIssue(self.issue_1))

  def testBatchGetIssues(self):
    """We can batch get issues."""
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    request = issues_pb2.BatchGetIssuesRequest(
        names=['projects/cow/issues/1235', 'projects/cow/issues/1236'])
    actual_response = self.CallWrapped(
        self.issues_svcr.BatchGetIssues, mc, request)
    self.assertEqual(
        [issue.name for issue in actual_response.issues],
        ['projects/cow/issues/1235', 'projects/cow/issues/1236'])

  def testBatchGetIssues_Empty(self):
    """We can return a response if the request has no names."""
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    request = issues_pb2.BatchGetIssuesRequest(names=[])
    actual_response = self.CallWrapped(
        self.issues_svcr.BatchGetIssues, mc, request)
    self.assertEqual(
        actual_response, issues_pb2.BatchGetIssuesResponse(issues=[]))

  def testBatchGetIssues_WithParent(self):
    """We can batch get issues with a given parent."""
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    request = issues_pb2.BatchGetIssuesRequest(
        parent='projects/cow',
        names=['projects/cow/issues/1235', 'projects/cow/issues/1236'])
    actual_response = self.CallWrapped(
        self.issues_svcr.BatchGetIssues, mc, request)
    self.assertEqual(
        [issue.name for issue in actual_response.issues],
        ['projects/cow/issues/1235', 'projects/cow/issues/1236'])

  def testBatchGetIssues_FromMultipleProjects(self):
    """We can batch get issues from multiple projects."""
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    request = issues_pb2.BatchGetIssuesRequest(
        names=[
            'projects/chicken/issues/1234', 'projects/cow/issues/1235',
            'projects/cow/issues/1236'
        ])
    actual_response = self.CallWrapped(
        self.issues_svcr.BatchGetIssues, mc, request)
    self.assertEqual(
        [issue.name for issue in actual_response.issues], [
            'projects/chicken/issues/1234', 'projects/cow/issues/1235',
            'projects/cow/issues/1236'
        ])

  def testBatchGetIssues_WithBadInput(self):
    """We raise an exception with bad input to batch get issues."""
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    request = issues_pb2.BatchGetIssuesRequest(
        parent='projects/cow',
        names=['projects/cow/issues/1235', 'projects/chicken/issues/1234'])
    with self.assertRaisesRegexp(
        exceptions.InputException,
        'projects/chicken/issues/1234 is not a child issue of projects/cow.'):
      self.CallWrapped(self.issues_svcr.BatchGetIssues, mc, request)

    request = issues_pb2.BatchGetIssuesRequest(
        parent='projects/sheep',
        names=['projects/cow/issues/1235', 'projects/chicken/issues/1234'])
    with self.assertRaisesRegexp(
        exceptions.InputException,
        'projects/cow/issues/1235 is not a child issue of projects/sheep.\n' +
        'projects/chicken/issues/1234 is not a child issue of projects/sheep.'):
      self.CallWrapped(self.issues_svcr.BatchGetIssues, mc, request)

    request = issues_pb2.BatchGetIssuesRequest(
        parent='projects/cow',
        names=['projects/cow/badformat/1235', 'projects/chicken/issues/1234'])
    with self.assertRaisesRegexp(
        exceptions.InputException,
        'Invalid resource name: projects/cow/badformat/1235.'):
      self.CallWrapped(self.issues_svcr.BatchGetIssues, mc, request)

  def testBatchGetIssues_NonExistentIssues(self):
    """We raise an exception with bad input to batch get issues."""
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    request = issues_pb2.BatchGetIssuesRequest(
        parent='projects/chicken',
        names=['projects/chicken/issues/1', 'projects/chicken/issues/2'])
    with self.assertRaisesRegexp(
        exceptions.NoSuchIssueException,
        "\['projects/chicken/issues/1', 'projects/chicken/issues/2'\] not found"
    ):
      self.CallWrapped(self.issues_svcr.BatchGetIssues, mc, request)

  @mock.patch('api.v3.api_constants.MAX_BATCH_ISSUES', 2)
  def testBatchGetIssues(self):
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    request = issues_pb2.BatchGetIssuesRequest(
        parent='projects/cow',
        names=[
            'projects/cow/issues/1235', 'projects/chicken/issues/1234',
            'projects/cow/issues/1233'
        ])
    with self.assertRaises(exceptions.InputException):
      self.CallWrapped(self.issues_svcr.BatchGetIssues, mc, request)

  @mock.patch('search.frontendsearchpipeline.FrontendSearchPipeline')
  @mock.patch('api.v3.api_constants.MAX_ISSUES_PER_PAGE', 2)
  def testSearchIssues(self, mock_pipeline):
    """We can search for issues in some projects."""
    request = issues_pb2.SearchIssuesRequest(
        projects=['projects/chicken', 'projects/cow'],
        query='label:find-me',
        order_by='-pri',
        page_size=3)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.user_2.email)

    instance = mock.Mock(
        spec=True, total_count=3, visible_results=[self.issue_1, self.issue_3])
    mock_pipeline.return_value = instance
    instance.SearchForIIDs = mock.Mock()
    instance.MergeAndSortIssues = mock.Mock()
    instance.Paginate = mock.Mock()

    actual_response = self.CallWrapped(
        self.issues_svcr.SearchIssues, mc, request)
    # start index is 0.
    # number of items is coerced from 3 -> 2
    mock_pipeline.assert_called_once_with(
        self.cnxn,
        self.services,
        mc.auth, [222],
        'label:find-me', ['chicken', 'cow'],
        2,
        0,
        1,
        '',
        '-pri',
        mc.warnings,
        mc.errors,
        True,
        mc.profiler,
        project=None)
    self.assertEqual(
        [issue.name for issue in actual_response.issues],
        ['projects/chicken/issues/1234', 'projects/cow/issues/1236'])

    # Check the `next_page_token` can be used to get the next page of results.
    request.page_token = actual_response.next_page_token
    self.CallWrapped(self.issues_svcr.SearchIssues, mc, request)
    # start index is now 2.
    mock_pipeline.assert_called_with(
        self.cnxn,
        self.services,
        mc.auth, [222],
        'label:find-me', ['chicken', 'cow'],
        2,
        2,
        1,
        '',
        '-pri',
        mc.warnings,
        mc.errors,
        True,
        mc.profiler,
        project=None)

  @mock.patch('search.frontendsearchpipeline.FrontendSearchPipeline')
  @mock.patch('api.v3.api_constants.MAX_ISSUES_PER_PAGE', 2)
  def testSearchIssues_PaginationErrorOrderByChanged(self, mock_pipeline):
    """Error when changing the order_by and using the same page_otoken."""
    request = issues_pb2.SearchIssuesRequest(
        projects=['projects/chicken', 'projects/cow'],
        query='label:find-me',
        order_by='-pri',
        page_size=3)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.user_2.email)

    instance = mock.Mock(
        spec=True, total_count=3, visible_results=[self.issue_1, self.issue_3])
    mock_pipeline.return_value = instance
    instance.SearchForIIDs = mock.Mock()
    instance.MergeAndSortIssues = mock.Mock()
    instance.Paginate = mock.Mock()

    actual_response = self.CallWrapped(
        self.issues_svcr.SearchIssues, mc, request)

    # The request should fail if we use `next_page_token` and change parameters.
    request.page_token = actual_response.next_page_token
    request.order_by = 'owner'
    with self.assertRaises(exceptions.PageTokenException):
      self.CallWrapped(self.issues_svcr.SearchIssues, mc, request)

  # Note the 'empty' case doesn't make sense for ListComments, as one is created
  # for every issue.
  def testListComments(self):
    comment_2 = tracker_pb2.IssueComment(
        id=123,
        issue_id=self.issue_1.issue_id,
        project_id=self.issue_1.project_id,
        user_id=self.owner.user_id,
        content='comment 2')
    self.services.issue.TestAddComment(comment_2, self.issue_1.local_id)
    request = issues_pb2.ListCommentsRequest(
        parent=self.issue_1_resource_name, page_size=1)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    actual_response = self.CallWrapped(
        self.issues_svcr.ListComments, mc, request)
    self.assertEqual(len(actual_response.comments), 1)

    # Check the `next_page_token` can be used to get the next page of results
    request.page_token = actual_response.next_page_token
    next_actual_response = self.CallWrapped(
        self.issues_svcr.ListComments, mc, request)
    self.assertEqual(len(next_actual_response.comments), 1)
    self.assertEqual(next_actual_response.comments[0].content, 'comment 2')

  def testListComments_UnsupportedFilter(self):
    """If anything other than approval is provided, it's an error."""
    filter_str = 'content = "x"'
    request = issues_pb2.ListCommentsRequest(
        parent=self.issue_1_resource_name, page_size=1, filter=filter_str)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    with self.assertRaises(exceptions.InputException):
      self.CallWrapped(self.issues_svcr.ListComments, mc, request)

  def testListComments_TwoApprovalsErrors(self):
    """If anything other than a single approval is provided, it's an error."""
    filter_str = (
        'approval = "projects/chicken/approvalDefs/404" OR '
        'approval = "projects/chicken/approvalDefs/405')
    request = issues_pb2.ListCommentsRequest(
        parent=self.issue_1_resource_name, page_size=1, filter=filter_str)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    with self.assertRaises(exceptions.InputException):
      self.CallWrapped(self.issues_svcr.ListComments, mc, request)

  def testListComments_FilterTypoError(self):
    """Even an extra space is an error."""
    filter_str = 'approval = "projects/chicken/approvalDefs/404" '
    request = issues_pb2.ListCommentsRequest(
        parent=self.issue_1_resource_name, page_size=1, filter=filter_str)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    with self.assertRaises(exceptions.InputException):
      self.CallWrapped(self.issues_svcr.ListComments, mc, request)

  def testListComments_UnknownApprovalInFilter(self):
    """Filter with unknown approval returns no error and no comments."""
    approval_comment = tracker_pb2.IssueComment(
        id=123,
        issue_id=self.issue_1.issue_id,
        project_id=self.issue_1.project_id,
        user_id=self.owner.user_id,
        content='comment 2 - approval 1',
        approval_id=1)
    self.services.issue.TestAddComment(approval_comment, self.issue_1.local_id)
    request = issues_pb2.ListCommentsRequest(
        parent=self.issue_1_resource_name, page_size=1,
        filter='approval = "projects/chicken/approvalDefs/404"')
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    response = self.CallWrapped(self.issues_svcr.ListComments, mc, request)
    self.assertEqual(len(response.comments), 0)

  def testListComments_ApprovalInFilter(self):
    approval_comment = tracker_pb2.IssueComment(
        id=123,
        issue_id=self.issue_1.issue_id,
        project_id=self.issue_1.project_id,
        user_id=self.owner.user_id,
        content='comment 2 - approval 1',
        approval_id=1)
    self.services.issue.TestAddComment(approval_comment, self.issue_1.local_id)
    request = issues_pb2.ListCommentsRequest(
        parent=self.issue_1_resource_name, page_size=1,
        filter='approval = "projects/chicken/approvalDefs/1"')
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    response = self.CallWrapped(self.issues_svcr.ListComments, mc, request)
    self.assertEqual(len(response.comments), 1)
    self.assertEqual(response.comments[0].content, approval_comment.content)

  def testListApprovalValues(self):
    config = fake.MakeTestConfig(self.project_2.project_id, [], [])
    self.services.config.StoreConfig(self.cnxn, config)

    # Make regular field def and value
    fd_1 = fake.MakeTestFieldDef(
        1, self.project_2.project_id, tracker_pb2.FieldTypes.STR_TYPE,
        field_name='field1')
    self.services.config.TestAddFieldDef(fd_1)
    fv_1 = fake.MakeFieldValue(
        field_id=fd_1.field_id, str_value='value1', derived=False)

    # Make testing approval def and its associated field def
    approval_gate = fake.MakeTestFieldDef(
        2, self.project_2.project_id, tracker_pb2.FieldTypes.APPROVAL_TYPE,
        field_name='approval-gate-1')
    self.services.config.TestAddFieldDef(approval_gate)
    ad = fake.MakeTestApprovalDef(2, approver_ids=[self.user_2.user_id])
    self.services.config.TestAddApprovalDef(ad, self.project_2.project_id)

    # Make approval value
    av = fake.MakeApprovalValue(2, set_on=self.PAST_TIME,
          approver_ids=[self.user_2.user_id], setter_id=self.user_2.user_id)

    # Make field def that belongs to above approval_def
    fd_2 = fake.MakeTestFieldDef(
        3, self.project_2.project_id, tracker_pb2.FieldTypes.STR_TYPE,
        field_name='field2', approval_id=2)
    self.services.config.TestAddFieldDef(fd_2)
    fv_2 = fake.MakeFieldValue(
        field_id=fd_2.field_id, str_value='value2', derived=False)

    issue_resource_name = 'projects/cow/issues/1237'
    issue = fake.MakeTestIssue(
        self.project_2.project_id,
        1237,
        'sum',
        'New',
        self.user_2.user_id,
        project_name=self.project_2.project_name,
        field_values=[fv_1, fv_2],
        approval_values=[av])
    self.services.issue.TestAddIssue(issue)

    request = issues_pb2.ListApprovalValuesRequest(parent=issue_resource_name)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    actual_response = self.CallWrapped(
        self.issues_svcr.ListApprovalValues, mc, request)

    self.assertEqual(len(actual_response.approval_values), 1)
    expected_fv = issue_objects_pb2.FieldValue(
        field='projects/cow/fieldDefs/3',
        value='value2',
        derivation=issue_objects_pb2.Derivation.Value('EXPLICIT'))
    expected = issue_objects_pb2.ApprovalValue(
        name='projects/cow/issues/1237/approvalValues/2',
        status=issue_objects_pb2.ApprovalValue.ApprovalStatus.Value('NOT_SET'),
        approvers=['users/222'],
        approval_def='projects/cow/approvalDefs/2',
        set_time=timestamp_pb2.Timestamp(seconds=self.PAST_TIME),
        setter='users/222',
        field_values=[expected_fv])
    self.assertEqual(actual_response.approval_values[0], expected)

  def testListApprovalValues_Empty(self):
    request = issues_pb2.ListApprovalValuesRequest(
        parent=self.issue_1_resource_name)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    actual_response = self.CallWrapped(
        self.issues_svcr.ListApprovalValues, mc, request)
    self.assertEqual(len(actual_response.approval_values), 0)

  @mock.patch(
      'features.send_notifications.PrepareAndSendIssueChangeNotification')
  def testMakeIssue(self, _fake_pasicn):
    request_issue = issue_objects_pb2.Issue(
        summary='sum',
        status=issue_objects_pb2.Issue.StatusValue(status='New'),
        cc_users=[issue_objects_pb2.Issue.UserValue(user='users/222')],
        labels=[issue_objects_pb2.Issue.LabelValue(label='foo-bar')]
    )

    request = issues_pb2.MakeIssueRequest(
        parent='projects/chicken',
        issue=request_issue,
        description='description',
        uploads=[issues_pb2.AttachmentUpload(
            filename='mowgli.gif', content='cute dog')],
    )
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    response = self.CallWrapped(
        self.issues_svcr.MakeIssue, mc, request)
    self.assertEqual(response.summary, 'sum')
    self.assertEqual(response.status.status, 'New')
    self.assertEqual(response.cc_users[0].user, 'users/222')
    self.assertEqual(response.labels[0].label, 'foo-bar')
    self.assertEqual(response.star_count, 1)
    self.assertEqual(response.attachment_count, 1)

    unValid_request = issues_pb2.MakeIssueRequest(
        parent='projects/chicken',
        issue=request_issue,
        description='description',
        uploads=[issues_pb2.AttachmentUpload(
            filename='mowgli.gif')],
    )
    with self.assertRaisesRegexp(
      exceptions.InputException,
      'Uploaded atachment missing filename or content'):
      self.CallWrapped(self.issues_svcr.MakeIssue, mc, unValid_request)


  @mock.patch(
      'features.send_notifications.PrepareAndSendIssueChangeNotification')
  @mock.patch('time.time')
  def testModifyIssues(self, fake_time, fake_notify):
    fake_time.return_value = 12345

    issue = _Issue(780, 1)
    self.services.project.TestAddProject(
        issue.project_name, project_id=issue.project_id,
        owner_ids=[self.owner.user_id])

    issue.labels = ['keep-me', 'remove-me']
    self.services.issue.TestAddIssue(issue)
    exp_issue = copy.deepcopy(issue)

    self.services.issue.CreateIssueComment = mock.Mock()
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)

    request = issues_pb2.ModifyIssuesRequest(
        deltas=[
            issues_pb2.IssueDelta(
                issue=issue_objects_pb2.Issue(
                    name='projects/proj-780/issues/1',
                    labels=[issue_objects_pb2.Issue.LabelValue(
                        label='add-me')]),
                update_mask=field_mask_pb2.FieldMask(paths=['labels']),
                labels_remove=['remove-me'])],
        uploads=[issues_pb2.AttachmentUpload(
            filename='mowgli.gif', content='cute dog')],
        comment_content='Release the chicken.',
        notify_type=issues_pb2.NotifyType.Value('NO_NOTIFICATION'))

    response = self.CallWrapped(
        self.issues_svcr.ModifyIssues, mc, request)
    exp_issue.labels = ['keep-me', 'add-me']
    exp_issue.modified_timestamp = 12345
    exp_api_issue = self.issues_svcr.converter.ConvertIssue(exp_issue)
    self.assertEqual([iss for iss in response.issues], [exp_api_issue])

    # All updated issues should have been fetched from DB, skipping cache.
    # So we expect assume_stale=False was applied to all issues during the
    # the fetch.
    exp_issue.assume_stale = False
    # These derived values get set to the following when an issue goes through
    # the ApplyFilterRules path. (see filter_helpers._ComputeDerivedFields)
    exp_issue.derived_owner_id = 0
    exp_issue.derived_status = ''
    exp_attachments = [framework_helpers.AttachmentUpload(
        'mowgli.gif', 'cute dog', 'image/gif')]
    exp_amendments = [tracker_pb2.Amendment(
        field=tracker_pb2.FieldID.LABELS, newvalue='-remove-me add-me')]
    self.services.issue.CreateIssueComment.assert_called_once_with(
        self.cnxn, exp_issue, mc.auth.user_id, 'Release the chicken.',
        attachments=exp_attachments, amendments=exp_amendments, commit=False)
    fake_notify.assert_called_once_with(
        issue.issue_id, 'testing-app.appspot.com', self.owner.user_id,
        comment_id=mock.ANY, old_owner_id=None, send_email=False)

  def testModifyIssues_Empty(self):
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    request = issues_pb2.ModifyIssuesRequest()
    response = self.CallWrapped(self.issues_svcr.ModifyIssues, mc, request)
    self.assertEqual(response, issues_pb2.ModifyIssuesResponse())

  @mock.patch('api.v3.api_constants.MAX_MODIFY_ISSUES', 2)
  @mock.patch('api.v3.api_constants.MAX_MODIFY_IMPACTED_ISSUES', 4)
  def testModifyIssues_TooMany(self):
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    request = issues_pb2.ModifyIssuesRequest(
        deltas=[
            issues_pb2.IssueDelta(),
            issues_pb2.IssueDelta(),
            issues_pb2.IssueDelta()
        ])
    with self.assertRaisesRegexp(
        exceptions.InputException,
        'Requesting 3 updates when the allowed maximum is 2 updates.'):
      self.CallWrapped(self.issues_svcr.ModifyIssues, mc, request)

    issue_ref_list = [issue_objects_pb2.IssueRef()]
    request = issues_pb2.ModifyIssuesRequest(
        deltas=[
            issues_pb2.IssueDelta(
                issue=issue_objects_pb2.Issue(
                    blocked_on_issue_refs=issue_ref_list),
                blocked_on_issues_remove=issue_ref_list,
                update_mask=field_mask_pb2.FieldMask(
                    paths=['merged_into_issue_ref'])),
            issues_pb2.IssueDelta(
                issue=issue_objects_pb2.Issue(
                    blocking_issue_refs=issue_ref_list),
                blocking_issues_remove=issue_ref_list)
        ])
    with self.assertRaisesRegexp(
        exceptions.InputException,
        'Updates include 5 impacted issues when the allowed maximum is 4.'):
      self.CallWrapped(self.issues_svcr.ModifyIssues, mc, request)

  @mock.patch('time.time', mock.MagicMock(return_value=CURRENT_TIME))
  @mock.patch(
      'features.send_notifications.PrepareAndSendApprovalChangeNotification')
  def testModifyIssueApprovalValues(self, fake_notify):
    self.services.issue.DeltaUpdateIssueApproval = mock.Mock()
    config = fake.MakeTestConfig(self.project_1.project_id, [], [])
    self.services.config.StoreConfig(self.cnxn, config)

    # Make testing approval def and its associated field def
    field_id = 2
    approval_field_def = fake.MakeTestFieldDef(
        field_id,
        self.project_1.project_id,
        tracker_pb2.FieldTypes.APPROVAL_TYPE,
        field_name='approval-gate-1')
    self.services.config.TestAddFieldDef(approval_field_def)
    ad = fake.MakeTestApprovalDef(field_id, approver_ids=[self.owner.user_id])
    self.services.config.TestAddApprovalDef(ad, self.project_1.project_id)

    # Make approval value
    av = fake.MakeApprovalValue(
        field_id,
        status=tracker_pb2.ApprovalStatus.NEEDS_REVIEW,
        set_on=self.PAST_TIME,
        approver_ids=[self.owner.user_id],
        setter_id=self.user_2.user_id)

    issue = fake.MakeTestIssue(
        self.project_1.project_id,
        1237,
        'sum',
        'New',
        self.owner.user_id,
        project_name=self.project_1.project_name,
        approval_values=[av])
    self.services.issue.TestAddIssue(issue)

    av_name = 'projects/%s/issues/%d/approvalValues/%d' % (
        self.project_1.project_name, issue.local_id, ad.approval_id)
    delta = issues_pb2.ApprovalDelta(
        approval_value=issue_objects_pb2.ApprovalValue(
            name=av_name,
            status=issue_objects_pb2.ApprovalValue.ApprovalStatus.Value('NA')),
        update_mask=field_mask_pb2.FieldMask(paths=['status']))

    request = issues_pb2.ModifyIssueApprovalValuesRequest(deltas=[delta],)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    response = self.CallWrapped(
        self.issues_svcr.ModifyIssueApprovalValues, mc, request)
    expected_ingested_delta = tracker_pb2.ApprovalDelta(
        status=tracker_pb2.ApprovalStatus.NA,
        set_on=int(CURRENT_TIME),
        setter_id=self.owner.user_id,
    )
    # NOTE: Because we mock out DeltaUpdateIssueApproval, the ApprovalValues
    # returned haven't been changed in this test. We can't test that it was
    # changed correctly, but we can make sure it's for the right ApprovalValue.
    self.assertEqual(len(response.approval_values), 1)
    self.assertEqual(response.approval_values[0].name, av_name)
    self.services.issue.DeltaUpdateIssueApproval.assert_called_once_with(
        mc.cnxn,
        self.owner.user_id,
        config,
        issue,
        av,
        expected_ingested_delta,
        comment_content=u'',
        is_description=False,
        attachments=None,
        kept_attachments=None)
    fake_notify.assert_called_once_with(
        issue.issue_id,
        ad.approval_id,
        'testing-app.appspot.com',
        mock.ANY,
        send_email=True)

  @mock.patch('api.v3.api_constants.MAX_MODIFY_APPROVAL_VALUES', 2)
  def testModifyIssueApprovalValues_TooMany(self):
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    request = issues_pb2.ModifyIssueApprovalValuesRequest(
        deltas=[
            issues_pb2.ApprovalDelta(),
            issues_pb2.ApprovalDelta(),
            issues_pb2.ApprovalDelta()
        ])
    with self.assertRaises(exceptions.InputException):
      self.CallWrapped(self.issues_svcr.ModifyIssueApprovalValues, mc, request)

  def testModifyIssueApprovalValues_Empty(self):
    request = issues_pb2.ModifyIssueApprovalValuesRequest()
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    response = self.CallWrapped(
        self.issues_svcr.ModifyIssueApprovalValues, mc, request)
    self.assertEqual(len(response.approval_values), 0)

  @mock.patch(
      'businesslogic.work_env.WorkEnv.GetIssue',
      return_value=tracker_pb2.Issue(
          owner_id=0,
          project_name='chicken',
          project_id=789,
          local_id=1234,
          issue_id=80134))
  def testModifyCommentState(self, mocked_get_issue):
    name = self.issue_1_resource_name + '/comments/1'
    state = issue_objects_pb2.IssueContentState.Value('DELETED')
    request = issues_pb2.ModifyCommentStateRequest(name=name, state=state)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    with self.assertRaises(exceptions.NoSuchCommentException):
      self.CallWrapped(self.issues_svcr.ModifyCommentState, mc, request)
    mocked_get_issue.assert_any_call(self.issue_1.issue_id, use_cache=False)

  def testModifyCommentState_Delete(self):
    comment_1 = tracker_pb2.IssueComment(
        id=124,
        issue_id=self.issue_1.issue_id,
        project_id=self.issue_1.project_id,
        user_id=self.owner.user_id,
        content='first actual comment')
    self.services.issue.TestAddComment(comment_1, self.issue_1.local_id)

    name = self.issue_1_resource_name + '/comments/1'
    state = issue_objects_pb2.IssueContentState.Value('DELETED')
    request = issues_pb2.ModifyCommentStateRequest(name=name, state=state)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    response = self.CallWrapped(
        self.issues_svcr.ModifyCommentState, mc, request)
    self.assertEqual(response.comment.state, state)
    self.assertEqual(response.comment.content, 'first actual comment')

    # Test noop
    response = self.CallWrapped(
        self.issues_svcr.ModifyCommentState, mc, request)
    self.assertEqual(response.comment.state, state)

    # Test undelete
    state = issue_objects_pb2.IssueContentState.Value('ACTIVE')
    request = issues_pb2.ModifyCommentStateRequest(name=name, state=state)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    response = self.CallWrapped(
        self.issues_svcr.ModifyCommentState, mc, request)
    self.assertEqual(response.comment.state, state)

  @mock.patch(
      'framework.permissions.UpdateIssuePermissions',
      return_value=permissions.ADMIN_PERMISSIONSET)
  def testModifyCommentState_Spam(self, _mocked):
    comment_1 = tracker_pb2.IssueComment(
        id=124,
        issue_id=self.issue_1.issue_id,
        project_id=self.issue_1.project_id,
        user_id=self.owner.user_id,
        content='first actual comment')
    self.services.issue.TestAddComment(comment_1, self.issue_1.local_id)

    name = self.issue_1_resource_name + '/comments/1'
    state = issue_objects_pb2.IssueContentState.Value('SPAM')
    request = issues_pb2.ModifyCommentStateRequest(name=name, state=state)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    response = self.CallWrapped(
        self.issues_svcr.ModifyCommentState, mc, request)
    self.assertEqual(response.comment.state, state)

    # Test noop
    response = self.CallWrapped(
        self.issues_svcr.ModifyCommentState, mc, request)
    self.assertEqual(response.comment.state, state)

    # Test unflag as spam
    state = issue_objects_pb2.IssueContentState.Value('ACTIVE')
    request = issues_pb2.ModifyCommentStateRequest(name=name, state=state)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    response = self.CallWrapped(
        self.issues_svcr.ModifyCommentState, mc, request)
    self.assertEqual(response.comment.state, state)

  def testModifyCommentState_Active(self):
    comment_1 = tracker_pb2.IssueComment(
        id=124,
        issue_id=self.issue_1.issue_id,
        project_id=self.issue_1.project_id,
        user_id=self.owner.user_id,
        content='first actual comment')
    self.services.issue.TestAddComment(comment_1, self.issue_1.local_id)

    name = self.issue_1_resource_name + '/comments/1'
    state = issue_objects_pb2.IssueContentState.Value('ACTIVE')
    request = issues_pb2.ModifyCommentStateRequest(name=name, state=state)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    response = self.CallWrapped(
        self.issues_svcr.ModifyCommentState, mc, request)
    self.assertEqual(response.comment.state, state)

  def testModifyCommentState_Spam_ActionNotSupported(self):
    # Cannot transition from deleted to spam
    comment_1 = tracker_pb2.IssueComment(
        id=124,
        issue_id=self.issue_1.issue_id,
        project_id=self.issue_1.project_id,
        user_id=self.owner.user_id,
        content='first actual comment',
        deleted_by=self.owner.user_id)
    self.services.issue.TestAddComment(comment_1, self.issue_1.local_id)

    name = self.issue_1_resource_name + '/comments/1'
    state = issue_objects_pb2.IssueContentState.Value('SPAM')
    request = issues_pb2.ModifyCommentStateRequest(name=name, state=state)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    with self.assertRaises(exceptions.ActionNotSupported):
      self.CallWrapped(self.issues_svcr.ModifyCommentState, mc, request)

  def testModifyCommentState_Delete_ActionNotSupported(self):
    # Cannot transition from spam to deleted
    comment_1 = tracker_pb2.IssueComment(
        id=124,
        issue_id=self.issue_1.issue_id,
        project_id=self.issue_1.project_id,
        user_id=self.owner.user_id,
        content='first actual comment',
        is_spam=True)
    self.services.issue.TestAddComment(comment_1, self.issue_1.local_id)

    name = self.issue_1_resource_name + '/comments/1'
    state = issue_objects_pb2.IssueContentState.Value('DELETED')
    request = issues_pb2.ModifyCommentStateRequest(name=name, state=state)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    with self.assertRaises(exceptions.ActionNotSupported):
      self.CallWrapped(self.issues_svcr.ModifyCommentState, mc, request)

  def testModifyCommentState_NoSuchComment(self):
    name = self.issue_1_resource_name + '/comments/1'
    state = issue_objects_pb2.IssueContentState.Value('DELETED')
    request = issues_pb2.ModifyCommentStateRequest(name=name, state=state)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.owner.email)
    with self.assertRaises(exceptions.NoSuchCommentException):
      self.CallWrapped(self.issues_svcr.ModifyCommentState, mc, request)

  def testModifyCommentState_Delete_PermissionException(self):
    comment_1 = tracker_pb2.IssueComment(
        id=124,
        issue_id=self.issue_1.issue_id,
        project_id=self.issue_1.project_id,
        user_id=self.owner.user_id,
        content='first actual comment')
    self.services.issue.TestAddComment(comment_1, self.issue_1.local_id)

    name = self.issue_1_resource_name + '/comments/1'
    state = issue_objects_pb2.IssueContentState.Value('DELETED')
    request = issues_pb2.ModifyCommentStateRequest(name=name, state=state)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.user_2.email)
    with self.assertRaises(permissions.PermissionException):
      self.CallWrapped(self.issues_svcr.ModifyCommentState, mc, request)

  @mock.patch(
      'framework.permissions.UpdateIssuePermissions',
      return_value=permissions.READ_ONLY_PERMISSIONSET)
  def testModifyCommentState_Spam_PermissionException(self, _mocked):
    comment_1 = tracker_pb2.IssueComment(
        id=124,
        issue_id=self.issue_1.issue_id,
        project_id=self.issue_1.project_id,
        user_id=self.owner.user_id,
        content='first actual comment')
    self.services.issue.TestAddComment(comment_1, self.issue_1.local_id)

    name = self.issue_1_resource_name + '/comments/1'
    state = issue_objects_pb2.IssueContentState.Value('SPAM')
    request = issues_pb2.ModifyCommentStateRequest(name=name, state=state)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.user_2.email)
    with self.assertRaises(permissions.PermissionException):
      self.CallWrapped(self.issues_svcr.ModifyCommentState, mc, request)