api/v3/test/hotlists_servicer_test.py

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

import unittest

from google.protobuf import empty_pb2
from google.protobuf import field_mask_pb2

from api import resource_name_converters as rnc
from api.v3 import hotlists_servicer
from api.v3 import converters
from api.v3.api_proto import hotlists_pb2
from api.v3.api_proto import feature_objects_pb2
from api.v3.api_proto import issue_objects_pb2
from api.v3.api_proto import user_objects_pb2
from framework import exceptions
from framework import monorailcontext
from framework import permissions
from framework import sorting
from features import features_constants
from testing import fake
from services import features_svc
from services import service_manager


class HotlistsServicerTest(unittest.TestCase):

  def setUp(self):
    self.cnxn = fake.MonorailConnection()
    self.services = service_manager.Services(
        features=fake.FeaturesService(),
        issue=fake.IssueService(),
        project=fake.ProjectService(),
        config=fake.ConfigService(),
        cache_manager=fake.CacheManager(),
        user=fake.UserService(),
        usergroup=fake.UserGroupService())
    self.hotlists_svcr = hotlists_servicer.HotlistsServicer(
        self.services, make_rate_limiter=False)
    self.converter = None
    self.PAST_TIME = 12345
    self.user_1 = self.services.user.TestAddUser('user_111@example.com', 111)
    self.user_2 = self.services.user.TestAddUser('user_222@example.com', 222)
    self.user_3 = self.services.user.TestAddUser('user_333@example.com', 333)

    user_ids = [self.user_1.user_id, self.user_2.user_id, self.user_3.user_id]
    self.user_ids_to_name = rnc.ConvertUserNames(user_ids)

    self.project_1 = self.services.project.TestAddProject(
        'proj', project_id=789)

    self.issue_1 = fake.MakeTestIssue(
        self.project_1.project_id, 1, 'sum', 'New', 111,
        project_name=self.project_1.project_name)
    self.issue_2 = fake.MakeTestIssue(
        self.project_1.project_id, 2, 'sum', 'New', 111,
        project_name=self.project_1.project_name)
    self.issue_3 = fake.MakeTestIssue(
        self.project_1.project_id, 3, 'sum', 'New', 111,
        project_name=self.project_1.project_name)
    self.issue_4 = fake.MakeTestIssue(
        self.project_1.project_id, 4, 'sum', 'New', 111,
        project_name=self.project_1.project_name)
    self.issue_5 = fake.MakeTestIssue(
        self.project_1.project_id, 5, 'sum', 'New', 111,
        project_name=self.project_1.project_name)
    self.issue_6 = fake.MakeTestIssue(
        self.project_1.project_id, 6, 'sum', 'New', 111,
        project_name=self.project_1.project_name)
    self.services.issue.TestAddIssue(self.issue_1)
    self.services.issue.TestAddIssue(self.issue_2)
    self.services.issue.TestAddIssue(self.issue_3)
    self.services.issue.TestAddIssue(self.issue_4)
    self.services.issue.TestAddIssue(self.issue_5)
    self.services.issue.TestAddIssue(self.issue_6)
    issue_ids = [
        self.issue_1.issue_id, self.issue_2.issue_id, self.issue_3.issue_id,
        self.issue_4.issue_id, self.issue_5.issue_id, self.issue_6.issue_id
    ]
    self.issue_ids_to_name = rnc.ConvertIssueNames(
        self.cnxn, issue_ids, self.services)

    hotlist_items = [
        (
            self.issue_4.issue_id, 31, self.user_3.user_id, self.PAST_TIME,
            'note5'),
        (
            self.issue_3.issue_id, 21, self.user_1.user_id, self.PAST_TIME,
            'note1'),
        (
            self.issue_2.issue_id, 11, self.user_2.user_id, self.PAST_TIME,
            'note2'),
        (
            self.issue_1.issue_id, 1, self.user_1.user_id, self.PAST_TIME,
            'note4')
    ]
    self.hotlist_1 = self.services.features.TestAddHotlist(
        'HotlistName',
        summary='summary',
        description='description',
        owner_ids=[self.user_1.user_id],
        editor_ids=[self.user_2.user_id],
        hotlist_item_fields=hotlist_items,
        default_col_spec='',
        is_private=True)
    self.hotlist_resource_name = rnc.ConvertHotlistName(
        self.hotlist_1.hotlist_id)
    sorting.InitializeArtValues(self.services)

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

  # TODO(crbug/monorail/7104): Add page_token tests when implemented.
  def testListHotlistItems(self):
    """We can list a Hotlist's HotlistItems."""
    request = hotlists_pb2.ListHotlistItemsRequest(
        parent=self.hotlist_resource_name, page_size=2, order_by='note,stars')
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.user_1.email)
    mc.LookupLoggedInUserPerms(None)
    response = self.CallWrapped(
        self.hotlists_svcr.ListHotlistItems, mc, request)
    expected_items = self.converter.ConvertHotlistItems(
        self.hotlist_1.hotlist_id,
        [self.hotlist_1.items[1], self.hotlist_1.items[2]])
    self.assertEqual(
        response, hotlists_pb2.ListHotlistItemsResponse(items=expected_items))

  def testListHotlistItems_Empty(self):
    """We can return a response if the Hotlist has no items"""
    empty_hotlist = self.services.features.TestAddHotlist(
        'Empty',
        owner_ids=[self.user_1.user_id],
        editor_ids=[self.user_2.user_id],
        hotlist_item_fields=[])
    hotlist_resource_name = rnc.ConvertHotlistName(empty_hotlist.hotlist_id)
    request = hotlists_pb2.ListHotlistItemsRequest(parent=hotlist_resource_name)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.user_1.email)
    mc.LookupLoggedInUserPerms(None)
    response = self.CallWrapped(
        self.hotlists_svcr.ListHotlistItems, mc, request)
    self.assertEqual(response, hotlists_pb2.ListHotlistItemsResponse(items=[]))

  def testListHotlistItems_InvalidPageSize(self):
    """We raise an exception if `page_size` is negative."""
    request = hotlists_pb2.ListHotlistItemsRequest(
        parent=self.hotlist_resource_name, page_size=-1)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.user_1.email)
    with self.assertRaises(exceptions.InputException):
      self.CallWrapped(self.hotlists_svcr.ListHotlistItems, mc, request)

  def testListHotlistItems_DefaultPageSize(self):
    """We use our default page size when no `page_size` is given."""
    request = hotlists_pb2.ListHotlistItemsRequest(
        parent=self.hotlist_resource_name)
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.user_1.email)
    mc.LookupLoggedInUserPerms(None)
    response = self.CallWrapped(
        self.hotlists_svcr.ListHotlistItems, mc, request)
    self.assertEqual(
        len(response.items),
        min(
            features_constants.DEFAULT_RESULTS_PER_PAGE,
            len(self.hotlist_1.items)))

  def testRerankHotlistItems(self):
    """We can rerank a Hotlist."""
    item_names_dict = rnc.ConvertHotlistItemNames(
        self.cnxn, self.hotlist_1.hotlist_id,
        [item.issue_id for item in self.hotlist_1.items], self.services)
    request = hotlists_pb2.RerankHotlistItemsRequest(
        name=self.hotlist_resource_name,
        hotlist_items=[
            item_names_dict[self.issue_4.issue_id],
            item_names_dict[self.issue_3.issue_id]
        ],
        target_position=0)

    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.user_1.email)
    mc.LookupLoggedInUserPerms(None)
    self.CallWrapped(self.hotlists_svcr.RerankHotlistItems, mc, request)
    updated_hotlist = self.services.features.GetHotlist(
        self.cnxn, self.hotlist_1.hotlist_id)
    self.assertEqual(
        [item.issue_id for item in updated_hotlist.items],
        [self.issue_4.issue_id, self.issue_3.issue_id,
         self.issue_1.issue_id, self.issue_2.issue_id])

  def testRemoveHotlistItems(self):
    """We can remove items from a Hotlist."""
    issue_1_name = self.issue_ids_to_name[self.issue_1.issue_id]
    issue_2_name = self.issue_ids_to_name[self.issue_2.issue_id]
    request = hotlists_pb2.RemoveHotlistItemsRequest(
        parent=self.hotlist_resource_name, issues=[issue_1_name, issue_2_name])

    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.user_1.email)
    mc.LookupLoggedInUserPerms(None)
    self.CallWrapped(self.hotlists_svcr.RemoveHotlistItems, mc, request)
    updated_hotlist = self.services.features.GetHotlist(
        self.cnxn, self.hotlist_1.hotlist_id)
    # The hotlist used to have 4 items and we've removed two.
    self.assertEqual(len(updated_hotlist.items), 2)

  def testAddHotlistItems(self):
    """We can add items to a Hotlist."""
    issue_5_name = self.issue_ids_to_name[self.issue_5.issue_id]
    issue_6_name = self.issue_ids_to_name[self.issue_6.issue_id]
    request = hotlists_pb2.AddHotlistItemsRequest(
        parent=self.hotlist_resource_name, issues=[issue_5_name, issue_6_name])

    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.user_1.email)
    mc.LookupLoggedInUserPerms(None)
    self.CallWrapped(self.hotlists_svcr.AddHotlistItems, mc, request)
    updated_hotlist = self.services.features.GetHotlist(
        self.cnxn, self.hotlist_1.hotlist_id)
    # The hotlist used to have 4 items and we've added two.
    self.assertEqual(len(updated_hotlist.items), 6)

  def testRemoveHotlistEditors(self):
    """We can remove editors from a Hotlist."""
    user_2_name = self.user_ids_to_name[self.user_2.user_id]
    request = hotlists_pb2.RemoveHotlistEditorsRequest(
        name=self.hotlist_resource_name, editors=[user_2_name])

    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.user_1.email)
    mc.LookupLoggedInUserPerms(None)
    self.CallWrapped(self.hotlists_svcr.RemoveHotlistEditors, mc, request)
    updated_hotlist = self.services.features.GetHotlist(
        self.cnxn, self.hotlist_1.hotlist_id)
    # User 2 was the only editor in the hotlist, and we removed them.
    self.assertEqual(len(updated_hotlist.editor_ids), 0)

  def testGetHotlist(self):
    """We can get a Hotlist."""
    request = hotlists_pb2.GetHotlistRequest(name=self.hotlist_resource_name)

    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.user_1.email)
    mc.LookupLoggedInUserPerms(None)
    api_hotlist = self.CallWrapped(self.hotlists_svcr.GetHotlist, mc, request)
    self.assertEqual(api_hotlist, self.converter.ConvertHotlist(self.hotlist_1))

  def testGatherHotlistsForUser(self):
    """We can get all visible hotlists of a user."""
    request = hotlists_pb2.GatherHotlistsForUserRequest(
        user=self.user_ids_to_name[self.user_2.user_id])

    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.user_1.email)
    mc.LookupLoggedInUserPerms(None)
    response = self.CallWrapped(
        self.hotlists_svcr.GatherHotlistsForUser, mc, request)

    user_names_by_id = rnc.ConvertUserNames(
        [self.user_2.user_id, self.user_1.user_id])
    expected_api_hotlists = [
        feature_objects_pb2.Hotlist(
            name=self.hotlist_resource_name,
            display_name='HotlistName',
            summary='summary',
            description='description',
            hotlist_privacy=feature_objects_pb2.Hotlist.HotlistPrivacy.Value(
                'PRIVATE'),
            owner=user_names_by_id[self.user_1.user_id],
            editors=[user_names_by_id[self.user_2.user_id]])
    ]
    self.assertEqual(
        response,
        hotlists_pb2.GatherHotlistsForUserResponse(
            hotlists=expected_api_hotlists))

  def testUpdateHotlist_AllFields(self):
    """We can update a Hotlist."""
    request = hotlists_pb2.UpdateHotlistRequest(
        update_mask=field_mask_pb2.FieldMask(
            paths=[
                'summary',
                'description',
                'default_columns',
                'hotlist_privacy',
                'display_name',
                'owner',
                'editors',
            ]),
        hotlist=feature_objects_pb2.Hotlist(
            name=self.hotlist_resource_name,
            display_name='newName',
            summary='new summary',
            description='new description',
            default_columns=[
                issue_objects_pb2.IssuesListColumn(column='new-chicken-egg')
            ],
            hotlist_privacy=feature_objects_pb2.Hotlist.HotlistPrivacy.Value(
                'PUBLIC'),
            owner=self.user_ids_to_name[self.user_2.user_id],
            editors=[self.user_ids_to_name[self.user_3.user_id]]))
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.user_1.email)
    mc.LookupLoggedInUserPerms(None)
    api_hotlist = self.CallWrapped(
        self.hotlists_svcr.UpdateHotlist, mc, request)
    user_names_by_id = rnc.ConvertUserNames(
        [self.user_3.user_id, self.user_2.user_id, self.user_1.user_id])
    expected_hotlist = feature_objects_pb2.Hotlist(
        name=self.hotlist_resource_name,
        display_name='newName',
        summary='new summary',
        description='new description',
        default_columns=[
            issue_objects_pb2.IssuesListColumn(column='new-chicken-egg')
        ],
        hotlist_privacy=feature_objects_pb2.Hotlist.HotlistPrivacy.Value(
            'PUBLIC'),
        owner=user_names_by_id[self.user_2.user_id],
        editors=[
            user_names_by_id[self.user_2.user_id],
            user_names_by_id[self.user_3.user_id]
        ])
    self.assertEqual(api_hotlist, expected_hotlist)

  def testUpdateHotlist_OneField(self):
    request = hotlists_pb2.UpdateHotlistRequest(
        update_mask=field_mask_pb2.FieldMask(paths=['summary']),
        hotlist=feature_objects_pb2.Hotlist(
            name=self.hotlist_resource_name,
            display_name='newName',
            summary='new summary',
            description='new description',
            default_columns=[
                issue_objects_pb2.IssuesListColumn(column='new-chicken-egg')
            ],
            hotlist_privacy=feature_objects_pb2.Hotlist.HotlistPrivacy.Value(
                'PUBLIC')))
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.user_1.email)
    mc.LookupLoggedInUserPerms(None)
    api_hotlist = self.CallWrapped(
        self.hotlists_svcr.UpdateHotlist, mc, request)
    user_names_by_id = rnc.ConvertUserNames(
        [self.user_2.user_id, self.user_1.user_id])
    expected_hotlist = feature_objects_pb2.Hotlist(
        name=self.hotlist_resource_name,
        display_name='HotlistName',
        summary='new summary',
        description='description',
        default_columns=[],
        hotlist_privacy=feature_objects_pb2.Hotlist.HotlistPrivacy.Value(
            'PRIVATE'),
        owner=user_names_by_id[self.user_1.user_id],
        editors=[user_names_by_id[self.user_2.user_id]])
    self.assertEqual(api_hotlist, expected_hotlist)

  def testUpdateHotlist_EmptyFieldMask(self):
    request = hotlists_pb2.UpdateHotlistRequest(
        hotlist=feature_objects_pb2.Hotlist(summary='new'))
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.user_1.email)
    mc.LookupLoggedInUserPerms(None)
    with self.assertRaises(exceptions.InputException):
      self.CallWrapped(self.hotlists_svcr.UpdateHotlist, mc, request)

  def testUpdateHotlist_EmptyHotlist(self):
    request = hotlists_pb2.UpdateHotlistRequest(
        update_mask=field_mask_pb2.FieldMask(paths=['summary']))
    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.user_1.email)
    mc.LookupLoggedInUserPerms(None)
    with self.assertRaises(exceptions.InputException):
      self.CallWrapped(self.hotlists_svcr.UpdateHotlist, mc, request)

  def testDeleteHotlist(self):
    """We can delete a Hotlist."""
    request = hotlists_pb2.GetHotlistRequest(name=self.hotlist_resource_name)

    mc = monorailcontext.MonorailContext(
        self.services, cnxn=self.cnxn, requester=self.user_1.email)
    mc.LookupLoggedInUserPerms(None)
    api_response = self.CallWrapped(
        self.hotlists_svcr.DeleteHotlist, mc, request)
    self.assertEqual(api_response, empty_pb2.Empty())

    with self.assertRaises(features_svc.NoSuchHotlistException):
      self.services.features.GetHotlist(
          self.cnxn, self.hotlist_1.hotlist_id)