recipes/repo_roller.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.
"""Roll a project in an Android Repo Tool workspace."""

from __future__ import annotations

import collections
import dataclasses
import re
from typing import TYPE_CHECKING
import urllib
import xml.etree.ElementTree

import attrs
from PB.recipes.pigweed.repo_roller import InputProperties
from PB.recipe_modules.pigweed.repo_roll.project import Project
from PB.recipe_modules.pigweed.checkout.options import (
    Options as CheckoutOptions,
)

if TYPE_CHECKING:  # pragma: no cover
    from typing import Generator
    from recipe_engine import config_types, recipe_test_api
    from RECIPE_MODULES.pigweed.roll_util import api as roll_util_api

DEPS = [
    'fuchsia/auto_roller',
    'pigweed/checkout',
    'pigweed/repo_roll',
    'pigweed/roll_util',
    'recipe_engine/properties',
    'recipe_engine/step',
]

PROPERTIES = InputProperties


def RunSteps(api, props):  # pylint: disable=invalid-name
    path_to_update = str(props.path_to_update)
    cc_authors_on_rolls = props.cc_authors_on_rolls
    cc_reviewers_on_rolls = props.cc_reviewers_on_rolls
    cc_domains = props.cc_domains
    always_cc = props.always_cc

    props.checkout_options.use_trigger = False
    checkout = api.checkout(props.checkout_options)

    projects = list(props.projects)
    if not projects:
        projects.append(Project(path_to_update=path_to_update))
    assert len(projects) == 1

    rolls = []
    for project in projects:
        rolls.extend(api.repo_roll.update_project(checkout, project))

    if not rolls:
        return

    assert len(rolls) == 1
    roll = rolls[0]

    authors = api.roll_util.authors(roll)

    max_commits_for_ccing = props.max_commits_for_ccing or 10
    if len(roll.commits) <= max_commits_for_ccing:
        cc = set()
        if cc_authors_on_rolls:
            cc.update(authors)
        if cc_reviewers_on_rolls:
            cc.update(api.roll_util.reviewers(roll))

        def include_cc(email):
            return api.roll_util.include_cc(
                email, cc_domains, checkout.gerrit_host()
            )

        # include_cc() writes steps, so we want things sorted before calling it.
        cc = sorted(set(cc))
        cc_emails = [x.email for x in cc if include_cc(x)]

        if always_cc:
            props.auto_roller_options.cc_emails.extend(cc_emails)
        else:
            props.auto_roller_options.cc_on_failure_emails.extend(cc_emails)

    author_override = None
    with api.step.nest('authors') as pres:
        pres.step_summary_text = repr(authors)
    if len(authors) == 1 and props.forge_author:
        author_override = attrs.asdict(
            api.roll_util.fake_author(next(iter(authors)))
        )

    # merge auto_roller_options and override_auto_roller_options.
    complete_auto_roller_options = api.roll_util.merge_auto_roller_overrides(
        props.auto_roller_options, props.override_auto_roller_options
    )

    change = api.auto_roller.attempt_roll(
        complete_auto_roller_options,
        repo_dir=checkout.root,
        commit_message=api.roll_util.message(roll),
        author_override=author_override,
    )

    return api.auto_roller.raw_result(change)


def GenTests(api) -> Generator[recipe_test_api.TestData, None, None]:
    """Create tests."""

    manifest = 'https://host.googlesource.com/manifest'

    def properties(api, path, dry_run=True, equivalent_remotes=(), **kwargs):
        props = InputProperties(**kwargs)
        props.checkout_options.CopyFrom(
            api.checkout.git_options(
                remote=manifest,
                equivalent_remotes=equivalent_remotes,
                match_branch=True,
            )
        )
        props.forge_author = True
        props.path_to_update = path
        props.auto_roller_options.CopyFrom(
            api.auto_roller.Options(dry_run=dry_run, remote=manifest)
        )

        return api.properties(props)

    def commit_data(name=None):
        return api.roll_util.commit_data(
            name, api.roll_util.commit('a' * 40, 'foo\nbar')
        )

    yield api.test(
        'success',
        properties(api, path='a1', cc_authors_on_rolls=True, always_cc=True),
        api.checkout.ci_test_data(git_repo='https://foo.googlesource.com/a'),
        commit_data('a1'),
        api.repo_roll.read_step_data(),
        api.roll_util.forward_roll(),
        api.auto_roller.dry_run_success(),
    )

    yield api.test(
        'equivalent',
        properties(
            api,
            path='b2',
            equivalent_remotes=(
                (
                    'https://equiv.googlesource.com/b',
                    'https://bar.googlesource.com/b',
                ),
            ),
            cc_reviewers_on_rolls=True,
        ),
        api.checkout.ci_test_data(git_repo='https://equiv.googlesource.com/b'),
        commit_data('b2'),
        api.repo_roll.read_step_data(),
        api.roll_util.forward_roll(),
        api.auto_roller.dry_run_success(),
    )

    yield api.test(
        'backwards',
        properties(api, path='a1'),
        api.checkout.ci_test_data(git_repo='https://foo.googlesource.com/a'),
        api.repo_roll.read_step_data(),
        api.roll_util.backward_roll(),
    )