blob: 4f17a7995f8e92f60e8dc09142d67925a6189587 [file] [log] [blame]
# 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 configparser
import re
import urllib
import attr
from PB.recipes.pigweed.txt_roller import InputProperties
from PB.recipe_modules.pigweed.checkout.options import (
Options as CheckoutOptions,
)
from recipe_engine import post_process
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
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.
props.checkout_options.use_trigger = False
checkout = api.checkout(props.checkout_options)
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 = f'https://{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 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')
project_checkout = api.checkout(
CheckoutOptions(
remote=project_remote, branch=project_branch, use_trigger=True,
),
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 = 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, f'{new_revision}\n'
)
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(
props.auto_roller_options,
repo_dir=checkout.root,
commit_message=api.roll_util.message(*rolls.values()),
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.setdefault(
'auto_roller_options', {'remote': api.checkout.pigweed_repo,},
)
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()
)