| # 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() |
| ) |