| # 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. |
| |
| """Fake object classes that are useful for unit tests.""" |
| from __future__ import print_function |
| from __future__ import division |
| from __future__ import absolute_import |
| |
| import collections |
| import itertools |
| import logging |
| import re |
| import sys |
| import time |
| |
| from six import string_types |
| |
| import settings |
| from features import filterrules_helpers |
| from framework import exceptions |
| from framework import framework_bizobj |
| from framework import framework_constants |
| from framework import framework_helpers |
| from framework import monorailrequest |
| from framework import permissions |
| from framework import profiler |
| from framework import validate |
| from mrproto import features_pb2 |
| from mrproto import project_pb2 |
| from mrproto import tracker_pb2 |
| from mrproto import user_pb2 |
| from mrproto import usergroup_pb2 |
| from services import caches |
| from services import config_svc |
| from services import features_svc |
| from services import project_svc |
| from tracker import tracker_bizobj |
| from tracker import tracker_constants |
| |
| # Many fakes return partial or constant values, regardless of their arguments. |
| # pylint: disable=unused-argument |
| |
| BOUNDARY = '-----thisisaboundary' |
| OWNER_ROLE = 'OWNER_ROLE' |
| COMMITTER_ROLE = 'COMMITTER_ROLE' |
| CONTRIBUTOR_ROLE = 'CONTRIBUTOR_ROLE' |
| EDITOR_ROLE = 'EDITOR_ROLE' |
| FOLLOWER_ROLE = 'FOLLOWER_ROLE' |
| |
| def Hotlist( |
| hotlist_name, hotlist_id, hotlist_item_fields=None, |
| is_private=False, owner_ids=None, editor_ids=None, follower_ids=None, |
| default_col_spec=None, summary=None, description=None): |
| hotlist_id = hotlist_id or hash(hotlist_name) |
| return features_pb2.MakeHotlist( |
| hotlist_name, hotlist_item_fields=hotlist_item_fields, |
| hotlist_id=hotlist_id, is_private=is_private, owner_ids=owner_ids or [], |
| editor_ids=editor_ids or [], follower_ids=follower_ids or [], |
| default_col_spec=default_col_spec, summary=summary, |
| description=description) |
| |
| def HotlistItem(issue_id, rank=None, adder_id=None, date_added=None, note=None): |
| return features_pb2.MakeHotlistItem(issue_id=issue_id, rank=rank, |
| adder_id=adder_id, date_added=date_added, |
| note=None) |
| |
| def Project( |
| project_name='proj', project_id=None, state=project_pb2.ProjectState.LIVE, |
| access=project_pb2.ProjectAccess.ANYONE, moved_to=None, |
| cached_content_timestamp=None, |
| owner_ids=None, committer_ids=None, contributor_ids=None): |
| """Returns a project protocol buffer with the given attributes.""" |
| project_id = project_id or hash(project_name) |
| return project_pb2.MakeProject( |
| project_name, project_id=project_id, state=state, access=access, |
| moved_to=moved_to, cached_content_timestamp=cached_content_timestamp, |
| owner_ids=owner_ids, committer_ids=committer_ids, |
| contributor_ids=contributor_ids) |
| |
| |
| def MakeTestFieldDef( |
| field_id, project_id, field_type, field_name='', applic_type=None, |
| applic_pred=None, is_required=False, is_niche=False, is_multivalued=False, |
| min_value=None, max_value=None, regex=None, needs_member=False, |
| needs_perm=None, grants_perm=None, notify_on=None, date_action_str=None, |
| docstring=None, admin_ids=None, editor_ids=None, approval_id=None, |
| is_phase_field=False, is_restricted_field=False): |
| return tracker_bizobj.MakeFieldDef( |
| field_id, project_id, field_name, field_type, applic_type, applic_pred, |
| is_required, is_niche, is_multivalued, min_value, max_value, regex, |
| needs_member, needs_perm, grants_perm, notify_on, date_action_str, |
| docstring, False, |
| approval_id=approval_id, is_phase_field=is_phase_field, |
| is_restricted_field=is_restricted_field, admin_ids=admin_ids, |
| editor_ids=editor_ids) |
| |
| def MakeTestApprovalDef(approval_id, approver_ids=None, survey=None): |
| return tracker_pb2.ApprovalDef( |
| approval_id=approval_id, |
| approver_ids = approver_ids, |
| survey = survey) |
| |
| def MakePhase(phase_id, name='', rank=0): |
| return tracker_pb2.Phase(phase_id=phase_id, name=name, rank=rank) |
| |
| |
| def MakeApprovalValue( |
| approval_id, |
| status=tracker_pb2.ApprovalStatus.NOT_SET, |
| setter_id=None, |
| set_on=None, |
| approver_ids=None, |
| phase_id=None): |
| if approver_ids is None: |
| approver_ids = [] |
| return tracker_pb2.ApprovalValue( |
| approval_id=approval_id, |
| status=status, |
| setter_id=setter_id, |
| set_on=set_on, |
| approver_ids=approver_ids, |
| phase_id=phase_id) |
| |
| |
| def MakeFieldValue( |
| field_id, |
| int_value=None, |
| str_value=None, |
| user_id=None, |
| date_value=None, |
| url_value=None, |
| derived=None, |
| phase_id=None): |
| return tracker_pb2.FieldValue( |
| field_id=field_id, |
| int_value=int_value, |
| str_value=str_value, |
| user_id=user_id, |
| date_value=date_value, |
| url_value=url_value, |
| derived=derived, |
| phase_id=phase_id) |
| |
| |
| def MakeTestIssue( |
| project_id, |
| local_id, |
| summary, |
| status, |
| owner_id, |
| labels=None, |
| derived_labels=None, |
| derived_status=None, |
| merged_into=0, |
| star_count=0, |
| derived_owner_id=0, |
| issue_id=None, |
| reporter_id=None, |
| opened_timestamp=None, |
| closed_timestamp=None, |
| modified_timestamp=None, |
| migration_modified_timestamp=None, |
| is_spam=False, |
| component_ids=None, |
| project_name=None, |
| field_values=None, |
| cc_ids=None, |
| derived_cc_ids=None, |
| assume_stale=True, |
| phases=None, |
| approval_values=None, |
| merged_into_external=None, |
| attachment_count=0, |
| derived_component_ids=None): |
| """Easily make an Issue for testing.""" |
| issue = tracker_pb2.Issue() |
| issue.project_id = project_id |
| issue.project_name = project_name |
| issue.local_id = local_id |
| issue.issue_id = issue_id if issue_id else 100000 + local_id |
| issue.reporter_id = reporter_id if reporter_id else owner_id |
| issue.summary = summary |
| issue.status = status |
| issue.owner_id = owner_id |
| issue.derived_owner_id = derived_owner_id |
| issue.star_count = star_count |
| issue.merged_into = merged_into |
| issue.merged_into_external = merged_into_external |
| issue.is_spam = is_spam |
| issue.attachment_count = attachment_count |
| if cc_ids: |
| issue.cc_ids = cc_ids |
| if derived_cc_ids: |
| issue.derived_cc_ids = derived_cc_ids |
| issue.assume_stale = assume_stale |
| if opened_timestamp: |
| issue.opened_timestamp = opened_timestamp |
| issue.owner_modified_timestamp = opened_timestamp |
| issue.status_modified_timestamp = opened_timestamp |
| issue.component_modified_timestamp = opened_timestamp |
| if modified_timestamp: |
| issue.modified_timestamp = modified_timestamp |
| # By default, make migration_modified_timestamp the same as |
| # modified_timestamp |
| issue.migration_modified_timestamp = modified_timestamp |
| if migration_modified_timestamp: |
| issue.migration_modified_timestamp = migration_modified_timestamp |
| if closed_timestamp: |
| issue.closed_timestamp = closed_timestamp |
| if labels is not None: |
| if isinstance(labels, string_types): |
| labels = labels.split() |
| issue.labels.extend(labels) |
| if derived_labels is not None: |
| if isinstance(derived_labels, string_types): |
| derived_labels = derived_labels.split() |
| issue.derived_labels.extend(derived_labels) |
| if derived_status is not None: |
| issue.derived_status = derived_status |
| if component_ids is not None: |
| issue.component_ids = component_ids |
| if derived_component_ids is not None: |
| issue.derived_component_ids = derived_component_ids |
| if field_values is not None: |
| issue.field_values = field_values |
| if phases is not None: |
| issue.phases = phases |
| if approval_values is not None: |
| issue.approval_values = approval_values |
| return issue |
| |
| |
| def MakeTestComponentDef(project_id, comp_id, path='', cc_ids=None): |
| if cc_ids is None: |
| cc_ids = [] |
| return tracker_bizobj.MakeComponentDef( |
| comp_id, project_id, path, '', False, [], cc_ids, None, None) |
| |
| |
| def MakeTestConfig(project_id, labels, statuses): |
| """Convenient function to make a ProjectIssueConfig object.""" |
| config = tracker_bizobj.MakeDefaultProjectIssueConfig(project_id) |
| if isinstance(labels, string_types): |
| labels = labels.split() |
| if isinstance(statuses, string_types): |
| statuses = statuses.split() |
| config.well_known_labels = [ |
| tracker_pb2.LabelDef(label=lab) for lab in labels] |
| config.well_known_statuses = [ |
| tracker_pb2.StatusDef(status=stat) for stat in statuses] |
| return config |
| |
| |
| class MonorailConnection(object): |
| """Fake connection to databases for use in tests.""" |
| |
| def Commit(self): |
| pass |
| |
| def Close(self): |
| pass |
| |
| |
| class MonorailRequest(monorailrequest.MonorailRequest): |
| """Subclass of MonorailRequest suitable for testing.""" |
| |
| def __init__(self, services, user_info=None, project=None, perms=None, |
| hotlist=None, **kwargs): |
| """Construct a test MonorailRequest. |
| |
| Typically, this is constructed via testing.helpers.GetRequestObjects, |
| which also causes url parsing and optionally initializes the user, |
| project, and permissions info. |
| |
| Args: |
| services: connections to backends. |
| user_info: a dict of user attributes to set on a MonorailRequest object. |
| For example, "user_id: 5" causes self.auth.user_id=5. |
| project: the Project pb for this request. |
| perms: a PermissionSet for this request. |
| """ |
| super(MonorailRequest, self).__init__(services, **kwargs) |
| |
| if user_info is not None: |
| for key in user_info: |
| setattr(self.auth, key, user_info[key]) |
| if 'user_id' in user_info: |
| self.auth.effective_ids = {user_info['user_id']} |
| |
| self.perms = perms or permissions.ADMIN_PERMISSIONSET |
| self.profiler = profiler.Profiler() |
| self.project = project |
| self.hotlist = hotlist |
| if hotlist is not None: |
| self.hotlist_id = hotlist.hotlist_id |
| |
| class UserGroupService(object): |
| """Fake UserGroupService class for testing other code.""" |
| |
| def __init__(self): |
| # Test-only sequence of expunged users. |
| self.expunged_users_in_groups = [] |
| |
| self.group_settings = {} |
| self.group_members = {} |
| self.group_addrs = {} |
| self.role_dict = {} |
| |
| def TestAddGroupSettings( |
| self, |
| group_id, |
| email, |
| who_can_view=None, |
| anyone_can_join=False, |
| who_can_add=None, |
| external_group_type=None, |
| last_sync_time=0, |
| friend_projects=None, |
| notify_members=True, |
| notify_group=False): |
| """Set up a fake group for testing. |
| |
| Args: |
| group_id: int user ID of the new user group. |
| email: string email address to identify the user group. |
| who_can_view: string enum 'owners', 'members', or 'anyone'. |
| anyone_can_join: optional boolean to allow any users to join the group. |
| who_can_add: optional list of int user IDs of users who can add |
| more members to the group. |
| notify_members: optional boolean for if emails to this group should be |
| sent directly to members. |
| notify_group: optional boolean for if emails to this group should be |
| sent directly to the group email. |
| """ |
| friend_projects = friend_projects or [] |
| group_settings = usergroup_pb2.MakeSettings( |
| who_can_view or 'members', external_group_type, last_sync_time, |
| friend_projects, notify_members, notify_group) |
| self.group_settings[group_id] = group_settings |
| self.group_addrs[group_id] = email |
| # TODO(jrobbins): store the other settings. |
| |
| def TestAddMembers(self, group_id, user_ids, role='member'): |
| self.group_members.setdefault(group_id, []).extend(user_ids) |
| for user_id in user_ids: |
| self.role_dict.setdefault(group_id, {})[user_id] = role |
| |
| def LookupAllMemberships(self, _cnxn, user_ids, use_cache=True): |
| return { |
| user_id: self.LookupMemberships(_cnxn, user_id) |
| for user_id in user_ids |
| } |
| |
| def LookupMemberships(self, _cnxn, user_id): |
| memberships = { |
| group_id for group_id, member_ids in self.group_members.items() |
| if user_id in member_ids} |
| return memberships |
| |
| def DetermineWhichUserIDsAreGroups(self, _cnxn, user_ids): |
| return [uid for uid in user_ids |
| if uid in self.group_settings] |
| |
| def GetAllUserGroupsInfo(self, cnxn): |
| infos = [] |
| for group_id in self.group_settings: |
| infos.append( |
| (self.group_addrs[group_id], |
| len(self.group_members.get(group_id, [])), |
| self.group_settings[group_id], group_id)) |
| |
| return infos |
| |
| def GetAllGroupSettings(self, _cnxn, group_ids): |
| return {gid: self.group_settings[gid] |
| for gid in group_ids |
| if gid in self.group_settings} |
| |
| def GetGroupSettings(self, cnxn, group_id): |
| return self.GetAllGroupSettings(cnxn, [group_id]).get(group_id) |
| |
| def CreateGroup(self, cnxn, services, email, who_can_view_members, |
| ext_group_type=None, friend_projects=None): |
| friend_projects = friend_projects or [] |
| group_id = services.user.LookupUserID( |
| cnxn, email, autocreate=True, allowgroups=True) |
| self.group_addrs[group_id] = email |
| group_settings = usergroup_pb2.MakeSettings( |
| who_can_view_members, ext_group_type, 0, friend_projects) |
| self.UpdateSettings(cnxn, group_id, group_settings) |
| return group_id |
| |
| def DeleteGroups(self, cnxn, group_ids): |
| member_ids_dict, owner_ids_dict = self.LookupMembers(cnxn, group_ids) |
| citizens_id_dict = collections.defaultdict(list) |
| for g_id, user_ids in member_ids_dict.items(): |
| citizens_id_dict[g_id].extend(user_ids) |
| for g_id, user_ids in owner_ids_dict.items(): |
| citizens_id_dict[g_id].extend(user_ids) |
| for g_id, citizen_ids in citizens_id_dict.items(): |
| # Remove group members, friend projects and settings |
| self.RemoveMembers(cnxn, g_id, citizen_ids) |
| self.group_settings.pop(g_id, None) |
| |
| def LookupComputedMemberships(self, cnxn, domain, use_cache=True): |
| group_email = 'everyone@%s' % domain |
| group_id = self.LookupUserGroupID(cnxn, group_email, use_cache=use_cache) |
| if group_id: |
| return [group_id] |
| |
| return [] |
| |
| def LookupUserGroupID(self, cnxn, group_email, use_cache=True): |
| for group_id in self.group_settings: |
| if group_email == self.group_addrs.get(group_id): |
| return group_id |
| return None |
| |
| def LookupMembers(self, _cnxn, group_id_list): |
| members_dict = {} |
| owners_dict = {} |
| for gid in group_id_list: |
| members_dict[gid] = [] |
| owners_dict[gid] = [] |
| for mid in self.group_members.get(gid, []): |
| if self.role_dict.get(gid, {}).get(mid) == 'owner': |
| owners_dict[gid].append(mid) |
| elif self.role_dict.get(gid, {}).get(mid) == 'member': |
| members_dict[gid].append(mid) |
| return members_dict, owners_dict |
| |
| def LookupAllMembers(self, _cnxn, group_id_list): |
| direct_members, direct_owners = self.LookupMembers( |
| _cnxn, group_id_list) |
| members_dict = {} |
| owners_dict = {} |
| for gid in group_id_list: |
| members = direct_members[gid] |
| owners = direct_owners[gid] |
| owners_dict[gid] = owners |
| members_dict[gid] = members |
| group_ids = set([uid for uid in members + owners |
| if uid in self.group_settings]) |
| while group_ids: |
| indirect_members, indirect_owners = self.LookupMembers( |
| _cnxn, group_ids) |
| child_members = set() |
| child_owners = set() |
| for _, children in indirect_members.items(): |
| child_members.update(children) |
| for _, children in indirect_owners.items(): |
| child_owners.update(children) |
| members_dict[gid].extend(list(child_members)) |
| owners_dict[gid].extend(list(child_owners)) |
| group_ids = set(self.DetermineWhichUserIDsAreGroups( |
| _cnxn, list(child_members) + list(child_owners))) |
| members_dict[gid] = list(set(members_dict[gid])) |
| return members_dict, owners_dict |
| |
| |
| def RemoveMembers(self, _cnxn, group_id, old_member_ids): |
| current_member_ids = self.group_members.get(group_id, []) |
| revised_member_ids = [mid for mid in current_member_ids |
| if mid not in old_member_ids] |
| self.group_members[group_id] = revised_member_ids |
| |
| def UpdateMembers(self, _cnxn, group_id, member_ids, new_role): |
| self.RemoveMembers(_cnxn, group_id, member_ids) |
| self.TestAddMembers(group_id, member_ids, new_role) |
| |
| def UpdateSettings(self, _cnxn, group_id, group_settings): |
| self.group_settings[group_id] = group_settings |
| |
| def ExpandAnyGroupEmailRecipients(self, cnxn, user_ids): |
| group_ids = set(self.DetermineWhichUserIDsAreGroups(cnxn, user_ids)) |
| group_settings_dict = self.GetAllGroupSettings(cnxn, group_ids) |
| member_ids_dict, owner_ids_dict = self.LookupAllMembers(cnxn, group_ids) |
| indirect_ids = set() |
| direct_ids = {uid for uid in user_ids if uid not in group_ids} |
| for gid, group_settings in group_settings_dict.items(): |
| if group_settings.notify_members: |
| indirect_ids.update(member_ids_dict.get(gid, set())) |
| indirect_ids.update(owner_ids_dict.get(gid, set())) |
| if group_settings.notify_group: |
| direct_ids.add(gid) |
| |
| return list(direct_ids), list(indirect_ids) |
| |
| def LookupVisibleMembers( |
| self, cnxn, group_id_list, perms, effective_ids, services): |
| settings_dict = self.GetAllGroupSettings(cnxn, group_id_list) |
| group_ids = list(settings_dict.keys()) |
| |
| direct_member_ids_dict, direct_owner_ids_dict = self.LookupMembers( |
| cnxn, group_ids) |
| all_member_ids_dict, all_owner_ids_dict = self.LookupAllMembers( |
| cnxn, group_ids) |
| visible_member_ids_dict = {} |
| visible_owner_ids_dict = {} |
| for gid in group_ids: |
| member_ids = all_member_ids_dict[gid] |
| owner_ids = all_owner_ids_dict[gid] |
| if permissions.CanViewGroupMembers( |
| perms, effective_ids, settings_dict[gid], member_ids, owner_ids, []): |
| visible_member_ids_dict[gid] = direct_member_ids_dict[gid] |
| visible_owner_ids_dict[gid] = direct_owner_ids_dict[gid] |
| |
| return visible_member_ids_dict, visible_owner_ids_dict |
| |
| def ValidateFriendProjects(self, cnxn, services, friend_projects): |
| project_names = list(filter(None, re.split('; |, | |;|,', friend_projects))) |
| id_dict = services.project.LookupProjectIDs(cnxn, project_names) |
| missed_projects = [] |
| result = [] |
| for p_name in project_names: |
| if p_name in id_dict: |
| result.append(id_dict[p_name]) |
| else: |
| missed_projects.append(p_name) |
| error_msg = '' |
| if missed_projects: |
| error_msg = 'Project(s) %s do not exist' % ', '.join(missed_projects) |
| return None, error_msg |
| else: |
| return result, None |
| |
| def ExpungeUsersInGroups(self, cnxn, ids): |
| self.expunged_users_in_groups.extend(ids) |
| |
| |
| class CacheManager(object): |
| |
| def __init__(self, invalidate_tbl=None): |
| self.last_call = None |
| self.cache_registry = collections.defaultdict(list) |
| self.processed_invalidations_up_to = 0 |
| |
| def RegisterCache(self, cache, kind): |
| """Register a cache to be notified of future invalidations.""" |
| self.cache_registry[kind].append(cache) |
| |
| def DoDistributedInvalidation(self, cnxn): |
| """Drop any cache entries that were invalidated by other jobs.""" |
| self.last_call = 'DoDistributedInvalidation', cnxn |
| |
| def StoreInvalidateRows(self, cnxn, kind, keys): |
| """Store database rows to let all frontends know to invalidate.""" |
| self.last_call = 'StoreInvalidateRows', cnxn, kind, keys |
| |
| def StoreInvalidateAll(self, cnxn, kind): |
| """Store a database row to let all frontends know to invalidate.""" |
| self.last_call = 'StoreInvalidateAll', cnxn, kind |
| |
| |
| |
| class UserService(object): |
| |
| def __init__(self): |
| """Creates a test-appropriate UserService object.""" |
| self.users_by_email = {} # {email: user_id, ...} |
| self.users_by_id = {} # {user_id: email, ...} |
| self.test_users = {} # {user_id: user_pb, ...} |
| self.visited_hotlists = {} # user_id:[(hotlist_id, viewed), ...] |
| self.invite_rows = [] # (parent_id, child_id) |
| self.linked_account_rows = [] # (parent_id, child_id) |
| self.prefs_dict = {} # {user_id: UserPrefs} |
| |
| def TestAddUser( |
| self, email, user_id, add_user=True, banned=False, obscure_email=True): |
| """Add a user to the fake UserService instance. |
| |
| Args: |
| email: Email of the user. |
| user_id: int user ID. |
| add_user: Flag whether user pb should be created, i.e. whether a |
| Monorail account should be created |
| banned: Boolean to set the user as banned |
| obscure_email: Boolean to determine whether to obscure the user's email. |
| |
| Returns: |
| The User PB that was added, or None. |
| """ |
| self.users_by_email[email] = user_id |
| self.users_by_id[user_id] = email |
| |
| user = None |
| if add_user: |
| user = user_pb2.MakeUser(user_id) |
| user.is_site_admin = False |
| user.email = email |
| user.obscure_email = obscure_email |
| if banned: |
| user.banned = 'is banned' |
| self.test_users[user_id] = user |
| |
| return user |
| |
| def GetUser(self, cnxn, user_id): |
| return self.GetUsersByIDs(cnxn, [user_id])[user_id] |
| |
| def _CreateUser(self, _cnxn, email): |
| if email in self.users_by_email: |
| return |
| user_id = framework_helpers.MurmurHash3_x86_32(email) |
| self.TestAddUser(email, user_id) |
| |
| def _CreateUsers(self, cnxn, emails): |
| for email in emails: |
| self._CreateUser(cnxn, email) |
| |
| def LookupUserID(self, cnxn, email, autocreate=False, allowgroups=False): |
| email_dict = self.LookupUserIDs( |
| cnxn, [email], autocreate=autocreate, allowgroups=allowgroups) |
| if email in email_dict: |
| return email_dict[email] |
| raise exceptions.NoSuchUserException('%r not found' % email) |
| |
| def GetUsersByIDs(self, cnxn, user_ids, use_cache=True, skip_missed=False): |
| user_dict = {} |
| for user_id in user_ids: |
| if user_id and self.test_users.get(user_id): |
| user_dict[user_id] = self.test_users[user_id] |
| elif not skip_missed: |
| user_dict[user_id] = user_pb2.MakeUser(user_id) |
| return user_dict |
| |
| def LookupExistingUserIDs(self, cnxn, emails): |
| email_dict = { |
| email: self.users_by_email[email] |
| for email in emails |
| if email in self.users_by_email} |
| return email_dict |
| |
| def LookupUserIDs(self, cnxn, emails, autocreate=False, |
| allowgroups=False): |
| email_dict = {} |
| needed_emails = [email.lower() for email in emails |
| if email |
| and not framework_constants.NO_VALUE_RE.match(email)] |
| for email in needed_emails: |
| user_id = self.users_by_email.get(email) |
| if not user_id: |
| if autocreate and validate.IsValidEmail(email): |
| self._CreateUser(cnxn, email) |
| user_id = self.users_by_email.get(email) |
| elif not autocreate: |
| raise exceptions.NoSuchUserException('%r' % email) |
| if user_id: |
| email_dict[email] = user_id |
| return email_dict |
| |
| def LookupUserEmail(self, _cnxn, user_id): |
| email = self.users_by_id.get(user_id) |
| if not email: |
| raise exceptions.NoSuchUserException('No user has ID %r' % user_id) |
| return email |
| |
| def LookupUserEmails(self, cnxn, user_ids, ignore_missed=False): |
| if ignore_missed: |
| user_dict = {} |
| for user_id in user_ids: |
| try: |
| user_dict[user_id] = self.LookupUserEmail(cnxn, user_id) |
| except exceptions.NoSuchUserException: |
| continue |
| return user_dict |
| user_dict = { |
| user_id: self.LookupUserEmail(cnxn, user_id) |
| for user_id in user_ids} |
| return user_dict |
| |
| def UpdateUser(self, _cnxn, user_id, user): |
| """Updates the user pb.""" |
| self.test_users[user_id] = user |
| |
| def UpdateUserBan(self, _cnxn, user_id, user, is_banned=None, |
| banned_reason=None): |
| """Updates the user pb.""" |
| self.test_users[user_id] = user |
| user.banned = banned_reason if is_banned else '' |
| |
| def GetPendingLinkedInvites(self, cnxn, user_id): |
| invite_as_parent = [row[1] for row in self.invite_rows |
| if row[0] == user_id] |
| invite_as_child = [row[0] for row in self.invite_rows |
| if row[1] == user_id] |
| return invite_as_parent, invite_as_child |
| |
| def InviteLinkedParent(self, cnxn, parent_id, child_id): |
| self.invite_rows.append((parent_id, child_id)) |
| |
| def AcceptLinkedChild(self, cnxn, parent_id, child_id): |
| if (parent_id, child_id) not in self.invite_rows: |
| raise exceptions.InputException('No such invite') |
| self.linked_account_rows.append((parent_id, child_id)) |
| self.invite_rows = [ |
| (p_id, c_id) for (p_id, c_id) in self.invite_rows |
| if p_id != parent_id and c_id != child_id] |
| self.GetUser(cnxn, parent_id).linked_child_ids.append(child_id) |
| self.GetUser(cnxn, child_id).linked_parent_id = parent_id |
| |
| def UnlinkAccounts(self, _cnxn, parent_id, child_id): |
| """Delete a linked-account relationship.""" |
| if not parent_id: |
| raise exceptions.InputException('Parent account is missing') |
| if not child_id: |
| raise exceptions.InputException('Child account is missing') |
| self.linked_account_rows = [(p, c) for (p, c) in self.linked_account_rows |
| if (p, c) != (parent_id, child_id)] |
| |
| def UpdateUserSettings( |
| self, cnxn, user_id, user, notify=None, notify_starred=None, |
| email_compact_subject=None, email_view_widget=None, |
| notify_starred_ping=None, obscure_email=None, after_issue_update=None, |
| is_site_admin=None, is_banned=None, banned_reason=None, |
| keep_people_perms_open=None, preview_on_hover=None, |
| vacation_message=None): |
| # notifications |
| if notify is not None: |
| user.notify_issue_change = notify |
| if notify_starred is not None: |
| user.notify_starred_issue_change = notify_starred |
| if notify_starred_ping is not None: |
| user.notify_starred_ping = notify_starred_ping |
| if email_compact_subject is not None: |
| user.email_compact_subject = email_compact_subject |
| if email_view_widget is not None: |
| user.email_view_widget = email_view_widget |
| |
| # display options |
| if after_issue_update is not None: |
| user.after_issue_update = user_pb2.IssueUpdateNav(after_issue_update) |
| if preview_on_hover is not None: |
| user.preview_on_hover = preview_on_hover |
| if keep_people_perms_open is not None: |
| user.keep_people_perms_open = keep_people_perms_open |
| |
| # misc |
| if obscure_email is not None: |
| user.obscure_email = obscure_email |
| |
| # admin |
| if is_site_admin is not None: |
| user.is_site_admin = is_site_admin |
| if is_banned is not None: |
| if is_banned: |
| user.banned = banned_reason or 'No reason given' |
| else: |
| user.reset('banned') |
| |
| # user availability |
| if vacation_message is not None: |
| user.vacation_message = vacation_message |
| |
| return self.UpdateUser(cnxn, user_id, user) |
| |
| def GetUsersPrefs(self, cnxn, user_ids, use_cache=True): |
| for user_id in user_ids: |
| if user_id not in self.prefs_dict: |
| self.prefs_dict[user_id] = user_pb2.UserPrefs(user_id=user_id) |
| return self.prefs_dict |
| |
| def GetUserPrefs(self, cnxn, user_id, use_cache=True): |
| """Return a UserPrefs PB for the requested user ID.""" |
| prefs_dict = self.GetUsersPrefs(cnxn, [user_id], use_cache=use_cache) |
| return prefs_dict[user_id] |
| |
| def GetUserPrefsByEmail(self, cnxn, email, use_cache=True): |
| """Return a UserPrefs PB for the requested email, or an empty UserPrefs.""" |
| try: |
| user_id = self.LookupUserID(cnxn, email) |
| user_prefs = self.GetUserPrefs(cnxn, user_id, use_cache=use_cache) |
| except exceptions.NoSuchUserException: |
| user_prefs = user_pb2.UserPrefs() |
| return user_prefs |
| |
| def SetUserPrefs(self, cnxn, user_id, pref_values): |
| userprefs = self.GetUserPrefs(cnxn, user_id) |
| names_to_overwrite = {upv.name for upv in pref_values} |
| userprefs.prefs = [upv for upv in userprefs.prefs |
| if upv.name not in names_to_overwrite] |
| userprefs.prefs.extend(pref_values) |
| |
| def ExpungeUsers(self, cnxn, user_ids): |
| for user_id in user_ids: |
| self.test_users.pop(user_id, None) |
| self.prefs_dict.pop(user_id, None) |
| email = self.users_by_id.pop(user_id, None) |
| if email: |
| self.users_by_email.pop(email, None) |
| |
| self.invite_rows = [row for row in self.invite_rows |
| if row[0] not in user_ids and row[1] not in user_ids] |
| self.linked_account_rows = [ |
| row for row in self.linked_account_rows |
| if row[0] not in user_ids and row[1] not in user_ids] |
| |
| def TotalUsersCount(self, cnxn): |
| return len(self.users_by_id) - 1 if ( |
| framework_constants.DELETED_USER_ID in self.users_by_id |
| ) else len(self.users_by_id) |
| |
| def GetAllUserEmailsBatch(self, cnxn, limit=1000, offset=0): |
| sorted_user_ids = sorted(self.users_by_id.keys()) |
| sorted_user_ids = [ |
| user_id for user_id in sorted_user_ids |
| if user_id != framework_constants.DELETED_USER_ID] |
| emails = [] |
| for i in range(offset, offset + limit): |
| try: |
| user_id = sorted_user_ids[i] |
| if user_id != framework_constants.DELETED_USER_ID: |
| emails.append(self.users_by_id[user_id]) |
| except IndexError: |
| break |
| return emails |
| |
| def GetRecentlyVisitedHotlists(self, _cnxn, user_id): |
| try: |
| return self.visited_hotlists[user_id] |
| except KeyError: |
| return [] |
| |
| def AddVisitedHotlist(self, _cnxn, user_id, hotlist_id, commit=True): |
| try: |
| user_visited_tuples = self.visited_hotlists[user_id] |
| self.visited_hotlists[user_id] = [ |
| hid for hid in user_visited_tuples if hid != hotlist_id] |
| except KeyError: |
| self.visited_hotlists[user_id] = [] |
| self.visited_hotlists[user_id].append(hotlist_id) |
| |
| def ExpungeUsersHotlistsHistory(self, cnxn, user_ids, commit=True): |
| for user_id in user_ids: |
| self.visited_hotlists.pop(user_id, None) |
| |
| |
| class AbstractStarService(object): |
| """Fake StarService.""" |
| |
| def __init__(self): |
| self.stars_by_item_id = {} |
| self.stars_by_starrer_id = {} |
| self.expunged_item_ids = [] |
| |
| def ExpungeStars(self, _cnxn, item_id, commit=True, limit=None): |
| self.expunged_item_ids.append(item_id) |
| old_starrers = self.stars_by_item_id.get(item_id, []) |
| self.stars_by_item_id[item_id] = [] |
| for old_starrer in old_starrers: |
| if self.stars_by_starrer_id.get(old_starrer): |
| self.stars_by_starrer_id[old_starrer] = [ |
| it for it in self.stars_by_starrer_id[old_starrer] |
| if it != item_id] |
| |
| def ExpungeStarsByUsers(self, _cnxn, user_ids, limit=None): |
| for user_id in user_ids: |
| item_ids = self.stars_by_starrer_id.pop(user_id, []) |
| for item_id in item_ids: |
| starrers = self.stars_by_item_id.get(item_id, None) |
| if starrers: |
| self.stars_by_item_id[item_id] = [ |
| starrer for starrer in starrers if starrer != user_id] |
| |
| def LookupItemStarrers(self, _cnxn, item_id): |
| return self.stars_by_item_id.get(item_id, []) |
| |
| def LookupItemsStarrers(self, cnxn, item_ids): |
| return { |
| item_id: self.LookupItemStarrers(cnxn, item_id) for item_id in item_ids} |
| |
| def LookupStarredItemIDs(self, _cnxn, starrer_user_id): |
| return self.stars_by_starrer_id.get(starrer_user_id, []) |
| |
| def IsItemStarredBy(self, cnxn, item_id, starrer_user_id): |
| return item_id in self.LookupStarredItemIDs(cnxn, starrer_user_id) |
| |
| def CountItemStars(self, cnxn, item_id): |
| return len(self.LookupItemStarrers(cnxn, item_id)) |
| |
| def CountItemsStars(self, cnxn, item_ids): |
| return {item_id: self.CountItemStars(cnxn, item_id) |
| for item_id in item_ids} |
| |
| def _SetStar(self, cnxn, item_id, starrer_user_id, starred): |
| if starred and not self.IsItemStarredBy(cnxn, item_id, starrer_user_id): |
| self.stars_by_item_id.setdefault(item_id, []).append(starrer_user_id) |
| self.stars_by_starrer_id.setdefault(starrer_user_id, []).append(item_id) |
| |
| elif not starred and self.IsItemStarredBy(cnxn, item_id, starrer_user_id): |
| self.stars_by_item_id[item_id].remove(starrer_user_id) |
| self.stars_by_starrer_id[starrer_user_id].remove(item_id) |
| |
| def SetStar(self, cnxn, item_id, starrer_user_id, starred): |
| self._SetStar(cnxn, item_id, starrer_user_id, starred) |
| |
| def SetStarsBatch( |
| self, cnxn, item_id, starrer_user_ids, starred, commit=True): |
| for starrer_user_id in starrer_user_ids: |
| self._SetStar(cnxn, item_id, starrer_user_id, starred) |
| |
| |
| class UserStarService(AbstractStarService): |
| pass |
| |
| |
| class ProjectStarService(AbstractStarService): |
| pass |
| |
| |
| class HotlistStarService(AbstractStarService): |
| pass |
| |
| |
| class IssueStarService(AbstractStarService): |
| |
| # pylint: disable=arguments-differ |
| def SetStar( |
| self, cnxn, services, _config, issue_id, starrer_user_id, |
| starred): |
| super(IssueStarService, self).SetStar( |
| cnxn, issue_id, starrer_user_id, starred) |
| try: |
| issue = services.issue.GetIssue(cnxn, issue_id) |
| issue.star_count += (1 if starred else -1) |
| except exceptions.NoSuchIssueException: |
| pass |
| |
| # pylint: disable=arguments-differ |
| def SetStarsBatch( |
| self, cnxn, _service, _config, issue_id, starrer_user_ids, |
| starred): |
| super(IssueStarService, self).SetStarsBatch( |
| cnxn, issue_id, starrer_user_ids, starred) |
| |
| def SetStarsBatch_SkipIssueUpdate( |
| self, cnxn, issue_id, starrer_user_ids, starred, commit=True): |
| super(IssueStarService, self).SetStarsBatch( |
| cnxn, issue_id, starrer_user_ids, starred) |
| |
| |
| class ProjectService(object): |
| """Fake ProjectService object. |
| |
| Provides methods for creating users and projects, which are accessible |
| through parts of the real ProjectService interface. |
| """ |
| |
| def __init__(self): |
| self.test_projects = {} # project_name -> project_pb |
| self.projects_by_id = {} # project_id -> project_pb |
| self.test_star_manager = None |
| self.indexed_projects = {} |
| self.unindexed_projects = set() |
| self.index_counter = 0 |
| self.project_commitments = {} |
| self.ac_exclusion_ids = {} |
| self.no_expand_ids = {} |
| |
| def TestAddProject( |
| self, name, summary='', state=project_pb2.ProjectState.LIVE, |
| owner_ids=None, committer_ids=None, contrib_ids=None, |
| issue_notify_address=None, state_reason='', description=None, |
| project_id=None, process_inbound_email=None, access=None, |
| extra_perms=None): |
| """Add a project to the fake ProjectService object. |
| |
| Args: |
| name: The name of the project. Will replace any existing project under |
| the same name. |
| summary: The summary string of the project. |
| state: Initial state for the project from project_pb2.ProjectState. |
| owner_ids: List of user ids for project owners |
| committer_ids: List of user ids for project committers |
| contrib_ids: List of user ids for project contributors |
| issue_notify_address: email address to send issue change notifications |
| state_reason: string describing the reason the project is in its current |
| state. |
| description: The description string for this project |
| project_id: A unique integer identifier for the created project. |
| process_inbound_email: True to make this project accept inbound email. |
| access: One of the values of enum project_pb2.ProjectAccess. |
| extra_perms: List of ExtraPerms PBs for project members. |
| |
| Returns: |
| A populated project PB. |
| """ |
| proj_pb = project_pb2.Project() |
| proj_pb.project_id = project_id or hash(name) % 100000 |
| proj_pb.project_name = name |
| proj_pb.summary = summary |
| proj_pb.state = state |
| proj_pb.state_reason = state_reason |
| proj_pb.extra_perms = extra_perms or [] |
| if description is not None: |
| proj_pb.description = description |
| |
| self.TestAddProjectMembers(owner_ids, proj_pb, OWNER_ROLE) |
| self.TestAddProjectMembers(committer_ids, proj_pb, COMMITTER_ROLE) |
| self.TestAddProjectMembers(contrib_ids, proj_pb, CONTRIBUTOR_ROLE) |
| |
| if issue_notify_address is not None: |
| proj_pb.issue_notify_address = issue_notify_address |
| if process_inbound_email is not None: |
| proj_pb.process_inbound_email = process_inbound_email |
| if access is not None: |
| proj_pb.access = access |
| |
| self.test_projects[name] = proj_pb |
| self.projects_by_id[proj_pb.project_id] = proj_pb |
| return proj_pb |
| |
| def TestAddProjectMembers(self, user_id_list, proj_pb, role): |
| if user_id_list is not None: |
| for user_id in user_id_list: |
| if role == OWNER_ROLE: |
| proj_pb.owner_ids.append(user_id) |
| elif role == COMMITTER_ROLE: |
| proj_pb.committer_ids.append(user_id) |
| elif role == CONTRIBUTOR_ROLE: |
| proj_pb.contributor_ids.append(user_id) |
| |
| def LookupProjectIDs(self, cnxn, project_names): |
| return { |
| project_name: self.test_projects[project_name].project_id |
| for project_name in project_names |
| if project_name in self.test_projects} |
| |
| def LookupProjectNames(self, cnxn, project_ids): |
| projects_dict = self.GetProjects(cnxn, project_ids) |
| return {p.project_id: p.project_name |
| for p in projects_dict.values()} |
| |
| def CreateProject( |
| self, _cnxn, project_name, owner_ids, committer_ids, |
| contributor_ids, summary, description, |
| state=project_pb2.ProjectState.LIVE, access=None, |
| read_only_reason=None, |
| home_page=None, docs_url=None, source_url=None, |
| logo_gcs_id=None, logo_file_name=None): |
| """Create and store a Project with the given attributes.""" |
| if project_name in self.test_projects: |
| raise exceptions.ProjectAlreadyExists() |
| project = self.TestAddProject( |
| project_name, summary=summary, state=state, |
| owner_ids=owner_ids, committer_ids=committer_ids, |
| contrib_ids=contributor_ids, description=description, |
| access=access) |
| return project.project_id |
| |
| def ExpungeProject(self, _cnxn, project_id): |
| project = self.projects_by_id.get(project_id) |
| if project: |
| self.test_projects.pop(project.project_name, None) |
| |
| def GetProjectsByName(self, _cnxn, project_name_list, use_cache=True): |
| return { |
| pn: self.test_projects[pn] for pn in project_name_list |
| if pn in self.test_projects} |
| |
| def GetProjectByName(self, _cnxn, name, use_cache=True): |
| return self.test_projects.get(name) |
| |
| def GetProjectList(self, cnxn, project_id_list, use_cache=True): |
| project_dict = self.GetProjects(cnxn, project_id_list, use_cache=use_cache) |
| return [project_dict[pid] for pid in project_id_list |
| if pid in project_dict] |
| |
| def GetVisibleProjects( |
| self, _cnxn, logged_in_user, effective_ids, domain=None, use_cache=True): |
| project_ids = sorted(self.projects_by_id.keys()) |
| visible_project_ids = [] |
| for pid in project_ids: |
| can_view = permissions.UserCanViewProject( |
| logged_in_user, effective_ids, self.projects_by_id[pid]) |
| different_domain = framework_helpers.GetNeededDomain( |
| self.projects_by_id[pid].project_name, domain) |
| if can_view and not different_domain: |
| visible_project_ids.append(pid) |
| |
| return visible_project_ids |
| |
| def GetProjects(self, _cnxn, project_ids, use_cache=True): |
| result = {} |
| for project_id in project_ids: |
| project = self.projects_by_id.get(project_id) |
| if project: |
| result[project_id] = project |
| else: |
| raise exceptions.NoSuchProjectException(project_id) |
| return result |
| |
| def GetAllProjects(self, _cnxn, use_cache=True): |
| result = {} |
| for project_id in self.projects_by_id: |
| project = self.projects_by_id.get(project_id) |
| result[project_id] = project |
| return result |
| |
| |
| def GetProject(self, cnxn, project_id, use_cache=True): |
| """Load the specified project from the database.""" |
| project_id_dict = self.GetProjects(cnxn, [project_id], use_cache=use_cache) |
| if project_id not in project_id_dict: |
| raise exceptions.NoSuchProjectException() |
| return project_id_dict[project_id] |
| |
| def GetProjectCommitments(self, _cnxn, project_id): |
| if project_id in self.project_commitments: |
| return self.project_commitments[project_id] |
| |
| project_commitments = project_pb2.ProjectCommitments() |
| project_commitments.project_id = project_id |
| return project_commitments |
| |
| def TestStoreProjectCommitments(self, project_commitments): |
| key = project_commitments.project_id |
| self.project_commitments[key] = project_commitments |
| |
| def GetProjectAutocompleteExclusion(self, cnxn, project_id): |
| return (self.ac_exclusion_ids.get(project_id, []), |
| self.no_expand_ids.get(project_id, [])) |
| |
| def UpdateProject( |
| self, |
| _cnxn, |
| project_id, |
| summary=None, |
| description=None, |
| state=None, |
| state_reason=None, |
| access=None, |
| issue_notify_address=None, |
| attachment_bytes_used=None, |
| attachment_quota=None, |
| moved_to=None, |
| process_inbound_email=None, |
| only_owners_remove_restrictions=None, |
| read_only_reason=None, |
| cached_content_timestamp=None, |
| only_owners_see_contributors=None, |
| delete_time=None, |
| recent_activity=None, |
| revision_url_format=None, |
| home_page=None, |
| docs_url=None, |
| source_url=None, |
| logo_gcs_id=None, |
| logo_file_name=None, |
| issue_notify_always_detailed=None, |
| commit=True): |
| project = self.projects_by_id.get(project_id) |
| if not project: |
| raise exceptions.NoSuchProjectException( |
| 'Project "%s" not found!' % project_id) |
| |
| # TODO(jrobbins): implement all passed arguments - probably as a utility |
| # method shared with the real persistence implementation. |
| if read_only_reason is not None: |
| project.read_only_reason = read_only_reason |
| if attachment_bytes_used is not None: |
| project.attachment_bytes_used = attachment_bytes_used |
| |
| def UpdateProjectRoles( |
| self, _cnxn, project_id, owner_ids, committer_ids, |
| contributor_ids, now=None): |
| project = self.projects_by_id.get(project_id) |
| if not project: |
| raise exceptions.NoSuchProjectException( |
| 'Project "%s" not found!' % project_id) |
| |
| project.owner_ids = owner_ids |
| project.committer_ids = committer_ids |
| project.contributor_ids = contributor_ids |
| |
| def MarkProjectDeletable( |
| self, _cnxn, project_id, _config_service): |
| project = self.projects_by_id[project_id] |
| project.project_name = 'DELETABLE_%d' % project_id |
| project.state = project_pb2.ProjectState.DELETABLE |
| |
| def UpdateRecentActivity(self, _cnxn, _project_id, now=None): |
| pass |
| |
| def GetUserRolesInAllProjects(self, _cnxn, effective_ids): |
| owned_project_ids = set() |
| membered_project_ids = set() |
| contrib_project_ids = set() |
| |
| for project in self.projects_by_id.values(): |
| if not effective_ids.isdisjoint(project.owner_ids): |
| owned_project_ids.add(project.project_id) |
| elif not effective_ids.isdisjoint(project.committer_ids): |
| membered_project_ids.add(project.project_id) |
| elif not effective_ids.isdisjoint(project.contributor_ids): |
| contrib_project_ids.add(project.project_id) |
| |
| return owned_project_ids, membered_project_ids, contrib_project_ids |
| |
| def GetProjectMemberships(self, _cnxn, effective_ids, use_cache=True): |
| # type: MonorailConnection, Collection[int], bool -> |
| # Mapping[int, Collection[int]] |
| projects_by_user_id = collections.defaultdict(set) |
| |
| for project in self.projects_by_id.values(): |
| member_ids = set( |
| itertools.chain( |
| project.owner_ids, project.committer_ids, |
| project.contributor_ids)) |
| for user_id in effective_ids: |
| if user_id in member_ids: |
| projects_by_user_id[user_id].add(project.project_id) |
| return projects_by_user_id |
| |
| def ExpungeUsersInProjects(self, cnxn, user_ids, limit=None): |
| for project in self.projects_by_id.values(): |
| project.owner_ids = [owner_id for owner_id in project.owner_ids |
| if owner_id not in user_ids] |
| project.committer_ids = [com_id for com_id in project.committer_ids |
| if com_id not in user_ids] |
| project.contributor_ids = [con_id for con_id in project.contributor_ids |
| if con_id not in user_ids] |
| |
| |
| class ConfigService(object): |
| """Fake version of ConfigService that just works in-RAM.""" |
| |
| def __init__(self, user_id=None): |
| self.project_configs = {} |
| self.next_field_id = 123 |
| self.next_component_id = 345 |
| self.next_template_id = 23 |
| self.expunged_configs = [] |
| self.expunged_users_in_configs = [] |
| self.component_ids_to_templates = {} |
| self.label_to_id = {} |
| self.id_to_label = {} |
| self.strict = False # Set true to raise more exceptions like real class. |
| |
| def TestAddLabelsDict(self, label_to_id): |
| self.label_to_id = label_to_id |
| self.id_to_label = { |
| label_id: label |
| for label, label_id in list(self.label_to_id.items())} |
| |
| def TestAddFieldDef(self, fd): |
| self.project_configs[fd.project_id].field_defs.append(fd) |
| |
| def TestAddApprovalDef(self, ad, project_id): |
| self.project_configs[project_id].approval_defs.append(ad) |
| |
| def ExpungeConfig(self, _cnxn, project_id): |
| self.expunged_configs.append(project_id) |
| |
| def ExpungeUsersInConfigs(self, _cnxn, user_ids, limit=None): |
| self.expunged_users_in_configs.extend(user_ids) |
| |
| def GetLabelDefRows(self, cnxn, project_id, use_cache=True): |
| """This always returns empty results. Mock it to test other cases.""" |
| return [] |
| |
| def GetLabelDefRowsAnyProject(self, cnxn, where=None): |
| """This always returns empty results. Mock it to test other cases.""" |
| return [] |
| |
| def LookupLabel(self, cnxn, project_id, label_id): |
| if label_id in self.id_to_label: |
| return self.id_to_label[label_id] |
| if label_id == 999: |
| return None |
| return 'label_%d_%d' % (project_id, label_id) |
| |
| def LookupLabelID( |
| self, cnxn, project_id, label, autocreate=True, case_sensitive=False): |
| if label in self.label_to_id: |
| return self.label_to_id[label] |
| # TODO: The condition here is specifically added to return 'None' and |
| # allow testing for label freezing. This can be removed after refactoring |
| # other dependent tests to not fail for returning 'None' instead of '1' |
| # when label is not found in 'label_to_id' dict. |
| if label == 'freeze_new_label': |
| return None |
| return 1 |
| |
| def LookupLabelIDs(self, cnxn, project_id, labels, autocreate=False): |
| ids = [] |
| next_label_id = 0 |
| if self.id_to_label.keys(): |
| existing_ids = sorted(self.id_to_label.keys()) |
| next_label_id = existing_ids[-1] + 1 |
| for label in labels: |
| if self.label_to_id.get(label) is not None: |
| ids.append(self.label_to_id[label]) |
| elif autocreate: |
| self.label_to_id[label] = next_label_id |
| self.id_to_label[next_label_id] = label |
| ids.append(next_label_id) |
| next_label_id += 1 |
| return ids |
| |
| def LookupIDsOfLabelsMatching(self, cnxn, project_id, regex): |
| return [1, 2, 3] |
| |
| def LookupStatus(self, cnxn, project_id, status_id): |
| return 'status_%d_%d' % (project_id, status_id) |
| |
| def LookupStatusID(self, cnxn, project_id, status, autocreate=True): |
| if status: |
| return 1 |
| else: |
| return 0 |
| |
| def LookupStatusIDs(self, cnxn, project_id, statuses): |
| return [idx for idx, _status in enumerate(statuses)] |
| |
| def LookupClosedStatusIDs(self, cnxn, project_id): |
| return [7, 8, 9] |
| |
| def StoreConfig(self, _cnxn, config): |
| self.project_configs[config.project_id] = config |
| |
| def GetProjectConfig(self, _cnxn, project_id, use_cache=True): |
| if project_id in self.project_configs: |
| return self.project_configs[project_id] |
| elif self.strict: |
| raise exceptions.NoSuchProjectException() |
| else: |
| return tracker_bizobj.MakeDefaultProjectIssueConfig(project_id) |
| |
| def GetProjectConfigs(self, _cnxn, project_ids, use_cache=True): |
| config_dict = {} |
| for project_id in project_ids: |
| if project_id in self.project_configs: |
| config_dict[project_id] = self.project_configs[project_id] |
| elif not self.strict: |
| config_dict[project_id] = tracker_bizobj.MakeDefaultProjectIssueConfig( |
| project_id) |
| return config_dict |
| |
| def UpdateConfig( |
| self, cnxn, project, well_known_statuses=None, |
| statuses_offer_merge=None, well_known_labels=None, |
| excl_label_prefixes=None, default_template_for_developers=None, |
| default_template_for_users=None, list_prefs=None, restrict_to_known=None, |
| approval_defs=None): |
| project_id = project.project_id |
| project_config = self.GetProjectConfig(cnxn, project_id, use_cache=False) |
| |
| if well_known_statuses is not None: |
| tracker_bizobj.SetConfigStatuses(project_config, well_known_statuses) |
| |
| if statuses_offer_merge is not None: |
| project_config.statuses_offer_merge = statuses_offer_merge |
| |
| if well_known_labels is not None: |
| tracker_bizobj.SetConfigLabels(project_config, well_known_labels) |
| |
| if excl_label_prefixes is not None: |
| project_config.exclusive_label_prefixes = excl_label_prefixes |
| |
| if approval_defs is not None: |
| tracker_bizobj.SetConfigApprovals(project_config, approval_defs) |
| |
| if default_template_for_developers is not None: |
| project_config.default_template_for_developers = ( |
| default_template_for_developers) |
| if default_template_for_users is not None: |
| project_config.default_template_for_users = default_template_for_users |
| |
| if list_prefs: |
| default_col_spec, default_sort_spec, x_attr, y_attr, m_d_q = list_prefs |
| project_config.default_col_spec = default_col_spec |
| project_config.default_sort_spec = default_sort_spec |
| project_config.default_x_attr = x_attr |
| project_config.default_y_attr = y_attr |
| project_config.member_default_query = m_d_q |
| |
| if restrict_to_known is not None: |
| project_config.restrict_to_known = restrict_to_known |
| |
| self.StoreConfig(cnxn, project_config) |
| return project_config |
| |
| def CreateFieldDef( |
| self, |
| cnxn, |
| project_id, |
| field_name, |
| field_type_str, |
| applic_type, |
| applic_pred, |
| is_required, |
| is_niche, |
| is_multivalued, |
| min_value, |
| max_value, |
| regex, |
| needs_member, |
| needs_perm, |
| grants_perm, |
| notify_on, |
| date_action_str, |
| docstring, |
| admin_ids, |
| editor_ids, |
| approval_id=None, |
| is_phase_field=False, |
| is_restricted_field=False): |
| config = self.GetProjectConfig(cnxn, project_id) |
| field_type = tracker_pb2.FieldTypes(field_type_str) |
| field_id = self.next_field_id |
| self.next_field_id += 1 |
| fd = tracker_bizobj.MakeFieldDef( |
| field_id, project_id, field_name, field_type, applic_type, applic_pred, |
| is_required, is_niche, is_multivalued, min_value, max_value, regex, |
| needs_member, needs_perm, grants_perm, notify_on, date_action_str, |
| docstring, False, approval_id, is_phase_field, is_restricted_field, |
| admin_ids=admin_ids, editor_ids=editor_ids) |
| config.field_defs.append(fd) |
| self.StoreConfig(cnxn, config) |
| return field_id |
| |
| def LookupFieldID(self, cnxn, project_id, field): |
| config = self.GetProjectConfig(cnxn, project_id) |
| for fd in config.field_defs: |
| if fd.field_name == field: |
| return fd.field_id |
| |
| return None |
| |
| def SoftDeleteFieldDefs(self, cnxn, project_id, field_ids): |
| config = self.GetProjectConfig(cnxn, project_id) |
| for fd in config.field_defs: |
| if fd.field_id in field_ids: |
| fd.is_deleted = True |
| self.StoreConfig(cnxn, config) |
| |
| def UpdateFieldDef( |
| self, |
| cnxn, |
| project_id, |
| field_id, |
| field_name=None, |
| applicable_type=None, |
| applicable_predicate=None, |
| is_required=None, |
| is_niche=None, |
| is_multivalued=None, |
| min_value=None, |
| max_value=None, |
| regex=None, |
| needs_member=None, |
| needs_perm=None, |
| grants_perm=None, |
| notify_on=None, |
| date_action=None, |
| docstring=None, |
| admin_ids=None, |
| editor_ids=None, |
| is_restricted_field=None): |
| config = self.GetProjectConfig(cnxn, project_id) |
| fd = tracker_bizobj.FindFieldDefByID(field_id, config) |
| # pylint: disable=multiple-statements |
| if field_name is not None: fd.field_name = field_name |
| if applicable_type is not None: fd.applicable_type = applicable_type |
| if applicable_predicate is not None: |
| fd.applicable_predicate = applicable_predicate |
| if is_required is not None: fd.is_required = is_required |
| if is_niche is not None: fd.is_niche = is_niche |
| if is_multivalued is not None: fd.is_multivalued = is_multivalued |
| if min_value is not None: fd.min_value = min_value |
| if max_value is not None: fd.max_value = max_value |
| if regex is not None: fd.regex = regex |
| if date_action is not None: |
| fd.date_action = config_svc.DATE_ACTION_ENUM.index(date_action) |
| if docstring is not None: fd.docstring = docstring |
| if admin_ids is not None: fd.admin_ids = admin_ids |
| if editor_ids is not None: |
| fd.editor_ids = editor_ids |
| if is_restricted_field is not None: |
| fd.is_restricted_field = is_restricted_field |
| self.StoreConfig(cnxn, config) |
| |
| def CreateComponentDef( |
| self, cnxn, project_id, path, docstring, deprecated, admin_ids, cc_ids, |
| created, creator_id, label_ids): |
| config = self.GetProjectConfig(cnxn, project_id) |
| cd = tracker_bizobj.MakeComponentDef( |
| self.next_component_id, project_id, path, docstring, deprecated, |
| admin_ids, cc_ids, created, creator_id, label_ids=label_ids) |
| config.component_defs.append(cd) |
| self.next_component_id += 1 |
| self.StoreConfig(cnxn, config) |
| return self.next_component_id - 1 |
| |
| def UpdateComponentDef( |
| self, cnxn, project_id, component_id, path=None, docstring=None, |
| deprecated=None, admin_ids=None, cc_ids=None, created=None, |
| creator_id=None, modified=None, modifier_id=None, label_ids=None): |
| config = self.GetProjectConfig(cnxn, project_id) |
| cd = tracker_bizobj.FindComponentDefByID(component_id, config) |
| if path is not None: |
| assert path |
| cd.path = path |
| # pylint: disable=multiple-statements |
| if docstring is not None: cd.docstring = docstring |
| if deprecated is not None: cd.deprecated = deprecated |
| if admin_ids is not None: cd.admin_ids = admin_ids |
| if cc_ids is not None: cd.cc_ids = cc_ids |
| if created is not None: cd.created = created |
| if creator_id is not None: cd.creator_id = creator_id |
| if modified is not None: cd.modified = modified |
| if modifier_id is not None: cd.modifier_id = modifier_id |
| if label_ids is not None: cd.label_ids = label_ids |
| self.StoreConfig(cnxn, config) |
| |
| def DeleteComponentDef(self, cnxn, project_id, component_id): |
| """Delete the specified component definition.""" |
| config = self.GetProjectConfig(cnxn, project_id) |
| config.component_defs = [ |
| cd for cd in config.component_defs |
| if cd.component_id != component_id] |
| self.StoreConfig(cnxn, config) |
| |
| def InvalidateMemcache(self, issues, key_prefix=''): |
| pass |
| |
| def InvalidateMemcacheForEntireProject(self, project_id): |
| pass |
| |
| |
| class IssueService(object): |
| """Fake version of IssueService that just works in-RAM.""" |
| # pylint: disable=unused-argument |
| |
| def __init__(self, user_id=None): |
| self.user_id = user_id |
| # Dictionary {project_id: issue_pb_dict} |
| # where issue_pb_dict is a dictionary of the form |
| # {local_id: issue_pb} |
| self.issues_by_project = {} |
| self.issues_by_iid = {} |
| # Dictionary {project_id: comment_pb_dict} |
| # where comment_pb_dict is a dictionary of the form |
| # {local_id: comment_pb_list} |
| self.comments_by_project = {} |
| self.comments_by_iid = {} |
| self.comments_by_cid = {} |
| self.attachments_by_id = {} |
| |
| # Set of issue IDs for issues that have been indexed by calling |
| # IndexIssues(). |
| self.indexed_issue_iids = set() |
| |
| # Set of issue IDs for issues that have been moved by calling MoveIssue(). |
| self.moved_back_iids = set() |
| |
| # Dict of issue IDs mapped to other issue IDs to represent moved issues. |
| self.moved_issues = {} |
| |
| # Test-only indication that the indexer would have been called |
| # by the real DITPersist. |
| self.indexer_called = False |
| |
| # Test-only sequence of updated and enqueued. |
| self.updated_issues = [] |
| self.enqueued_issues = [] # issue_ids |
| |
| # Test-only sequence of expunged issues and projects. |
| self.expunged_issues = [] |
| self.expunged_former_locations = [] |
| self.expunged_local_ids = [] |
| self.expunged_users_in_issues = [] |
| |
| # Test-only indicators that methods were called. |
| self.get_all_issues_in_project_called = False |
| self.update_issues_called = False |
| self.enqueue_issues_called = False |
| self.get_issue_acitivity_called = False |
| |
| # The next id to return if it is > 0. |
| self.next_id = -1 |
| |
| def UpdateIssue( |
| self, |
| cnxn, |
| issue, |
| update_cols=None, |
| just_derived=False, |
| commit=True, |
| invalidate=True): |
| self.UpdateIssues( |
| cnxn, [issue], update_cols, just_derived, commit, invalidate) |
| |
| def UpdateIssues( |
| self, cnxn, issues, update_cols=None, just_derived=False, |
| commit=True, invalidate=True): |
| self.update_issues_called = True |
| assert all(issue.assume_stale == False for issue in issues) |
| self.updated_issues.extend(issues) |
| |
| def GetIssueActivity( |
| self, cnxn, num=50, before=None, after=None, |
| project_ids=None, user_ids=None, ascending=False): |
| self.get_issue_acitivity_called = True |
| comments_dict = self.comments_by_cid |
| comments = [] |
| for value in comments_dict.values(): |
| if project_ids is not None: |
| if value.issue_id > 0 and value.issue_id in self.issues_by_iid: |
| issue = self.issues_by_iid[value.issue_id] |
| if issue.project_id in project_ids: |
| comments.append(value) |
| elif user_ids is not None: |
| if value.user_id in user_ids: |
| comments.append(value) |
| else: |
| comments.append(value) |
| return comments |
| |
| def EnqueueIssuesForIndexing(self, _cnxn, issue_ids, commit=True): |
| self.enqueue_issues_called = True |
| for i in issue_ids: |
| if i not in self.enqueued_issues: |
| self.enqueued_issues.extend(issues) |
| |
| def ExpungeIssues(self, _cnxn, issue_ids): |
| self.expunged_issues.extend(issue_ids) |
| |
| def ExpungeFormerLocations(self, _cnxn, project_id): |
| self.expunged_former_locations.append(project_id) |
| |
| def ExpungeLocalIDCounters(self, _cnxn, project_id): |
| self.expunged_local_ids.append(project_id) |
| |
| def TestAddIssue(self, issue, importer_id=None): |
| project_id = issue.project_id |
| self.issues_by_project.setdefault(project_id, {}) |
| self.issues_by_project[project_id][issue.local_id] = issue |
| self.issues_by_iid[issue.issue_id] = issue |
| if issue.issue_id not in self.enqueued_issues: |
| self.enqueued_issues.append(issue.issue_id) |
| self.enqueue_issues_called = True |
| |
| # Adding a new issue should add the first comment to the issue |
| comment = tracker_pb2.IssueComment() |
| comment.project_id = issue.project_id |
| comment.issue_id = issue.issue_id |
| comment.content = issue.summary |
| comment.timestamp = issue.opened_timestamp |
| comment.is_description = True |
| if issue.reporter_id: |
| comment.user_id = issue.reporter_id |
| if importer_id: |
| comment.importer_id = importer_id |
| comment.sequence = 0 |
| self.TestAddComment(comment, issue.local_id) |
| |
| def TestAddMovedIssueRef(self, source_project_id, source_local_id, |
| target_project_id, target_local_id): |
| self.moved_issues[(source_project_id, source_local_id)] = ( |
| target_project_id, target_local_id) |
| |
| def TestAddComment(self, comment, local_id): |
| pid = comment.project_id |
| if not comment.id: |
| comment.id = len(self.comments_by_cid) |
| |
| self.comments_by_project.setdefault(pid, {}) |
| self.comments_by_project[pid].setdefault(local_id, []).append(comment) |
| self.comments_by_iid.setdefault(comment.issue_id, []).append(comment) |
| self.comments_by_cid[comment.id] = comment |
| |
| def TestAddAttachment(self, attachment, comment_id, issue_id): |
| if not attachment.attachment_id: |
| attachment.attachment_id = len(self.attachments_by_id) |
| |
| aid = attachment.attachment_id |
| self.attachments_by_id[aid] = attachment, comment_id, issue_id |
| comment = self.comments_by_cid[comment_id] |
| if attachment not in comment.attachments: |
| comment.attachments.extend([attachment]) |
| |
| def SoftDeleteAttachment( |
| self, _cnxn, _issue, comment, attach_id, _user_service, delete=True, |
| index_now=False): |
| attachment = None |
| for attach in comment.attachments: |
| if attach.attachment_id == attach_id: |
| attachment = attach |
| if not attachment: |
| return |
| attachment.deleted = delete |
| |
| def GetAttachmentAndContext(self, _cnxn, attachment_id): |
| if attachment_id in self.attachments_by_id: |
| attach, comment_id, issue_id = self.attachments_by_id[attachment_id] |
| if not attach.deleted: |
| return attach, comment_id, issue_id |
| |
| raise exceptions.NoSuchAttachmentException() |
| |
| def GetComments( |
| self, _cnxn, where=None, order_by=None, content_only=False, **kwargs): |
| # This is a very limited subset of what the real GetComments() can do. |
| cid = kwargs.get('id') |
| |
| comment = self.comments_by_cid.get(cid) |
| if comment: |
| return [comment] |
| else: |
| return [] |
| |
| def GetComment(self, cnxn, comment_id): |
| """Get the requested comment, or raise an exception.""" |
| comments = self.GetComments(cnxn, id=comment_id) |
| if len(comments) == 1: |
| return comments[0] |
| |
| raise exceptions.NoSuchCommentException() |
| |
| def ResolveIssueRefs(self, cnxn, ref_projects, default_project_name, refs): |
| result = [] |
| misses = [] |
| for project_name, local_id in refs: |
| project = ref_projects.get(project_name or default_project_name) |
| if not project or project.state == project_pb2.ProjectState.DELETABLE: |
| continue # ignore any refs to issues in deleted projects |
| try: |
| issue = self.GetIssueByLocalID(cnxn, project.project_id, local_id) |
| result.append(issue.issue_id) |
| except exceptions.NoSuchIssueException: |
| misses.append((project.project_id, local_id)) |
| |
| return result, misses |
| |
| def LookupIssueRefs(self, cnxn, issue_ids): |
| issue_dict, _misses = self.GetIssuesDict(cnxn, issue_ids) |
| return { |
| issue_id: (issue.project_name, issue.local_id) |
| for issue_id, issue in issue_dict.items()} |
| |
| def GetAllIssuesInProject( |
| self, _cnxn, project_id, min_local_id=None, use_cache=True): |
| self.get_all_issues_in_project_called = True |
| if project_id in self.issues_by_project: |
| return list(self.issues_by_project[project_id].values()) |
| else: |
| return [] |
| |
| def GetIssuesByLocalIDs( |
| self, _cnxn, project_id, local_id_list, use_cache=True, shard_id=None): |
| results = [] |
| for local_id in local_id_list: |
| if (project_id in self.issues_by_project |
| and local_id in self.issues_by_project[project_id]): |
| results.append(self.issues_by_project[project_id][local_id]) |
| |
| return results |
| |
| def GetIssueByLocalID(self, _cnxn, project_id, local_id, use_cache=True): |
| try: |
| return self.issues_by_project[project_id][local_id] |
| except KeyError: |
| raise exceptions.NoSuchIssueException() |
| |
| def GetAnyOnHandIssue(self, issue_ids, start=None, end=None): |
| return None # Treat them all like misses. |
| |
| def GetIssue(self, cnxn, issue_id, use_cache=True): |
| issues = self.GetIssues(cnxn, [issue_id], use_cache=use_cache) |
| try: |
| return issues[0] |
| except IndexError: |
| raise exceptions.NoSuchIssueException() |
| |
| def GetCurrentLocationOfMovedIssue(self, cnxn, project_id, local_id): |
| key = (project_id, local_id) |
| if key in self.moved_issues: |
| ref = self.moved_issues[key] |
| return ref[0], ref[1] |
| return None, None |
| |
| def GetPreviousLocations(self, cnxn, issue): |
| return [] |
| |
| def GetCommentsByUser(self, cnxn, user_id): |
| """Get all comments created by a user""" |
| comments = [] |
| for cid in self.comments_by_cid: |
| comment = self.comments_by_cid[cid] |
| if comment.user_id == user_id and not comment.is_description: |
| comments.append(comment) |
| return comments |
| |
| def GetCommentsByID(self, cnxn, comment_ids, _sequences, use_cache=True, |
| shard_id=None): |
| """Return all IssueComment PBs by comment ids.""" |
| comments = [self.comments_by_cid[cid] for cid in comment_ids] |
| return comments |
| |
| def GetIssueIDsReportedByUser(self, cnxn, user_id): |
| """Get all issues created by a user""" |
| ids = [] |
| for iid in self.issues_by_iid: |
| issue = self.issues_by_iid[iid] |
| if issue.reporter_id == user_id: |
| ids.append(iid) |
| return ids |
| |
| def LookupIssueIDs(self, _cnxn, project_local_id_pairs): |
| hits = [] |
| misses = [] |
| for (project_id, local_id) in project_local_id_pairs: |
| try: |
| issue = self.issues_by_project[project_id][local_id] |
| hits.append(issue.issue_id) |
| except KeyError: |
| misses.append((project_id, local_id)) |
| |
| return hits, misses |
| |
| def LookupIssueIDsFollowMoves(self, _cnxn, project_local_id_pairs): |
| hits = [] |
| misses = [] |
| for pair in project_local_id_pairs: |
| project_id, local_id = self.moved_issues.get(pair, pair) |
| try: |
| issue = self.issues_by_project[project_id][local_id] |
| hits.append(issue.issue_id) |
| except KeyError: |
| misses.append((project_id, local_id)) |
| |
| return hits, misses |
| |
| def LookupIssueID(self, _cnxn, project_id, local_id): |
| try: |
| issue = self.issues_by_project[project_id][local_id] |
| except KeyError: |
| raise exceptions.NoSuchIssueException() |
| return issue.issue_id |
| |
| def GetCommentsForIssue(self, _cnxn, issue_id): |
| comments = self.comments_by_iid.get(issue_id, []) |
| for idx, c in enumerate(comments): |
| c.sequence = idx |
| |
| return comments |
| |
| def InsertIssue(self, cnxn, issue): |
| issue.issue_id = issue.project_id * 1000000 + issue.local_id |
| self.issues_by_project.setdefault(issue.project_id, {}) |
| self.issues_by_project[issue.project_id][issue.local_id] = issue |
| self.issues_by_iid[issue.issue_id] = issue |
| return issue.issue_id |
| |
| def CreateIssue( |
| self, |
| cnxn, |
| services, |
| issue, |
| marked_description, |
| attachments=None, |
| index_now=False, |
| importer_id=None): |
| project_id = issue.project_id |
| |
| issue.local_id = self.AllocateNextLocalID(cnxn, project_id) |
| issue.issue_id = project_id * 1000000 + issue.local_id |
| |
| self.TestAddIssue(issue, importer_id=importer_id) |
| comment = self.comments_by_iid[issue.issue_id][0] |
| comment.content = marked_description |
| return issue, comment |
| |
| def GetIssueApproval(self, cnxn, issue_id, approval_id, use_cache=True): |
| issue = self.GetIssue(cnxn, issue_id, use_cache=use_cache) |
| approval = tracker_bizobj.FindApprovalValueByID( |
| approval_id, issue.approval_values) |
| if approval: |
| return issue, approval |
| raise exceptions.NoSuchIssueApprovalException() |
| |
| def UpdateIssueApprovalStatus( |
| self, cnxn, issue_id, approval_id, status, setter_id, set_on, |
| commit=True): |
| issue = self.GetIssue(cnxn, issue_id) |
| for av in issue.approval_values: |
| if av.approval_id == approval_id: |
| av.status = status |
| av.setter_id = setter_id |
| av.set_on = set_on |
| return |
| return |
| |
| def UpdateIssueApprovalApprovers( |
| self, cnxn, issue_id, approval_id, approver_ids, commit=True): |
| issue = self.GetIssue(cnxn, issue_id) |
| for av in issue.approval_values: |
| if av.approval_id == approval_id: |
| av.approver_ids = approver_ids |
| return |
| return |
| |
| def UpdateIssueStructure( |
| self, cnxn, config, issue, template, reporter_id, comment_content, |
| commit=True, invalidate=True): |
| approval_defs_by_id = {ad.approval_id: ad for ad in config.approval_defs} |
| issue_avs_by_id = {av.approval_id: av for av in issue.approval_values} |
| |
| new_issue_approvals = [] |
| |
| for template_av in template.approval_values: |
| existing_issue_av = issue_avs_by_id.get(template_av.approval_id) |
| # Keep approval values as-if fi it exists in issue and template |
| if existing_issue_av: |
| existing_issue_av.phase_id = template_av.phase_id |
| new_issue_approvals.append(existing_issue_av) |
| else: |
| new_issue_approvals.append(template_av) |
| |
| # Update all approval surveys so latest ApprovalDef survey changes |
| # appear in the converted issue's approval values. |
| ad = approval_defs_by_id.get(template_av.approval_id) |
| if ad: |
| self.CreateIssueComment( |
| cnxn, issue, reporter_id, ad.survey, |
| is_description=True, approval_id=ad.approval_id, commit=False) |
| else: |
| logging.info('ApprovalDef not found for approval %r', template_av) |
| |
| template_phase_by_name = { |
| phase.name.lower(): phase for phase in template.phases} |
| issue_phase_by_id = {phase.phase_id: phase for phase in issue.phases} |
| updated_fvs = [] |
| # Trim issue FieldValues or update FieldValue phase_ids |
| for fv in issue.field_values: |
| # If a fv's phase has the same name as a template's phase, update |
| # the fv's phase_id to that of the template phase's. Otherwise, |
| # remove the fv. |
| if fv.phase_id: |
| issue_phase = issue_phase_by_id.get(fv.phase_id) |
| if issue_phase and issue_phase.name: |
| template_phase = template_phase_by_name.get(issue_phase.name.lower()) |
| if template_phase: |
| fv.phase_id = template_phase.phase_id |
| updated_fvs.append(fv) |
| # keep all fvs that do not belong to phases. |
| else: |
| updated_fvs.append(fv) |
| |
| fd_names_by_id = {fd.field_id: fd.field_name for fd in config.field_defs} |
| amendment = tracker_bizobj.MakeApprovalStructureAmendment( |
| [fd_names_by_id.get(av.approval_id) for av in new_issue_approvals], |
| [fd_names_by_id.get(av.approval_id) for av in issue.approval_values]) |
| |
| issue.approval_values = new_issue_approvals |
| issue.phases = template.phases |
| issue.field_values = updated_fvs |
| |
| return self.CreateIssueComment( |
| cnxn, issue, reporter_id, comment_content, |
| amendments=[amendment], commit=False) |
| |
| def SetUsedLocalID(self, cnxn, project_id): |
| self.next_id = self.GetHighestLocalID(cnxn, project_id) + 1 |
| |
| def AllocateNextLocalID(self, cnxn, project_id): |
| return self.GetHighestLocalID(cnxn, project_id) + 1 |
| |
| def GetHighestLocalID(self, _cnxn, project_id): |
| if self.next_id > 0: |
| return self.next_id - 1 |
| else: |
| issue_dict = self.issues_by_project.get(project_id, {}) |
| highest = max([0] + [issue.local_id for issue in issue_dict.values()]) |
| return highest |
| |
| def _MakeIssueComment( |
| self, project_id, user_id, content, inbound_message=None, |
| amendments=None, attachments=None, kept_attachments=None, timestamp=None, |
| is_spam=False, is_description=False, approval_id=None, importer_id=None): |
| comment = tracker_pb2.IssueComment() |
| comment.project_id = project_id |
| comment.user_id = user_id |
| comment.content = content or '' |
| comment.is_spam = is_spam |
| comment.is_description = is_description |
| if not timestamp: |
| timestamp = int(time.time()) |
| comment.timestamp = int(timestamp) |
| if inbound_message: |
| comment.inbound_message = inbound_message |
| if amendments: |
| comment.amendments.extend(amendments) |
| if approval_id: |
| comment.approval_id = approval_id |
| if importer_id: |
| comment.importer_id = importer_id |
| return comment |
| |
| def CopyIssues(self, cnxn, dest_project, issues, user_service, copier_id): |
| created_issues = [] |
| for target_issue in issues: |
| new_issue = tracker_pb2.Issue() |
| new_issue.project_id = dest_project.project_id |
| new_issue.project_name = dest_project.project_name |
| new_issue.summary = target_issue.summary |
| new_issue.labels.extend(target_issue.labels) |
| new_issue.field_values.extend(target_issue.field_values) |
| new_issue.reporter_id = copier_id |
| |
| timestamp = int(time.time()) |
| new_issue.opened_timestamp = timestamp |
| new_issue.modified_timestamp = timestamp |
| new_issue.migration_modified_timestamp = timestamp |
| |
| target_comments = self.GetCommentsForIssue(cnxn, target_issue.issue_id) |
| initial_summary_comment = target_comments[0] |
| |
| # Note that blocking and merge_into are not copied. |
| new_issue.blocked_on_iids = target_issue.blocked_on_iids |
| new_issue.blocked_on_ranks = target_issue.blocked_on_ranks |
| |
| # Create the same summary comment as the target issue. |
| comment = self._MakeIssueComment( |
| dest_project.project_id, copier_id, initial_summary_comment.content, |
| is_description=True) |
| |
| new_issue.local_id = self.AllocateNextLocalID( |
| cnxn, dest_project.project_id) |
| issue_id = self.InsertIssue(cnxn, new_issue) |
| comment.issue_id = issue_id |
| self.InsertComment(cnxn, comment) |
| created_issues.append(new_issue) |
| |
| return created_issues |
| |
| def MoveIssues(self, cnxn, dest_project, issues, user_service): |
| move_to = dest_project.project_id |
| self.issues_by_project.setdefault(move_to, {}) |
| moved_back_iids = set() |
| for issue in issues: |
| if issue.issue_id in self.moved_back_iids: |
| moved_back_iids.add(issue.issue_id) |
| self.moved_back_iids.add(issue.issue_id) |
| project_id = issue.project_id |
| self.issues_by_project[project_id].pop(issue.local_id) |
| issue.local_id = self.AllocateNextLocalID(cnxn, move_to) |
| self.issues_by_project[move_to][issue.local_id] = issue |
| issue.project_id = move_to |
| issue.project_name = dest_project.project_name |
| return moved_back_iids |
| |
| def GetCommentsForIssues(self, _cnxn, issue_ids, content_only=False): |
| comments_dict = {} |
| for issue_id in issue_ids: |
| comments_dict[issue_id] = self.comments_by_iid[issue_id] |
| |
| return comments_dict |
| |
| def InsertComment(self, cnxn, comment, commit=True): |
| issue = self.GetIssue(cnxn, comment.issue_id) |
| self.TestAddComment(comment, issue.local_id) |
| |
| # pylint: disable=unused-argument |
| def DeltaUpdateIssue( |
| self, cnxn, services, reporter_id, project_id, |
| config, issue, delta, index_now=False, comment=None, attachments=None, |
| iids_to_invalidate=None, rules=None, predicate_asts=None, |
| is_description=False, timestamp=None, kept_attachments=None, |
| importer_id=None, inbound_message=None): |
| # Return a bogus amendments list if any of the fields changed |
| amendments, _ = tracker_bizobj.ApplyIssueDelta( |
| cnxn, self, issue, delta, config) |
| |
| if not amendments and (not comment or not comment.strip()): |
| return [], None |
| |
| comment_pb = self.CreateIssueComment( |
| cnxn, issue, reporter_id, comment, attachments=attachments, |
| amendments=amendments, is_description=is_description, |
| kept_attachments=kept_attachments, importer_id=importer_id, |
| inbound_message=inbound_message) |
| |
| self.indexer_called = index_now |
| return amendments, comment_pb |
| |
| def InvalidateIIDs(self, cnxn, iids_to_invalidate): |
| pass |
| |
| # pylint: disable=unused-argument |
| def CreateIssueComment( |
| self, _cnxn, issue, user_id, content, |
| inbound_message=None, amendments=None, attachments=None, |
| kept_attachments=None, timestamp=None, is_spam=False, |
| is_description=False, approval_id=None, commit=True, |
| importer_id=None): |
| # Add a comment to an issue |
| comment = tracker_pb2.IssueComment() |
| comment.id = len(self.comments_by_cid) |
| comment.project_id = issue.project_id |
| comment.issue_id = issue.issue_id |
| comment.content = content |
| comment.user_id = user_id |
| if timestamp is not None: |
| comment.timestamp = timestamp |
| else: |
| comment.timestamp = 1234567890 |
| if amendments: |
| comment.amendments.extend(amendments) |
| if inbound_message: |
| comment.inbound_message = inbound_message |
| comment.is_spam = is_spam |
| comment.is_description = is_description |
| if approval_id: |
| comment.approval_id = approval_id |
| |
| pid = issue.project_id |
| self.comments_by_project.setdefault(pid, {}) |
| self.comments_by_project[pid].setdefault(issue.local_id, []).append(comment) |
| self.comments_by_iid.setdefault(issue.issue_id, []).append(comment) |
| self.comments_by_cid[comment.id] = comment |
| |
| if attachments: |
| for filename, filecontent, mimetype in attachments: |
| aid = len(self.attachments_by_id) |
| attach = tracker_pb2.Attachment( |
| attachment_id=aid, |
| filename=filename, |
| filesize=len(filecontent), |
| mimetype=mimetype, |
| gcs_object_id='gcs_object_id(%s)' % filename) |
| comment.attachments.append(attach) |
| self.attachments_by_id[aid] = attach, pid, comment.id |
| |
| if kept_attachments: |
| comment.attachments.extend([ |
| self.attachments_by_id[aid][0] |
| for aid in kept_attachments]) |
| |
| return comment |
| |
| def GetOpenAndClosedIssues(self, _cnxn, issue_ids): |
| open_issues = [] |
| closed_issues = [] |
| for issue_id in issue_ids: |
| try: |
| issue = self.issues_by_iid[issue_id] |
| if issue.status == 'Fixed': |
| closed_issues.append(issue) |
| else: |
| open_issues.append(issue) |
| except KeyError: |
| continue |
| |
| return open_issues, closed_issues |
| |
| def GetIssuesDict( |
| self, _cnxn, issue_ids, use_cache=True, shard_id=None): |
| missing_ids = [iid for iid in issue_ids if iid not in self.issues_by_iid] |
| issues_by_id = {} |
| for iid in issue_ids: |
| if iid in self.issues_by_iid: |
| issue = self.issues_by_iid[iid] |
| if not use_cache: |
| issue.assume_stale = False |
| issues_by_id[iid] = issue |
| |
| return issues_by_id, missing_ids |
| |
| def GetIssues(self, cnxn, issue_ids, use_cache=True, shard_id=None): |
| issues_by_iid, _misses = self.GetIssuesDict( |
| cnxn, issue_ids, use_cache=use_cache, shard_id=shard_id) |
| results = [ |
| issues_by_iid[issue_id] |
| for issue_id in issue_ids |
| if issue_id in issues_by_iid |
| ] |
| |
| return results |
| |
| def SoftDeleteIssue( |
| self, _cnxn, project_id, local_id, deleted, user_service): |
| issue = self.issues_by_project[project_id][local_id] |
| issue.deleted = deleted |
| |
| def SoftDeleteComment( |
| self, cnxn, issue, comment, deleted_by_user_id, user_service, |
| delete=True, reindex=False, is_spam=False): |
| pid = comment.project_id |
| # Find the original comment by the sequence number. |
| c = None |
| by_iid_idx = -1 |
| for by_iid_idx, c in enumerate(self.comments_by_iid[issue.issue_id]): |
| if c.sequence == comment.sequence: |
| break |
| comment = c |
| by_project_idx = ( |
| self.comments_by_project[pid][issue.local_id].index(comment)) |
| comment.is_spam = is_spam |
| if delete: |
| comment.deleted_by = deleted_by_user_id |
| else: |
| comment.reset('deleted_by') |
| self.comments_by_project[pid][issue.local_id][by_project_idx] = comment |
| self.comments_by_iid[issue.issue_id][by_iid_idx] = comment |
| self.comments_by_cid[comment.id] = comment |
| |
| def DeleteComponentReferences(self, _cnxn, component_id): |
| for _, issue in self.issues_by_iid.items(): |
| issue.component_ids = [ |
| cid for cid in issue.component_ids if cid != component_id] |
| |
| def RunIssueQuery( |
| self, cnxn, left_joins, where, order_by, shard_id=None, limit=None): |
| """This always returns empty results. Mock it to test other cases.""" |
| return [], False |
| |
| def GetIIDsByLabelIDs(self, cnxn, label_ids, project_id, shard_id): |
| """This always returns empty results. Mock it to test other cases.""" |
| return [] |
| |
| def GetIIDsByParticipant(self, cnxn, user_ids, project_ids, shard_id): |
| """This always returns empty results. Mock it to test other cases.""" |
| return [] |
| |
| def SortBlockedOn(self, cnxn, issue, blocked_on_iids): |
| return blocked_on_iids, [0] * len(blocked_on_iids) |
| |
| def ApplyIssueRerank( |
| self, cnxn, parent_id, relations_to_change, commit=True, invalidate=True): |
| issue = self.GetIssue(cnxn, parent_id) |
| relations_dict = dict( |
| list(zip(issue.blocked_on_iids, issue.blocked_on_ranks))) |
| relations_dict.update(relations_to_change) |
| issue.blocked_on_ranks = sorted(issue.blocked_on_ranks, reverse=True) |
| issue.blocked_on_iids = sorted( |
| issue.blocked_on_iids, key=relations_dict.get, reverse=True) |
| |
| def SplitRanks(self, cnxn, parent_id, target_id, open_ids, split_above=False): |
| pass |
| |
| def ExpungeUsersInIssues(self, cnxn, user_ids_by_email, limit=None): |
| user_ids = list(user_ids_by_email.values()) |
| self.expunged_users_in_issues.extend(user_ids) |
| return [] |
| |
| |
| class TemplateService(object): |
| """Fake version of TemplateService that just works in-RAM.""" |
| |
| def __init__(self): |
| self.templates_by_id = {} # template_id: template_pb |
| self.templates_by_project_id = {} # project_id: [template_id] |
| |
| def TestAddIssueTemplateDef( |
| self, template_id, project_id, name, content="", summary="", |
| summary_must_be_edited=False, status='New', members_only=False, |
| owner_defaults_to_member=False, component_required=False, owner_id=None, |
| labels=None, component_ids=None, admin_ids=None, field_values=None, |
| phases=None, approval_values=None): |
| template = tracker_bizobj.MakeIssueTemplate( |
| name, |
| summary, |
| status, |
| owner_id, |
| content, |
| labels, |
| field_values or [], |
| admin_ids or [], |
| component_ids, |
| summary_must_be_edited=summary_must_be_edited, |
| owner_defaults_to_member=owner_defaults_to_member, |
| component_required=component_required, |
| members_only=members_only, |
| phases=phases, |
| approval_values=approval_values) |
| template.template_id = template_id |
| self.templates_by_id[template_id] = template |
| if project_id not in self.templates_by_project_id: |
| self.templates_by_project_id[project_id] = [] |
| self.templates_by_project_id[project_id].append(template_id) |
| return template |
| |
| def GetTemplateByName(self, cnxn, template_name, project_id): |
| if project_id not in self.templates_by_project_id: |
| return None |
| else: |
| project_templates = self.templates_by_project_id[project_id] |
| for template_id in project_templates: |
| template = self.GetTemplateById(cnxn, template_id) |
| if template.name == template_name: |
| return template |
| return None |
| |
| def GetTemplateById(self, cnxn, template_id): |
| return self.templates_by_id.get(template_id) |
| |
| def GetTemplatesById(self, cnxn, template_ids): |
| return filter( |
| lambda template: template.template_id in template_ids, |
| self.templates_by_id.values()) |
| |
| def GetProjectTemplates(self, cnxn, project_id): |
| template_ids = self.templates_by_project_id[project_id] |
| return self.GetTemplatesById(cnxn, template_ids) |
| |
| def ExpungeUsersInTemplates(self, cnxn, user_ids, limit=None): |
| for _, template in self.templates_by_id.items(): |
| template.admin_ids = [user_id for user_id in template.admin_ids |
| if user_id not in user_ids] |
| if template.owner_id in user_ids: |
| template.owner_id = None |
| template.field_values = [fv for fv in template.field_values |
| if fv.user_id in user_ids] |
| |
| class SpamService(object): |
| """Fake version of SpamService that just works in-RAM.""" |
| |
| def __init__(self, user_id=None): |
| self.user_id = user_id |
| self.reports_by_issue_id = collections.defaultdict(list) |
| self.comment_reports_by_issue_id = collections.defaultdict(dict) |
| self.manual_verdicts_by_issue_id = collections.defaultdict(dict) |
| self.manual_verdicts_by_comment_id = collections.defaultdict(dict) |
| self.expunged_users_in_spam = [] |
| |
| def LookupIssuesFlaggers(self, cnxn, issue_ids): |
| return { |
| issue_id: (self.reports_by_issue_id.get(issue_id, []), |
| self.comment_reports_by_issue_id.get(issue_id, {})) |
| for issue_id in issue_ids} |
| |
| def LookupIssueFlaggers(self, cnxn, issue_id): |
| return self.LookupIssuesFlaggers(cnxn, [issue_id])[issue_id] |
| |
| def FlagIssues(self, cnxn, issue_service, issues, user_id, flagged_spam): |
| for issue in issues: |
| if flagged_spam: |
| self.reports_by_issue_id[issue.issue_id].append(user_id) |
| else: |
| self.reports_by_issue_id[issue.issue_id].remove(user_id) |
| |
| def FlagComment( |
| self, cnxn, issue, comment_id, reported_user_id, user_id, flagged_spam): |
| if not comment_id in self.comment_reports_by_issue_id[issue.issue_id]: |
| self.comment_reports_by_issue_id[issue.issue_id][comment_id] = [] |
| if flagged_spam: |
| self.comment_reports_by_issue_id[issue.issue_id][comment_id].append( |
| user_id) |
| else: |
| self.comment_reports_by_issue_id[issue.issue_id][comment_id].remove( |
| user_id) |
| |
| def RecordManualIssueVerdicts( |
| self, cnxn, issue_service, issues, user_id, is_spam): |
| for issue in issues: |
| self.manual_verdicts_by_issue_id[issue.issue_id][user_id] = is_spam |
| |
| def RecordManualCommentVerdict( |
| self, cnxn, issue_service, user_service, comment_id, |
| user_id, is_spam): |
| self.manual_verdicts_by_comment_id[comment_id][user_id] = is_spam |
| comment = issue_service.GetComment(cnxn, comment_id) |
| comment.is_spam = is_spam |
| issue = issue_service.GetIssue(cnxn, comment.issue_id, use_cache=False) |
| issue_service.SoftDeleteComment( |
| cnxn, issue, comment, user_id, user_service, is_spam, True, is_spam) |
| |
| def RecordClassifierIssueVerdict(self, cnxn, issue, is_spam, confidence, |
| failed_open): |
| return |
| |
| def RecordClassifierCommentVerdict(self, cnxn, issue, is_spam, confidence, |
| failed_open): |
| return |
| |
| def ClassifyComment(self, comment, commenter): |
| return {'outputLabel': 'ham', |
| 'outputMulti': [{'label': 'ham', 'score': '1.0'}], |
| 'failed_open': False} |
| |
| def ClassifyIssue(self, issue, firstComment, reporter): |
| return {'outputLabel': 'ham', |
| 'outputMulti': [{'label': 'ham', 'score': '1.0'}], |
| 'failed_open': False} |
| |
| def ExpungeUsersInSpam(self, cnxn, user_ids): |
| self.expunged_users_in_spam.extend(user_ids) |
| |
| |
| class FeaturesService(object): |
| """A fake implementation of FeaturesService.""" |
| def __init__(self): |
| # Test-only sequence of expunged projects and users. |
| self.expunged_saved_queries = [] |
| self.expunged_users_in_saved_queries = [] |
| self.expunged_filter_rules = [] |
| self.expunged_users_in_filter_rules = [] |
| self.expunged_quick_edit = [] |
| self.expunged_users_in_quick_edits = [] |
| self.expunged_hotlist_ids = [] |
| self.expunged_users_in_hotlists = [] |
| |
| # filter rules, project_id => filterrule_pb |
| self.test_rules = collections.defaultdict(list) |
| |
| # TODO(crbug/monorail/7104): Confirm that these are never reassigned |
| # to empty {} and then change these to collections.defaultdicts instead. |
| # hotlists |
| self.test_hotlists = {} # (hotlist_name, owner_id) => hotlist_pb |
| self.hotlists_by_id = {} |
| self.hotlists_id_by_user = {} # user_id => [hotlist_id, hotlist_id, ...] |
| self.hotlists_id_by_issue = {} # issue_id => [hotlist_id, hotlist_id, ...] |
| |
| # saved queries |
| self.saved_queries = [] # [(pid, uid, sq), ...] |
| |
| def TestAddFilterRule( |
| self, project_id, predicate, default_status=None, default_owner_id=None, |
| add_cc_ids=None, add_labels=None, add_notify=None, warning=None, |
| error=None): |
| rule = filterrules_helpers.MakeRule( |
| predicate, default_status=default_status, |
| default_owner_id=default_owner_id, add_cc_ids=add_cc_ids, |
| add_labels=add_labels, add_notify=add_notify, warning=warning, |
| error=error) |
| self.test_rules[project_id].append(rule) |
| return rule |
| |
| def TestAddHotlist(self, name, summary='', owner_ids=None, editor_ids=None, |
| follower_ids=None, description=None, hotlist_id=None, |
| is_private=False, hotlist_item_fields=None, |
| default_col_spec=None): |
| """Add a hotlist to the fake FeaturesService object. |
| |
| Args: |
| name: the name of the hotlist. Will replace any existing hotlist under |
| the same name. |
| summary: the summary string of the hotlist |
| owner_ids: List of user ids for the hotlist owners |
| editor_ids: List of user ids for the hotlist editors |
| follower_ids: List of user ids for the hotlist followers |
| description: The description string for this hotlist |
| hotlist_id: A unique integer identifier for the created hotlist |
| is_private: A boolean indicating whether the hotlist is private/public |
| hotlist_item_fields: a list of tuples -> |
| [(issue_id, rank, adder_id, date_added, note),...] |
| default_col_spec: string of default columns for the hotlist. |
| |
| Returns: |
| A populated hotlist PB. |
| """ |
| hotlist_pb = features_pb2.Hotlist() |
| hotlist_pb.hotlist_id = hotlist_id or hash(name) % 100000 |
| hotlist_pb.name = name |
| hotlist_pb.summary = summary |
| hotlist_pb.is_private = is_private |
| hotlist_pb.default_col_spec = default_col_spec |
| if description is not None: |
| hotlist_pb.description = description |
| |
| self.TestAddHotlistMembers(owner_ids, hotlist_pb, OWNER_ROLE) |
| self.TestAddHotlistMembers(follower_ids, hotlist_pb, FOLLOWER_ROLE) |
| self.TestAddHotlistMembers(editor_ids, hotlist_pb, EDITOR_ROLE) |
| |
| if hotlist_item_fields is not None: |
| for(issue_id, rank, adder_id, date, note) in hotlist_item_fields: |
| hotlist_pb.items.append( |
| features_pb2.Hotlist.HotlistItem( |
| issue_id=issue_id, rank=rank, adder_id=adder_id, |
| date_added=date, note=note)) |
| try: |
| self.hotlists_id_by_issue[issue_id].append(hotlist_pb.hotlist_id) |
| except KeyError: |
| self.hotlists_id_by_issue[issue_id] = [hotlist_pb.hotlist_id] |
| |
| owner_id = None |
| if hotlist_pb.owner_ids: |
| owner_id = hotlist_pb.owner_ids[0] |
| self.test_hotlists[(name, owner_id)] = hotlist_pb |
| self.hotlists_by_id[hotlist_pb.hotlist_id] = hotlist_pb |
| return hotlist_pb |
| |
| def TestAddHotlistMembers(self, user_id_list, hotlist_pb, role): |
| if user_id_list is not None: |
| for user_id in user_id_list: |
| if role == OWNER_ROLE: |
| hotlist_pb.owner_ids.append(user_id) |
| elif role == EDITOR_ROLE: |
| hotlist_pb.editor_ids.append(user_id) |
| elif role == FOLLOWER_ROLE: |
| hotlist_pb.follower_ids.append(user_id) |
| try: |
| self.hotlists_id_by_user[user_id].append(hotlist_pb.hotlist_id) |
| except KeyError: |
| self.hotlists_id_by_user[user_id] = [hotlist_pb.hotlist_id] |
| |
| def CheckHotlistName(self, cnxn, name, owner_ids): |
| if not framework_bizobj.IsValidHotlistName(name): |
| raise exceptions.InputException( |
| '%s is not a valid name for a Hotlist' % name) |
| if self.LookupHotlistIDs(cnxn, [name], owner_ids): |
| raise features_svc.HotlistAlreadyExists() |
| |
| def CreateHotlist( |
| self, _cnxn, hotlist_name, summary, description, owner_ids, editor_ids, |
| issue_ids=None, is_private=None, default_col_spec=None, ts=None): |
| """Create and store a Hotlist with the given attributes.""" |
| if not framework_bizobj.IsValidHotlistName(hotlist_name): |
| raise exceptions.InputException() |
| if not owner_ids: # Should never happen. |
| raise features_svc.UnownedHotlistException() |
| if (hotlist_name, owner_ids[0]) in self.test_hotlists: |
| raise features_svc.HotlistAlreadyExists() |
| hotlist_item_fields = [ |
| (issue_id, rank*100, owner_ids[0] or None, ts, '') for |
| rank, issue_id in enumerate(issue_ids or [])] |
| return self.TestAddHotlist(hotlist_name, summary=summary, |
| owner_ids=owner_ids, editor_ids=editor_ids, |
| description=description, is_private=is_private, |
| hotlist_item_fields=hotlist_item_fields, |
| default_col_spec=default_col_spec) |
| |
| def UpdateHotlist( |
| self, cnxn, hotlist_id, name=None, summary=None, description=None, |
| is_private=None, default_col_spec=None, owner_id=None, |
| add_editor_ids=None): |
| hotlist = self.hotlists_by_id.get(hotlist_id) |
| if not hotlist: |
| raise features_svc.NoSuchHotlistException( |
| 'Hotlist "%s" not found!' % hotlist_id) |
| |
| if owner_id: |
| old_owner_id = hotlist.owner_ids[0] |
| self.test_hotlists.pop((hotlist.name, old_owner_id), None) |
| self.test_hotlists[(hotlist.name, owner_id)] = hotlist |
| |
| if add_editor_ids: |
| for editor_id in add_editor_ids: |
| self.hotlists_id_by_user.get(editor_id, []).append(hotlist_id) |
| |
| if name is not None: |
| hotlist.name = name |
| if summary is not None: |
| hotlist.summary = summary |
| if description is not None: |
| hotlist.description = description |
| if is_private is not None: |
| hotlist.is_private = is_private |
| if default_col_spec is not None: |
| hotlist.default_col_spec = default_col_spec |
| if owner_id is not None: |
| hotlist.owner_ids = [owner_id] |
| if add_editor_ids: |
| hotlist.editor_ids.extend(add_editor_ids) |
| |
| def RemoveHotlistEditors(self, cnxn, hotlist_id, remove_editor_ids): |
| hotlist = self.hotlists_by_id.get(hotlist_id) |
| if not hotlist: |
| raise features_svc.NoSuchHotlistException( |
| 'Hotlist "%s" not found!' % hotlist_id) |
| for editor_id in remove_editor_ids: |
| hotlist.editor_ids.remove(editor_id) |
| self.hotlists_id_by_user[editor_id].remove(hotlist_id) |
| |
| def AddIssuesToHotlists(self, cnxn, hotlist_ids, added_tuples, issue_svc, |
| chart_svc, commit=True): |
| for hotlist_id in hotlist_ids: |
| self.UpdateHotlistItems(cnxn, hotlist_id, [], added_tuples, commit=commit) |
| |
| def RemoveIssuesFromHotlists(self, cnxn, hotlist_ids, issue_ids, issue_svc, |
| chart_svc, commit=True): |
| for hotlist_id in hotlist_ids: |
| self.UpdateHotlistItems(cnxn, hotlist_id, issue_ids, [], commit=commit) |
| |
| def UpdateHotlistIssues( |
| self, |
| cnxn, |
| hotlist_id, |
| updated_items, |
| remove_issue_ids, |
| issue_svc, |
| chart_svc, |
| commit=True): |
| if not updated_items and not remove_issue_ids: |
| raise exceptions.InputException('No changes to make') |
| |
| hotlist = self.hotlists_by_id.get(hotlist_id) |
| if not hotlist: |
| raise NoSuchHotlistException() |
| |
| updated_ids = [item.issue_id for item in updated_items] |
| items = [ |
| item for item in hotlist.items |
| if item.issue_id not in updated_ids + remove_issue_ids |
| ] |
| hotlist.items = sorted(updated_items + items, key=lambda item: item.rank) |
| |
| # Remove all removed and updated issues. |
| for issue_id in remove_issue_ids + updated_ids: |
| try: |
| self.hotlists_id_by_issue[issue_id].remove(hotlist_id) |
| except (ValueError, KeyError): |
| pass |
| # Add all new or updated issues. |
| for item in updated_items: |
| self.hotlists_id_by_issue.setdefault(item.issue_id, []).append(hotlist_id) |
| |
| def UpdateHotlistItems( |
| self, cnxn, hotlist_id, remove, added_issue_tuples, commit=True): |
| hotlist = self.hotlists_by_id.get(hotlist_id) |
| if not hotlist: |
| raise features_svc.NoSuchHotlistException( |
| 'Hotlist "%s" not found!' % hotlist_id) |
| current_issues_ids = { |
| item.issue_id for item in hotlist.items} |
| items = [ |
| item for item in hotlist.items if |
| item.issue_id not in remove] |
| |
| if hotlist.items: |
| items_sorted = sorted(hotlist.items, key=lambda item: item.rank) |
| rank_base = items_sorted[-1].rank + 10 |
| else: |
| rank_base = 1 |
| |
| new_hotlist_items = [ |
| features_pb2.MakeHotlistItem( |
| issue_id, rank+rank_base*10, adder_id, date, note) |
| for rank, (issue_id, adder_id, date, note) in |
| enumerate(added_issue_tuples) |
| if issue_id not in current_issues_ids] |
| items.extend(new_hotlist_items) |
| hotlist.items = items |
| |
| for issue_id in remove: |
| try: |
| self.hotlists_id_by_issue[issue_id].remove(hotlist_id) |
| except ValueError: |
| pass |
| for item in new_hotlist_items: |
| try: |
| self.hotlists_id_by_issue[item.issue_id].append(hotlist_id) |
| except KeyError: |
| self.hotlists_id_by_issue[item.issue_id] = [hotlist_id] |
| |
| def UpdateHotlistItemsFields( |
| self, cnxn, hotlist_id, new_ranks=None, new_notes=None, commit=True): |
| hotlist = self.hotlists_by_id.get(hotlist_id) |
| if not hotlist: |
| raise features_svc.NoSuchHotlistException( |
| 'Hotlist "%s" not found!' % hotlist_id) |
| if new_ranks is None: |
| new_ranks = {} |
| if new_notes is None: |
| new_notes = {} |
| for hotlist_item in hotlist.items: |
| if hotlist_item.issue_id in new_ranks: |
| hotlist_item.rank = new_ranks[hotlist_item.issue_id] |
| if hotlist_item.issue_id in new_notes: |
| hotlist_item.note = new_notes[hotlist_item.issue_id] |
| |
| hotlist.items.sort(key=lambda item: item.rank) |
| |
| def TransferHotlistOwnership( |
| self, cnxn, hotlist, new_owner_id, remain_editor, commit=True): |
| """Transfers ownership of a hotlist to a new owner.""" |
| new_editor_ids = hotlist.editor_ids |
| if remain_editor: |
| new_editor_ids.extend(hotlist.owner_ids) |
| if new_owner_id in new_editor_ids: |
| new_editor_ids.remove(new_owner_id) |
| new_follower_ids = hotlist.follower_ids |
| if new_owner_id in new_follower_ids: |
| new_follower_ids.remove(new_owner_id) |
| self.UpdateHotlistRoles( |
| cnxn, hotlist.hotlist_id, [new_owner_id], new_editor_ids, |
| new_follower_ids, commit=commit) |
| |
| def LookupUserHotlists(self, cnxn, user_ids): |
| """Return dict of {user_id: [hotlist_id, hotlist_id...]}.""" |
| users_hotlists_dict = { |
| user_id: self.hotlists_id_by_user.get(user_id, []) |
| for user_id in user_ids |
| } |
| return users_hotlists_dict |
| |
| def LookupIssueHotlists(self, cnxn, issue_ids): |
| """Return dict of {issue_id: [hotlist_id, hotlist_id...]}.""" |
| issues_hotlists_dict = { |
| issue_id: self.hotlists_id_by_issue[issue_id] |
| for issue_id in issue_ids |
| if issue_id in self.hotlists_id_by_issue} |
| return issues_hotlists_dict |
| |
| def LookupHotlistIDs(self, cnxn, hotlist_names, owner_ids): |
| id_dict = {} |
| for name in hotlist_names: |
| for owner_id in owner_ids: |
| hotlist = self.test_hotlists.get((name, owner_id)) |
| if hotlist: |
| if not hotlist.owner_ids: # Should never happen. |
| logging.warning( |
| 'Unowned Hotlist: id:%r, name:%r', hotlist.hotlist_id, |
| hotlist.name) |
| continue |
| id_dict[(name.lower(), owner_id)] = hotlist.hotlist_id |
| return id_dict |
| |
| def GetHotlists(self, cnxn, hotlist_ids, use_cache=True): |
| """Returns dict of {hotlist_id: hotlist PB}.""" |
| result = {} |
| for hotlist_id in hotlist_ids: |
| hotlist = self.hotlists_by_id.get(hotlist_id) |
| if hotlist: |
| result[hotlist_id] = hotlist |
| else: |
| raise features_svc.NoSuchHotlistException() |
| return result |
| |
| def GetHotlistsByUserID(self, cnxn, user_id, use_cache=True): |
| """Get a list of hotlist PBs for a given user.""" |
| hotlist_id_dict = self.LookupUserHotlists(cnxn, [user_id]) |
| hotlists = self.GetHotlists(cnxn, hotlist_id_dict.get( |
| user_id, []), use_cache=use_cache) |
| return list(hotlists.values()) |
| |
| def GetHotlistsByIssueID(self, cnxn, issue_id, use_cache=True): |
| """Get a list of hotlist PBs for a given issue.""" |
| hotlist_id_dict = self.LookupIssueHotlists(cnxn, [issue_id]) |
| hotlists = self.GetHotlists(cnxn, hotlist_id_dict.get( |
| issue_id, []), use_cache=use_cache) |
| return list(hotlists.values()) |
| |
| def GetHotlist(self, cnxn, hotlist_id, use_cache=True): |
| """Return hotlist PB.""" |
| hotlist_id_dict = self.GetHotlists(cnxn, [hotlist_id], use_cache=use_cache) |
| return hotlist_id_dict.get(hotlist_id) |
| |
| def GetHotlistsByID(self, cnxn, hotlist_ids, use_cache=True): |
| hotlists_dict = {} |
| missed_ids = [] |
| for hotlist_id in hotlist_ids: |
| hotlist = self.hotlists_by_id.get(hotlist_id) |
| if hotlist: |
| hotlists_dict[hotlist_id] = hotlist |
| else: |
| missed_ids.append(hotlist_id) |
| return hotlists_dict, missed_ids |
| |
| def GetHotlistByID(self, cnxn, hotlist_id, use_cache=True): |
| hotlists_dict, _ = self.GetHotlistsByID( |
| cnxn, [hotlist_id], use_cache=use_cache) |
| return hotlists_dict[hotlist_id] |
| |
| def UpdateHotlistRoles( |
| self, cnxn, hotlist_id, owner_ids, editor_ids, follower_ids, commit=True): |
| hotlist = self.hotlists_by_id.get(hotlist_id) |
| if not hotlist: |
| raise features_svc.NoSuchHotlistException( |
| 'Hotlist "%s" not found!' % hotlist_id) |
| |
| # Remove hotlist_ids to clear old roles |
| for user_id in (hotlist.owner_ids + hotlist.editor_ids + |
| hotlist.follower_ids): |
| if hotlist_id in self.hotlists_id_by_user[user_id]: |
| self.hotlists_id_by_user[user_id].remove(hotlist_id) |
| old_owner_id = None |
| if hotlist.owner_ids: |
| old_owner_id = hotlist.owner_ids[0] |
| self.test_hotlists.pop((hotlist.name, old_owner_id), None) |
| |
| hotlist.owner_ids = owner_ids |
| hotlist.editor_ids = editor_ids |
| hotlist.follower_ids = follower_ids |
| |
| # Add new hotlist roles |
| for user_id in owner_ids+editor_ids+follower_ids: |
| try: |
| if hotlist_id not in self.hotlists_id_by_user[user_id]: |
| self.hotlists_id_by_user[user_id].append(hotlist_id) |
| except KeyError: |
| self.hotlists_id_by_user[user_id] = [hotlist_id] |
| new_owner_id = None |
| if owner_ids: |
| new_owner_id = owner_ids[0] |
| self.test_hotlists[(hotlist.name, new_owner_id)] = hotlist |
| |
| def DeleteHotlist(self, cnxn, hotlist_id, commit=True): |
| hotlist = self.hotlists_by_id.pop(hotlist_id, None) |
| if hotlist is not None: |
| self.test_hotlists.pop((hotlist.name, hotlist.owner_ids[0]), None) |
| user_ids = hotlist.owner_ids+hotlist.editor_ids+hotlist.follower_ids |
| for user_id in user_ids: |
| try: |
| self.hotlists_id_by_user[user_id].remove(hotlist_id) |
| except (ValueError, KeyError): |
| pass |
| for item in hotlist.items: |
| try: |
| self.hotlists_id_by_issue[item.issue_id].remove(hotlist_id) |
| except (ValueError, KeyError): |
| pass |
| for owner_id in hotlist.owner_ids: |
| self.test_hotlists.pop((hotlist.name, owner_id), None) |
| |
| def ExpungeHotlists( |
| self, cnxn, hotlist_ids, star_svc, user_svc, chart_svc, commit=True): |
| self.expunged_hotlist_ids.extend(hotlist_ids) |
| for hotlist_id in hotlist_ids: |
| self.DeleteHotlist(cnxn, hotlist_id) |
| |
| def ExpungeUsersInHotlists( |
| self, cnxn, user_ids, star_svc, user_svc, chart_svc): |
| self.expunged_users_in_hotlists.extend(user_ids) |
| |
| # end of Hotlist functions |
| |
| def GetRecentCommands(self, cnxn, user_id, project_id): |
| return [], [] |
| |
| def ExpungeSavedQueriesExecuteInProject(self, _cnxn, project_id): |
| self.expunged_saved_queries.append(project_id) |
| |
| def ExpungeSavedQueriesByUsers(self, cnxn, user_ids, limit=None): |
| self.expunged_users_in_saved_queries.extend(user_ids) |
| |
| def ExpungeFilterRules(self, _cnxn, project_id): |
| self.expunged_filter_rules.append(project_id) |
| |
| def ExpungeFilterRulesByUser(self, cnxn, user_ids_by_email): |
| emails = user_ids_by_email.keys() |
| user_ids = user_ids_by_email.values() |
| project_rules_dict = collections.defaultdict(list) |
| for project_id, rules in self.test_rules.items(): |
| for rule in rules: |
| if rule.default_owner_id in user_ids: |
| project_rules_dict[project_id].append(rule) |
| continue |
| if any(cc_id in user_ids for cc_id in rule.add_cc_ids): |
| project_rules_dict[project_id].append(rule) |
| continue |
| if any(addr in emails for addr in rule.add_notify_addrs): |
| project_rules_dict[project_id].append(rule) |
| continue |
| if any((email in rule.predicate) for email in emails): |
| project_rules_dict[project_id].append(rule) |
| continue |
| self.test_rules[project_id] = [ |
| rule for rule in rules |
| if rule not in project_rules_dict[project_id]] |
| return project_rules_dict |
| |
| def ExpungeQuickEditHistory(self, _cnxn, project_id): |
| self.expunged_quick_edit.append(project_id) |
| |
| def ExpungeQuickEditsByUsers(self, cnxn, user_ids, limit=None): |
| self.expunged_users_in_quick_edits.extend(user_ids) |
| |
| def GetFilterRules(self, cnxn, project_id): |
| return self.test_rules[project_id] |
| |
| def GetCannedQueriesByProjectID(self, cnxn, project_id): |
| return [sq for (pid, _, sq) in self.saved_queries if pid == project_id] |
| |
| def GetSavedQueriesByUserID(self, cnxn, user_id): |
| return [sq for (_, uid, sq) in self.saved_queries if uid == user_id] |
| |
| def UpdateCannedQueries(self, cnxn, project_id, canned_queries): |
| self.saved_queries.extend( |
| [(project_id, None, cq) for cq in canned_queries]) |
| |
| def UpdateUserSavedQueries(self, cnxn, user_id, saved_queries): |
| self.saved_queries = [ |
| (pid, uid, sq) for (pid, uid, sq) in self.saved_queries |
| if uid != user_id] |
| for sq in saved_queries: |
| if sq.executes_in_project_ids: |
| self.saved_queries.extend( |
| [(eipid, user_id, sq) for eipid in sq.executes_in_project_ids]) |
| else: |
| self.saved_queries.append((None, user_id, sq)) |
| |
| def GetSubscriptionsInProjects(self, cnxn, project_ids): |
| sq_by_uid = {} |
| for pid, uid, sq in self.saved_queries: |
| if pid in project_ids: |
| if uid in sq_by_uid: |
| sq_by_uid[uid].append(sq) |
| else: |
| sq_by_uid[uid] = [sq] |
| |
| return sq_by_uid |
| |
| def GetSavedQuery(self, cnxn, query_id): |
| return tracker_pb2.SavedQuery() |
| |
| |
| class PostData(object): |
| """A dictionary-like object that also implements getall().""" |
| |
| def __init__(self, *args, **kwargs): |
| self.dictionary = dict(*args, **kwargs) |
| |
| def getall(self, key): |
| """Return all values, assume that the value at key is already a list.""" |
| return self.dictionary.get(key, []) |
| |
| def getlist(self, key): |
| """Return all values, assume that the value at key is already a list.""" |
| return self.dictionary.get(key, []) |
| |
| def get(self, key, default=None): |
| """Return first value, assume that the value at key is already a list.""" |
| return self.dictionary.get(key, [default])[0] |
| |
| def __getitem__(self, key): |
| """Return first value, assume that the value at key is already a list.""" |
| return self.dictionary[key][0] |
| |
| def __contains__(self, key): |
| return key in self.dictionary |
| |
| def keys(self): |
| """Return the keys in the POST data.""" |
| return list(self.dictionary.keys()) |
| |
| |
| class FakeFile: |
| def __init__(self, data=None): |
| self.data = data |
| |
| def read(self): |
| return self.data |
| |
| def write(self, content): |
| return |
| |
| def __enter__(self): |
| return self |
| |
| def __exit__(self, __1, __2, __3): |
| return None |
| |
| |
| def gcs_open(filename, mode): |
| return FakeFile(filename) |
