289 строки
11 KiB
Python
289 строки
11 KiB
Python
# -*- coding: utf-8 -*-
|
|
# Copyright 2022 Google Inc.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License")
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
import logging
|
|
from typing import Any, Tuple
|
|
|
|
from chromestatus_openapi.models import (GetVotesResponse, GetGateResponse, PostGateRequest, SuccessMessage)
|
|
from google.cloud import ndb
|
|
from google.cloud.ndb.tasklets import Future # for type checking only
|
|
|
|
from api import converters
|
|
from framework import basehandlers, permissions
|
|
from framework.users import User
|
|
from internals import approval_defs, notifier_helpers
|
|
from internals.core_enums import *
|
|
from internals.core_models import FeatureEntry, Stage
|
|
from internals.review_models import Gate, Vote, Amendment, Activity
|
|
|
|
|
|
def get_user_feature_and_gate(handler, kwargs) -> Tuple[
|
|
User, FeatureEntry, Gate, int, int]:
|
|
"""Get common parameters from the request."""
|
|
feature_id: int = kwargs['feature_id']
|
|
gate_id: int = kwargs['gate_id']
|
|
fe: FeatureEntry = handler.get_specified_feature(feature_id=feature_id)
|
|
|
|
user: User = handler.get_current_user(required=True)
|
|
gate: Gate = Gate.get_by_id(gate_id)
|
|
if not gate:
|
|
handler.abort(404, msg='Gate not found')
|
|
if gate.feature_id != feature_id:
|
|
handler.abort(400, msg='Mismatched feature and gate')
|
|
|
|
return user, fe, gate, feature_id, gate_id
|
|
|
|
|
|
class VotesAPI(basehandlers.APIHandler):
|
|
"""Users may see the set of votes on a feature, and add their own,
|
|
if allowed."""
|
|
|
|
def do_get(self, **kwargs) -> dict[str, list[dict[str, Any]]]:
|
|
"""Return a list of all vote values for a given feature."""
|
|
feature_id = kwargs['feature_id']
|
|
gate_id = kwargs.get('gate_id', None)
|
|
# Note: We assume that anyone may view approvals.
|
|
votes = Vote.get_votes(feature_id=feature_id, gate_id=gate_id)
|
|
dicts = [converters.vote_value_to_json_dict(v) for v in votes]
|
|
return GetVotesResponse.from_dict({'votes': dicts}).to_dict()
|
|
|
|
def do_post(self, **kwargs) -> dict[str, str]:
|
|
"""Set a user's vote value for the specified feature and gate."""
|
|
user, fe, gate, feature_id, gate_id = get_user_feature_and_gate(
|
|
self, kwargs)
|
|
new_state = self.get_int_param('state', validator=Vote.is_valid_state)
|
|
|
|
old_votes = Vote.get_votes(
|
|
feature_id=feature_id, gate_id=gate_id, set_by=user.email())
|
|
old_state = old_votes[0].state if old_votes else Vote.NO_RESPONSE
|
|
self.require_permissions(user, fe, gate, new_state)
|
|
|
|
# Note: We no longer write Approval entities.
|
|
old_gate_state = gate.state
|
|
new_gate_state = approval_defs.set_vote(feature_id, None, new_state,
|
|
user.email(), gate_id)
|
|
|
|
recently_approved = (old_gate_state not in (Vote.APPROVED, Vote.NA) and
|
|
new_gate_state in (Vote.APPROVED, Vote.NA))
|
|
# Notify that trial extension has been approved.
|
|
if gate.gate_type == GATE_API_EXTEND_ORIGIN_TRIAL and recently_approved:
|
|
stage = Stage.get_by_id(gate.stage_id)
|
|
notifier_helpers.send_trial_extension_approved_notification(
|
|
fe, stage, gate_id)
|
|
|
|
if new_state in (Vote.REVIEW_REQUESTED, Vote.NA_REQUESTED):
|
|
old_assignees = gate.assignee_emails[:]
|
|
approval_defs.auto_assign_reviewer(gate)
|
|
new_assignees = gate.assignee_emails[:]
|
|
if old_assignees != new_assignees:
|
|
amendment = Amendment(
|
|
field_name='review_assignee',
|
|
old_value=', '.join(old_assignees),
|
|
new_value=', '.join(new_assignees))
|
|
activity = Activity(
|
|
feature_id=fe.key.integer_id(), gate_id=gate_id,
|
|
author=user.email(), amendments=[amendment])
|
|
activity.put()
|
|
# Note: We don't notify assignee about autoassignment, because they
|
|
# will get a notification of review state change anyway.
|
|
|
|
notifier_helpers.notify_approvers_of_reviews(
|
|
fe, gate, new_state, user.email())
|
|
else:
|
|
notifier_helpers.notify_subscribers_of_vote_changes(
|
|
fe, gate, user.email(), new_state, old_state)
|
|
|
|
# Callers don't use the JSON response for this API call.
|
|
return SuccessMessage(message='Done').to_dict()
|
|
|
|
def require_permissions(self, user, feature, gate, new_state):
|
|
"""Abort the request if the user lacks permission to set this vote."""
|
|
is_requesting_review = new_state in (Vote.REVIEW_REQUESTED, Vote.NA_REQUESTED)
|
|
is_editor = permissions.can_edit_feature(user, feature.key.integer_id())
|
|
approvers = approval_defs.get_approvers(gate.gate_type)
|
|
is_approver = permissions.can_review_gate(user, feature, gate, approvers)
|
|
|
|
if is_requesting_review and is_editor:
|
|
return
|
|
|
|
if is_approver:
|
|
return
|
|
|
|
if is_requesting_review:
|
|
self.abort(403, msg='User may not request a review')
|
|
else:
|
|
self.abort(403, msg='User is not an approver')
|
|
|
|
|
|
class GatesAPI(basehandlers.APIHandler):
|
|
"""Get gates for a feature."""
|
|
|
|
def do_get(self, **kwargs) -> dict[str, Any]:
|
|
"""Return a list of all gates associated with the given feature."""
|
|
feature_id = kwargs.get('feature_id', None)
|
|
gates: list[Gate] = Gate.query(Gate.feature_id == feature_id).fetch()
|
|
|
|
# No gates associated with this feature.
|
|
if len(gates) == 0:
|
|
return {
|
|
'gates': [],
|
|
}
|
|
|
|
dicts = [converters.gate_value_to_json_dict(g) for g in gates]
|
|
for g in dicts:
|
|
approvers = approval_defs.get_approvers(g['gate_type'])
|
|
g['possible_assignee_emails'] = approvers
|
|
|
|
return GetGateResponse.from_dict({
|
|
'gates': dicts,
|
|
}).to_dict()
|
|
|
|
def do_post(self, **kwargs) -> dict[str, str]:
|
|
"""Set the assignees for a gate."""
|
|
user, fe, gate, feature_id, gate_id = get_user_feature_and_gate(
|
|
self, kwargs)
|
|
request = PostGateRequest.from_dict(self.request.json)
|
|
assignees = request.assignees
|
|
|
|
self.require_permissions(user, fe, gate)
|
|
self.validate_assignees(assignees, fe, gate)
|
|
old_assignees = gate.assignee_emails
|
|
gate.assignee_emails = assignees
|
|
gate.put()
|
|
notifier_helpers.notify_assignees(
|
|
fe, gate, user.email(), old_assignees, assignees)
|
|
|
|
# Callers don't use the JSON response for this API call.
|
|
return SuccessMessage(message='Done').to_dict()
|
|
|
|
def require_permissions(self, user, feature, gate):
|
|
"""Abort the request if the user lacks permission to set assignees."""
|
|
is_editor = permissions.can_edit_feature(user, feature.key.integer_id())
|
|
approvers = approval_defs.get_approvers(gate.gate_type)
|
|
is_approver = permissions.can_review_gate(user, feature, gate, approvers)
|
|
|
|
if not is_editor and not is_approver:
|
|
self.abort(403, msg='User lacks permission to assign reviews')
|
|
|
|
def validate_assignees(self, assignees, fe, gate):
|
|
"""Each assignee must have permission to review."""
|
|
approvers = approval_defs.get_approvers(gate.gate_type)
|
|
for email in assignees:
|
|
reviewer = User(email=email)
|
|
is_approver = permissions.can_review_gate(reviewer, fe, gate, approvers)
|
|
if not is_approver:
|
|
self.abort(400, 'Assignee is not a reviewer')
|
|
|
|
|
|
class PendingGatesAPI(basehandlers.APIHandler):
|
|
|
|
def get_stage_ids_of_gates_pending_my_approval(self) -> list[int] | Future:
|
|
"""Return a list of stage_id needing approval by current user."""
|
|
user = self.get_current_user()
|
|
if not user:
|
|
return []
|
|
|
|
approvable_gate_types = approval_defs.fields_approvable_by(user)
|
|
if not approvable_gate_types:
|
|
logging.info('User has no approvable_gate_types')
|
|
return []
|
|
query = Gate.query(
|
|
Gate.state.IN(Gate.PENDING_STATES),
|
|
Gate.gate_type.IN(approvable_gate_types))
|
|
future_stage_ids = query.fetch_async(projection=['stage_id'])
|
|
return future_stage_ids
|
|
|
|
def do_get(self, **kwargs):
|
|
"""Return a list of gates on features that have some active gates."""
|
|
# 1. Get the feature IDs of any active gates. Also, prefetch approvers.
|
|
future_projections = self.get_stage_ids_of_gates_pending_my_approval()
|
|
prefetched_approvers = {
|
|
gate_type: approval_defs.get_approvers(gate_type)
|
|
for gate_type in approval_defs.APPROVAL_FIELDS_BY_ID}
|
|
if type(future_projections) == list:
|
|
stage_ids = set(future_projections)
|
|
else:
|
|
projections = future_projections.get_result()
|
|
stage_ids = set(proj.stage_id for proj in projections)
|
|
if not stage_ids:
|
|
return GetGateResponse.from_dict({'gates': []}).to_dict()
|
|
|
|
# 2. Fetch all the gates on those stages.
|
|
gates: list[Gate] = Gate.query(Gate.stage_id.IN(stage_ids)).fetch()
|
|
|
|
# 3. Convert to dicts and add possible assignees.
|
|
dicts = [converters.gate_value_to_json_dict(g) for g in gates]
|
|
for g in dicts:
|
|
g['possible_assignee_emails'] = prefetched_approvers.get(g['gate_type'], [])
|
|
|
|
return GetGateResponse.from_dict({'gates': dicts}).to_dict()
|
|
|
|
|
|
class XfnGatesAPI(basehandlers.APIHandler):
|
|
|
|
def do_get(self, **kwargs):
|
|
"""Reject unneeded GET requests without triggering Error Reporting."""
|
|
self.abort(405, valid_methods=['POST'])
|
|
|
|
def do_post(self, **kwargs) -> dict[str, str]:
|
|
"""Add a full set of cross-functional gates to a stage."""
|
|
feature_id: int = kwargs['feature_id']
|
|
fe: FeatureEntry | None = self.get_specified_feature(feature_id=feature_id)
|
|
if fe is None:
|
|
self.abort(404, msg=f'Feature {feature_id} not found')
|
|
stage_id: int = kwargs['stage_id']
|
|
stage: Stage | None = Stage.get_by_id(stage_id)
|
|
if stage is None:
|
|
self.abort(404, msg=f'Stage {stage_id} not found')
|
|
|
|
user: User = self.get_current_user(required=True)
|
|
is_editor = fe and permissions.can_edit_feature(user, fe.key.integer_id())
|
|
is_approver = approval_defs.fields_approvable_by(user)
|
|
if not is_editor and not is_approver:
|
|
self.abort(403, msg='User lacks permission to create gates')
|
|
|
|
count = self.create_xfn_gates(feature_id, stage_id)
|
|
return SuccessMessage(message=f'Created {count} gates').to_dict()
|
|
|
|
def get_needed_gate_types(self) -> list[int]:
|
|
"""Return a list of gate types normally used to ship a new feature."""
|
|
needed_gate_tuples = STAGES_AND_GATES_BY_FEATURE_TYPE[
|
|
FEATURE_TYPE_INCUBATE_ID]
|
|
for stage_type, gate_types in needed_gate_tuples:
|
|
if stage_type == STAGE_BLINK_SHIPPING:
|
|
return gate_types
|
|
raise ValueError('Could not find expected list of gate types')
|
|
|
|
def create_xfn_gates(self, feature_id, stage_id) -> int:
|
|
"""Create all new incubation gates on a PSA stage"""
|
|
logging.info('Creating xfn gates')
|
|
existing_gates = Gate.query(
|
|
Gate.feature_id == feature_id, Gate.stage_id == stage_id).fetch()
|
|
existing_gate_types = set([eg.gate_type for eg in existing_gates])
|
|
logging.info('Found existing: %r', existing_gate_types)
|
|
new_gates = []
|
|
for gate_type in self.get_needed_gate_types():
|
|
if gate_type not in existing_gate_types:
|
|
logging.info(f'Creating gate type {gate_type}')
|
|
gate = Gate(
|
|
feature_id=feature_id, stage_id=stage_id, gate_type=gate_type,
|
|
state=Gate.PREPARING)
|
|
new_gates.append(gate)
|
|
|
|
ndb.put_multi(new_gates)
|
|
num_new = len(new_gates)
|
|
logging.info(f'Created {num_new} gates')
|
|
return num_new
|