recipe_modules/roll_util/api.py

# Copyright 2020 The Pigweed Authors
#
# 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
#
#     https://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.
"""Utility functions for rollers."""

import pprint
import re

import attr
from recipe_engine import recipe_api

# If we're embedding the original commit message, prepend 'Original-' to lines
# which begin with these tags.
ESCAPE_TAGS = [
    'Bug:',
    'Fixed:',
    'Reviewed-on:',
]

# If we're embedding the original commit message, remove lines which contain
# these tags.
FILTER_TAGS = [
    'API-Review:',
    'Acked-by:',
    'CC:',
    'CQ-Do-Not-Cancel-Tryjobs:',
    'Change-Id:',
    'Commit-Queue:',
    'Reviewed-by:',
    'Signed-off-by:',
    'Testability-Review:',
    'Tested-by:',
]


def _sanitize_message(message):
    """Sanitize lines of a commit message.

    Prepend 'Original-' to lines which begin with ESCAPE_TAGS. Filter
    out lines which begin with FILTER_TAGS.
    """
    return '\n'.join(
        "Original-" + line
        if any((line.startswith(tag) for tag in ESCAPE_TAGS))
        else line
        for line in message.splitlines()
        if not any((tag in line for tag in FILTER_TAGS))
    )


@attr.s
class Commit(object):
    hash = attr.ib()
    message = attr.ib()


def _is_hash(value):
    return re.match(r'^[0-9a-fA-F]{40}', value)


class RollUtilApi(recipe_api.RecipeApi):
    def _single_commit_roll_message(
        self, project_name, commit, old_revision, new_revision
    ):
        template = """
[roll {project_name}] {sanitized_message}

Rolled-Commits: {old_revision:.15}..{new_revision:.15}
CQ-Do-Not-Cancel-Tryjobs: true
    """.strip()

        kwargs = {
            'project_name': project_name,
            'original_message': commit.message,
            'sanitized_message': _sanitize_message(commit.message),
            'old_revision': old_revision,
            'new_revision': new_revision,
        }

        message = template.format(**kwargs)

        with self.m.step.nest('message') as pres:
            pres.logs['template'] = template
            pres.logs['kwargs'] = pprint.pformat(kwargs)
            pres.logs['message'] = message

        return message

    def _multiple_commits_roll_message(
        self, project_name, commits, old_revision, new_revision
    ):
        template = """
[{project_name}] Roll {number} commits

{one_liners}

Rolled-Commits: {old_revision:.15}..{new_revision:.15}
CQ-Do-Not-Cancel-Tryjobs: true
    """.strip()

        one_liners = [
            '{:.15} {}'.format(commit.hash, commit.message.splitlines()[0][:50])
            for commit in commits
        ]

        number = len(commits)
        if not _is_hash(old_revision):
            number = 'multiple'
            one_liners.append('...')

        kwargs = {
            'project_name': project_name,
            'number': number,
            'one_liners': '\n'.join(one_liners),
            'old_revision': old_revision,
            'new_revision': new_revision,
        }

        message = template.format(**kwargs)

        with self.m.step.nest('message') as pres:
            pres.logs['template'] = template
            pres.logs['kwargs'] = pprint.pformat(kwargs)
            pres.logs['message'] = message

        return message

    def _commits(self, proj_dir, old_revision, new_revision):
        with self.m.context(cwd=proj_dir):
            commits = []
            if _is_hash(old_revision):
                base = old_revision
            else:
                base = '{}~5'.format(new_revision)

            for commit in (
                self.m.git(
                    'log',
                    '{}..{}'.format(base, new_revision),
                    '--pretty=format:%H %B',
                    '-z',
                    stdout=self.m.raw_io.output(),
                )
                .stdout.strip('\0')
                .split('\0')
            ):
                hash, message = commit.split(' ', 1)
                commits.append(Commit(hash, message))
            return commits

    def message(self, project_name, proj_dir, old_revision, new_revision):
        with self.m.step.nest('roll message'):
            commits = self._commits(proj_dir, old_revision, new_revision)

            if len(commits) > 1:
                return self._multiple_commits_roll_message(
                    project_name, commits, old_revision, new_revision
                )

            else:
                return self._single_commit_roll_message(
                    project_name, commits[0], old_revision, new_revision
                )

    def check_roll_direction(
        self, git_dir, old, new, name='check roll direction'
    ):
        """Return if old is an ancestor of new (i.e., a "forward" roll)."""
        if old == new:
            with self.m.step.nest(name) as pres:
                pres.step_summary_text = 'up-to-date'
            return False

        with self.m.context(git_dir):
            step = self.m.git(
                'merge-base',
                '--is-ancestor',
                old,
                new,
                name=name,
                ok_ret=(0, 1),
            )

            step.presentation.step_summary_text = 'backward'
            if step.exc_result.retcode == 0:
                step.presentation.step_summary_text = 'forward'

            return step.exc_result.retcode == 0

    def backwards_roll_step(self, remote, old_revision, new_revision):
        with self.m.step.nest('cancelling roll') as pres:
            fmt = (
                'not updating from {old} to {new} because {old} is newer '
                'than {new}'
            )
            if old_revision == new_revision:
                fmt = (
                    'not updating from {old} to {new} because they are '
                    'identical'
                )
            pres.step_summary_text = fmt.format(
                old=old_revision[0:7], new=new_revision[0:7]
            )
            pres.links[old_revision] = '{}/+/{}'.format(remote, old_revision)
            pres.links[new_revision] = '{}/+/{}'.format(remote, new_revision)