exonerator/gerrit_cq.py

# Copyright 2017 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Functions for forgiving false CQ rejections."""

from __future__ import print_function
from __future__ import absolute_import
from __future__ import division

import itertools

from chromite.lib import cros_logging as logging
from chromite.lib import gerrit

from exonerator import constants


COMMIT_QUEUE_LABEL = 'Commit-Queue'
CODE_REVIEW_LABEL = 'Code-Review'
# TODO(phobbs) is there a way to make this more robust than just checking a
# substring? Maybe a shared constant with chromite?
_FAILED_TO_APPLY_SNIPPET = 'failed to apply your change'
_VERIFIED_LABEL = 'Verified'
_EXONERATOR_CQ_MESSAGE = """\
This CL has been marked innocent in the CL annotator; re-marking as CQ+1.\
  See the annotator entry here:
https://chromiumos-build-annotator.googleplex.com/build_annotations/\
edit_annotations/master-paladin/%s"""
# TODO(phobbs): Make the reason dependent on how the CL was exonerated - maybe
# use the message_subtype of the annotations_finalized buildMessage?


def MaybeExonerate(change, build_id):
  """Mark a CL as CQ+1 only if it hasn't already been.

  Args:
    change: A GerritPatchTuple.
    build_id: The CIDB id for the build which the change was kicked out of.
  """
  helper = GetHelper(internal=change.internal)
  details = helper.GetChangeDetail(change.gerrit_number)
  if not details:
    logging.warning('Change details was None for %s', str(change))
    return False

  if not _ShouldBeMarkedCQReady(details, change.patch_number):
    logging.info("Change %s was not ready", str(change))
    return False

  try:
    helper.SetReview(
        change.gerrit_number,
        msg=_EXONERATOR_CQ_MESSAGE % build_id,
        labels={COMMIT_QUEUE_LABEL: 1})
    logging.info('Exonerated %s for a CQ failure.', str(change))
    return True
  # Network errors or permission errors won't get retried. That's ok for now.
  except Exception:
    logging.exception('Encountered an exception while marking %s with CQ+1',
                      str(change))
    return False


def CanBeMarkedReady(change):
  """Whether a change is ready to be exonerated.

  Args:
    change: A GerritPatchTuple to test.
  """
  helper = GetHelper(internal=change.internal)
  details = helper.GetChangeDetail(change.gerrit_number)
  if details:
    return _ShouldBeMarkedCQReady(details, change.patch_number)
  else:
    logging.warning('Change details was None for %s', str(change))
    return False


def SanityCheckChange(change_details, known_patch_number):
  """Whether the change's latest revision is the known patch number.

  Check that the change's details contains 'revisions', and that the latest
  revision matches the patch number we want to forgive.

  Args:
    change_details: The change details response from Gerrit.
    known_patch_number: The revision number which we want to forgive.
  """
  if 'revisions' not in change_details:
    logging.warning('Change details has no revisions')
    return False

  # If a newer patchset was uploaded, we don't want to forgive it.
  missing = -1
  max_rev = max(rev.get('_number', missing)
                for rev in change_details['revisions'].values())
  return not (max_rev == missing or max_rev > known_patch_number)


def _ShouldBeMarkedCQReady(change_details, known_patch_number):
  """Whether a CL should be re-marked as CQ ready.

  We only want to re-label CLs as CQ+1 if they haven't been marked as CQ+1 or
  CQ-1 by someone else, and it's still on the same patchset that was kicked
  out.

  Args:
    change_details: The change details response from Gerrit.
    known_patch_number: The revision number which we want to forgive.
  """
  return (
      SanityCheckChange(change_details, known_patch_number)
      # It shouldn't already have CQ+1 or CQ-1
      and not (LabelValues(change_details, COMMIT_QUEUE_LABEL) - {0})
      # The CL must have Verified+1 still.
      and any(label > 0
              for label in LabelValues(change_details, _VERIFIED_LABEL))
      # Must not have Verified-1
      and not any(label < 0
                  for label in LabelValues(change_details, _VERIFIED_LABEL))

      # Don't exonerate CLs with Code-Review-1 or Code-Review-2
      and not any(label < 0
                  for label in LabelValues(change_details, CODE_REVIEW_LABEL))
      and CQApprovalWasRemovedByBot(change_details)
      and not _FailedToApply(change_details)
  )


def CQApprovalWasRemovedByBot(change_details):
  """Returns whether the CQ+1 label was removed by a bot.

  Args:
    change_details: The response from gerrit with the change's details.
  """
  for message in reversed(change_details['messages']):
    if '-Commit-Queue' in message['message']:
      return (message['real_author']['email']
              == constants.CHROMEOS_COMMIT_BOT_EMAIL)

  logging.error('No Commit-Queue removal message was found for change %s.'
                ' This should never happen' % (
                    change_details.get('_number', change_details['id'])))
  return False


def _FailedToApply(change_details):
  """Tests for whether the change failed to apply recently.

  Checks to see whether the second to last message by chromeos-commit-bot
  indicates that the Commit Queue failed to apply the change.

  Args:
    change_details: The change details response from Gerrit.
  """
  commit_bot_messages = [
      m for m in change_details['messages']
      if m['real_author']['email'] == constants.CHROMEOS_COMMIT_BOT_EMAIL
  ]

  last_two_messages = itertools.islice(reversed(commit_bot_messages), 2)
  return any(
      _FAILED_TO_APPLY_SNIPPET in m['message']
      for m in last_two_messages)


def LabelValues(change_details, label):
  """Returns a set of values that reviewers have given for |label|.

  Args:
    change_details: The change details object from Gerrit.
    label: The label, e.g. 'Commit-Queue'

  Returns:
    A set of values that the reviewers have given the change.
  """
  return set(entry['value']
             for entry in change_details['labels'][label].get('all', ())
             if 'value' in entry)


def GetHelper(internal):
  """Returns the GerritHelper for internal or external changes.

  Args:
    internal: Boolean indicating whether to get the internal or external helper.
  """
  if internal:
    return gerrit.GetCrosInternal()
  else:
    return gerrit.GetCrosExternal()