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
import enum
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:',
    'Requires:',
    '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:',
    'Auto-Submit',
    re.compile(r'^\w+-Auto-Submit:'),
]


def _match_tag(line, tag):
    if hasattr(tag, 'match'):
        return tag.match(line)
    return line.startswith(tag)


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((_match_tag(line, tag) for tag in FILTER_TAGS))
    )


class _Direction(enum.Enum):
    CURRENT = 'CURRENT'
    FORWARD = 'FORWARD'
    BACKWARD = 'BACKWARD'
    REBASE = 'REBASE'


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


@attr.s
class Roll(object):
    _api = attr.ib()
    project_name = attr.ib(type=str)
    old_revision = attr.ib(type=str)
    new_revision = attr.ib(type=str)
    proj_dir = attr.ib(type=str)
    direction = attr.ib(type=str)
    _commit_data = attr.ib(default=None)

    @direction.validator
    def check(self, _, value):  # pragma: no cover
        if value not in _Direction:
            raise ValueError('invalid direction: {}'.format(value))
        if value == _Direction.CURRENT:
            raise ValueError('attempt to do a no-op roll')

    @property
    def commits(self):
        if self._commit_data:
            return self._commit_data

        with self._api.context(cwd=self.proj_dir):
            with self._api.step.nest(self.project_name):
                commits = []
                if (
                    _is_hash(self.old_revision) and
                    self.direction == _Direction.FORWARD
                ):
                    base = self.old_revision
                else:
                    base = '{}~5'.format(self.new_revision)

                for commit in (
                    self._api.git(
                        'git log',
                        'log',
                        '{}..{}'.format(base, self.new_revision),
                        '--pretty=format:%H %B',
                        # Separate entries with null bytes since most entries
                        # will contain newlines ("%B" is the full commit
                        # message, not just the first line.)
                        '-z',
                        stdout=self._api.raw_io.output(),
                    )
                    .stdout.strip('\0')
                    .split('\0')
                ):
                    hash, message = commit.split(' ', 1)
                    commits.append(Commit(hash, message))

                self._commit_data = tuple(commits)
                return self._commit_data


@attr.s
class Message(object):
    name = attr.ib(type=str)
    template = attr.ib(type=str)
    kwargs = attr.ib(type=dict)
    num_commits = attr.ib(type=int)
    footer = attr.ib(type=tuple, default=())

    def render(self, with_footer=True):
        result = [self.template.format(**self.kwargs)]
        if with_footer:
            result.extend(x for x in self.footer)
        return '\n'.join(result)


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


class RollUtilApi(recipe_api.RecipeApi):
    def __init__(self, props, *args, **kwargs):
        super(RollUtilApi, self).__init__(*args, **kwargs)
        self.labels_to_set = {x.label: x.value for x in props.labels_to_set}
        self.labels_to_wait_on = props.labels_to_wait_on
        self.footer = ['CQ-Do-Not-Cancel-Tryjobs: true']

    def _single_commit_roll_message(self, roll):
        template = """
[roll {project_name}] {sanitized_message}

{project_name} Rolled-Commits: {old_revision:.15}..{new_revision:.15}
        """.strip()

        commit = roll.commits[0]

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

        message = Message(
            name=roll.project_name,
            template=template,
            kwargs=kwargs,
            num_commits=1,
            footer=tuple(self.footer),
        )

        with self.m.step.nest(
            'message for {}'.format(roll.project_name)
        ) as pres:
            pres.logs['template'] = template
            pres.logs['kwargs'] = pprint.pformat(kwargs)
            pres.logs['message'] = message.render()

        return message

    def _multiple_commits_roll_message(self, roll):
        template = """
[{project_name}] Roll {num_commits} commits

{one_liners}

{project_name} Rolled-Commits: {old_revision:.15}..{new_revision:.15}
    """.strip()

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

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

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

        message = Message(
            name=roll.project_name,
            template=template,
            kwargs=kwargs,
            num_commits=num_commits,
            footer=tuple(self.footer),
        )

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

        return message

    def _single_roll_message(self, roll):
        if len(roll.commits) > 1:
            return self._multiple_commits_roll_message(roll)
        return self._single_commit_roll_message(roll)

    def _multiple_rolls_message(self, *rolls):
        rolls = sorted(rolls, key=lambda x: x.project_name)

        messages = []
        for roll in rolls:
            messages.append(self._single_roll_message(roll))

        texts = [
            '[roll {}] Roll {} commits'.format(
                ', '.join(x.name for x in messages),
                sum(x.num_commits for x in messages),
            )
        ]
        texts.extend(x.render(with_footer=False) for x in messages)
        texts.append('\n'.join('{}'.format(x) for x in self.footer))

        return '\n\n'.join(texts)

    def Roll(self, **kwargs):
        """Create a Roll. See Roll class above for details."""
        return Roll(api=self.m, **kwargs)

    def message(self, *rolls):
        with self.m.step.nest('roll message'):
            if len(rolls) > 1:
                return self._multiple_rolls_message(*rolls)
            return self._single_roll_message(*rolls).render()

    Direction = _Direction

    def get_roll_direction(
        self, git_dir, old, new, name='get roll direction'
    ):
        """Return Direction of roll."""
        if old == new:
            with self.m.step.nest(name) as pres:
                pres.step_summary_text = 'up-to-date'
            return self.Direction.CURRENT

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

                backward = self.m.git(
                    'is backward',
                    'merge-base',
                    '--is-ancestor',
                    new,
                    old,
                    ok_ret=(0, 1),
                )

                if (
                    forward.exc_result.retcode == 0
                    and backward.exc_result.retcode != 0
                ):
                    pres.step_summary_text = 'forward'
                    return self.Direction.FORWARD

                if (
                    forward.exc_result.retcode != 0
                    and backward.exc_result.retcode == 0
                ):
                    pres.step_summary_text = 'backward'
                    return self.Direction.BACKWARD

                # If old is not an ancestor of new and new is not an ancestor
                # of old then history was rewritten in some manner but we still
                # need to update the pin.
                pres.step_summary_text = 'rebase'
                return self.Direction.REBASE

    def can_roll(self, direction):
        return direction in (self.Direction.FORWARD, self.Direction.REBASE)

    def skip_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)