recipes/txt_roller.py

# Copyright 2022 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.
"""Roll a pin of a git repository in a simple text file."""

import re

import attr
from PB.recipes.pigweed.txt_roller import InputProperties
from recipe_engine import post_process
from six.moves import configparser
from six.moves import urllib
from six import StringIO

DEPS = [
    'fuchsia/auto_roller',
    'fuchsia/status_check',
    'pigweed/checkout',
    'pigweed/roll_util',
    'recipe_engine/buildbucket',
    'recipe_engine/file',
    'recipe_engine/path',
    'recipe_engine/properties',
    'recipe_engine/step',
]

PROPERTIES = InputProperties

PYTHON_VERSION_COMPATIBILITY = "PY3"


def RunSteps(api, props):  # pylint: disable=invalid-name
    txt_path = props.txt_path
    project_remote = props.project_remote
    project_branch = props.project_branch or 'main'

    # The checkout module will try to use trigger data to pull in a specific
    # patch. Since the triggering commit is in a different repository that
    # needs to be disabled.
    api.checkout(use_trigger=False)

    new_revision = None

    # First, try to get new_revision from the trigger.
    bb_remote = None
    commit = api.buildbucket.build.input.gitiles_commit
    if commit and commit.project:
        new_revision = commit.id
        host = commit.host
        bb_remote = 'https://{}/{}'.format(host, commit.project)

    # If we still don't have a revision then it wasn't in the trigger. (Perhaps
    # this was manually triggered.) In this case we update to the
    # property-specified branch HEAD.
    if new_revision is None:
        new_revision = project_branch

    # If this was triggered by a gitiles poller, check that the triggering
    # repository matches project_remote.

    if bb_remote:
        if not api.checkout.remotes_equivalent(project_remote, bb_remote):
            raise api.step.StepFailure(
                'triggering repository ({}) does not match project remote '
                '({})'.format(bb_remote, project_remote)
            )

    project_dir = api.path['start_dir'].join('project')
    api.checkout(project_remote, branch=project_branch, root=project_dir)

    # In case new_revision is a branch name we need to retrieve the hash it
    # resolves to.
    if not re.search(r'^[0-9a-f]{40}$', new_revision):
        new_revision = api.checkout.get_revision(
            project_dir, 'get new revision', test_data='2' * 40
        )

    full_txt_path = api.checkout.root.join(txt_path)

    old_revision = api.file.read_text(
        'read old revision', full_txt_path, test_data='1' * 40,
    ).strip()

    api.file.write_text('write new revision', full_txt_path, new_revision)

    direction = api.roll_util.get_roll_direction(
        project_dir, old_revision, new_revision
    )

    # If the primary roll is not necessary or is backwards we can exit
    # immediately.
    if not api.roll_util.can_roll(direction):
        api.roll_util.skip_roll_step(project_remote, old_revision, new_revision)
        return

    with api.step.nest('txt_path') as pres:
        pres.step_summary_text = repr(txt_path)

    rolls = {
        txt_path: api.roll_util.Roll(
            project_name=txt_path,
            old_revision=old_revision,
            new_revision=new_revision,
            proj_dir=project_dir,
            direction=direction,
        ),
    }

    authors = api.roll_util.authors(*rolls.values())

    author_override = None
    if len(authors) == 1 and props.forge_author:
        author_override = api.roll_util.fake_author(
            next(iter(authors))
        )._asdict()

    change = api.auto_roller.attempt_roll(
        gerrit_host=api.checkout.gerrit_host(),
        gerrit_project=api.checkout.gerrit_project(),
        upstream_ref=api.checkout.branch,
        repo_dir=api.checkout.root,
        commit_message=api.roll_util.message(*rolls.values()),
        dry_run=props.dry_run,
        labels_to_set=api.roll_util.labels_to_set,
        labels_to_wait_on=api.roll_util.labels_to_wait_on,
        bot_commit=props.bot_commit,
        author_override=author_override,
    )

    return api.auto_roller.raw_result(change)


def GenTests(api):  # pylint: disable=invalid-name
    """Create tests."""

    def _url(x):
        assert ':' not in x
        return 'https://foo.googlesource.com/' + x

    def trigger(url, **kwargs):
        return api.checkout.ci_test_data(git_repo=_url(url), **kwargs)

    def properties(**kwargs):
        new_kwargs = api.checkout.git_properties()
        new_kwargs['forge_author'] = True
        new_kwargs.update(kwargs)
        return api.properties(**new_kwargs)

    def commit_data(name, **kwargs):
        return api.roll_util.commit_data(
            name,
            api.roll_util.commit('a' * 40, 'foo\nbar\n\nChange-Id: I1111'),
            **kwargs
        )

    yield (
        api.status_check.test('success')
        + properties(txt_path='foo.txt', project_remote=_url("foo"))
        + api.roll_util.properties(commit_divider='--divider--')
        + trigger('foo')
        + api.roll_util.forward_roll()
        + commit_data('foo.txt', prefix='')
        + api.auto_roller.success()
    )

    yield (
        api.status_check.test('bad-trigger', status='failure')
        + properties(txt_path='foo.txt', project_remote=_url("foo"))
        + trigger('bar')
    )

    yield (
        api.status_check.test('no-trigger')
        + properties(txt_path='foo.txt', project_remote=_url("foo"))
        + api.roll_util.forward_roll()
        + commit_data('foo.txt', prefix='')
        + api.auto_roller.success()
    )

    yield (
        api.status_check.test('backwards')
        + properties(txt_path='foo.txt', project_remote=_url("foo"))
        + api.roll_util.backward_roll()
    )