recipes/bazel_roller.py - infra/recipes - Git at Google

 # 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 Bazel WORKSPACE.

 Find a git repository pin in a WORKSPACE file like the following:

 git_repository(
     name = "pigweed",
     commit = "1111111111111111111111111111111111111111",
     remote = "https://pigweed.googlesource.com/pigweed/pigweed.git",
 )

 Specifically, find the 'remote' line that matches the repository to be rolled.
 Then check the two lines immediately before and the two lines immediately after
 the remote line for a 'commit' line, checking the adjacent lines first.
 Additionally, grab the name from these nearby lines (if not provided in a
 property).

 Then, update the 'commit' line to point to the new commit, and push a change to
 gerrit.
 """

 from __future__ import annotations

 import itertools
 import re
 from typing import Generator, Sequence, TypeVar

 import attrs
 from PB.go.chromium.org.luci.buildbucket.proto import common as common_pb2
 from PB.recipes.pigweed.bazel_roller import InputProperties
 from PB.recipe_modules.pigweed.checkout.options import (
     Options as CheckoutOptions,
 )
 from recipe_engine import post_process, recipe_api, recipe_test_api

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

 PROPERTIES = InputProperties


 def RunSteps(  # pylint: disable=invalid-name
     api: recipe_api.RecipeScriptApi,
     props: InputProperties,
 ):
     workspace_path: str = props.workspace_path or 'WORKSPACE'
     project_branch: str = 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.CheckoutContext = api.checkout(
         props.checkout_options
     )

     new_revision: Optional[str] = None

     # First, try to get new_revision from the trigger.
     bb_remote: Optional[str] = None
     commit: common_pb2.GitilesCommit = (
         api.buildbucket.build.input.gitiles_commit
     )
     if commit and commit.project:
         new_revision = commit.id
         host: str = commit.host
         bb_remote: str = 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 props.project_remote. Exception: allow a trigger to
     # be for the top-level project instead.
     use_trigger_for_project = True

     if bb_remote:
         if checkout.remotes_equivalent(
             props.checkout_options.remote, bb_remote,
         ):
             use_trigger_for_project = False

         elif not checkout.remotes_equivalent(props.project_remote, bb_remote):
             api.step.empty(
                 'triggering repository ({}) does not match project remote '
                 '({})'.format(bb_remote, props.project_remote),
                 status='FAILURE',
             )

     project_dir: config_types.Path = api.path.start_dir / 'project'

     project_checkout: api.checkout.CheckoutContext = api.checkout(
         CheckoutOptions(
             remote=props.project_remote,
             branch=project_branch,
             use_trigger=use_trigger_for_project,
         ),
         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_workspace_path = checkout.root / workspace_path

     update_result = api.bazel.update_commit_hash(
         checkout=checkout,
         project_remote=props.project_remote,
         new_revision=new_revision,
         path=full_workspace_path,
     )

     direction: api.roll_util.Direction = api.roll_util.get_roll_direction(
         project_dir, update_result.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(
             props.project_remote, update_result.old_revision, new_revision
         )
         return

     project_name = props.project_name or update_result.project_name
     if not project_name:
         api.step.empty(
             f'could not find name line in {full_workspace_path}',
             status='FAILURE',
         )

     rolls: dict[str, api.roll_util.Roll] = {
         workspace_path: api.roll_util.create_roll(
             project_name=project_name,
             old_revision=update_result.old_revision,
             new_revision=new_revision,
             proj_dir=project_dir,
             direction=direction,
         ),
     }

     authors: Sequence[api.roll_util.Account] = api.roll_util.authors(
         *rolls.values()
     )

     author_override: Optional[api.roll_util.Account] = None
     if len(authors) == 1 and props.forge_author:
         author_override = attrs.asdict(
             api.roll_util.fake_author(next(iter(authors)))
         )

     change: api.auto_roller.GerritChange = 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) -> Generator[recipe_test_api.TestData, None, None]:
     """Create tests."""

     def _url(x: str = 'pigweed/pigweed'):
         assert ':' not in x
         return f'https://pigweed.googlesource.com/{x}'

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

     def properties(**kwargs):
         props = InputProperties(**kwargs)
         props.checkout_options.CopyFrom(api.checkout.git_options())
         props.forge_author = True
         props.auto_roller_options.CopyFrom(
             api.auto_roller.Options(remote=api.checkout.pigweed_repo)
         )
         return api.properties(props)

     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.test(
         'success',
         properties(project_remote=_url('pigweed/pigweed')),
         api.roll_util.properties(commit_divider='--divider--'),
         trigger('pigweed/pigweed'),
         api.roll_util.forward_roll(),
         commit_data('pigweed', prefix=''),
         api.auto_roller.success(),
     )

     yield api.test(
         'bad-trigger',
         properties(project_remote=_url('foo')),
         trigger('bar'),
         status='FAILURE',
     )

     yield api.test(
         'name-not-found',
         properties(project_remote=_url('third/repo')),
         trigger('third/repo'),
         api.roll_util.forward_roll(),
         api.post_process(post_process.MustRunRE, r'could not find name.*'),
         api.post_process(post_process.DropExpectation),
         status='FAILURE',
     )

     yield api.test(
         'no-trigger',
         properties(project_remote=_url('pigweed/pigweed')),
         api.roll_util.forward_roll(),
         commit_data('pigweed', prefix=''),
         api.auto_roller.success(),
     )

     yield api.test(
         'backwards',
         properties(project_remote=_url('pigweed/pigweed')),
         api.roll_util.backward_roll(),
     )
	# 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 Bazel WORKSPACE.

	Find a git repository pin in a WORKSPACE file like the following:

	git_repository(
	name = "pigweed",
	commit = "1111111111111111111111111111111111111111",
	remote = "https://pigweed.googlesource.com/pigweed/pigweed.git",
	)

	Specifically, find the 'remote' line that matches the repository to be rolled.
	Then check the two lines immediately before and the two lines immediately after
	the remote line for a 'commit' line, checking the adjacent lines first.
	Additionally, grab the name from these nearby lines (if not provided in a
	property).

	Then, update the 'commit' line to point to the new commit, and push a change to
	gerrit.
	"""

	from __future__ import annotations

	import itertools
	import re
	from typing import Generator, Sequence, TypeVar

	import attrs
	from PB.go.chromium.org.luci.buildbucket.proto import common as common_pb2
	from PB.recipes.pigweed.bazel_roller import InputProperties
	from PB.recipe_modules.pigweed.checkout.options import (
	Options as CheckoutOptions,
	)
	from recipe_engine import post_process, recipe_api, recipe_test_api

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

	PROPERTIES = InputProperties


	def RunSteps( # pylint: disable=invalid-name
	api: recipe_api.RecipeScriptApi,
	props: InputProperties,
	):
	workspace_path: str = props.workspace_path or 'WORKSPACE'
	project_branch: str = 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.CheckoutContext = api.checkout(
	props.checkout_options
	)

	new_revision: Optional[str] = None

	# First, try to get new_revision from the trigger.
	bb_remote: Optional[str] = None
	commit: common_pb2.GitilesCommit = (
	api.buildbucket.build.input.gitiles_commit
	)
	if commit and commit.project:
	new_revision = commit.id
	host: str = commit.host
	bb_remote: str = 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 props.project_remote. Exception: allow a trigger to
	# be for the top-level project instead.
	use_trigger_for_project = True

	if bb_remote:
	if checkout.remotes_equivalent(
	props.checkout_options.remote, bb_remote,
	):
	use_trigger_for_project = False

	elif not checkout.remotes_equivalent(props.project_remote, bb_remote):
	api.step.empty(
	'triggering repository ({}) does not match project remote '
	'({})'.format(bb_remote, props.project_remote),
	status='FAILURE',
	)

	project_dir: config_types.Path = api.path.start_dir / 'project'

	project_checkout: api.checkout.CheckoutContext = api.checkout(
	CheckoutOptions(
	remote=props.project_remote,
	branch=project_branch,
	use_trigger=use_trigger_for_project,
	),
	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_workspace_path = checkout.root / workspace_path

	update_result = api.bazel.update_commit_hash(
	checkout=checkout,
	project_remote=props.project_remote,
	new_revision=new_revision,
	path=full_workspace_path,
	)

	direction: api.roll_util.Direction = api.roll_util.get_roll_direction(
	project_dir, update_result.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(
	props.project_remote, update_result.old_revision, new_revision
	)
	return

	project_name = props.project_name or update_result.project_name
	if not project_name:
	api.step.empty(
	f'could not find name line in {full_workspace_path}',
	status='FAILURE',
	)

	rolls: dict[str, api.roll_util.Roll] = {
	workspace_path: api.roll_util.create_roll(
	project_name=project_name,
	old_revision=update_result.old_revision,
	new_revision=new_revision,
	proj_dir=project_dir,
	direction=direction,
	),
	}

	authors: Sequence[api.roll_util.Account] = api.roll_util.authors(
	*rolls.values()
	)

	author_override: Optional[api.roll_util.Account] = None
	if len(authors) == 1 and props.forge_author:
	author_override = attrs.asdict(
	api.roll_util.fake_author(next(iter(authors)))
	)

	change: api.auto_roller.GerritChange = 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) -> Generator[recipe_test_api.TestData, None, None]:
	"""Create tests."""

	def _url(x: str = 'pigweed/pigweed'):
	assert ':' not in x
	return f'https://pigweed.googlesource.com/{x}'

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

	def properties(**kwargs):
	props = InputProperties(**kwargs)
	props.checkout_options.CopyFrom(api.checkout.git_options())
	props.forge_author = True
	props.auto_roller_options.CopyFrom(
	api.auto_roller.Options(remote=api.checkout.pigweed_repo)
	)
	return api.properties(props)

	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.test(
	'success',
	properties(project_remote=_url('pigweed/pigweed')),
	api.roll_util.properties(commit_divider='--divider--'),
	trigger('pigweed/pigweed'),
	api.roll_util.forward_roll(),
	commit_data('pigweed', prefix=''),
	api.auto_roller.success(),
	)

	yield api.test(
	'bad-trigger',
	properties(project_remote=_url('foo')),
	trigger('bar'),
	status='FAILURE',
	)

	yield api.test(
	'name-not-found',
	properties(project_remote=_url('third/repo')),
	trigger('third/repo'),
	api.roll_util.forward_roll(),
	api.post_process(post_process.MustRunRE, r'could not find name.*'),
	api.post_process(post_process.DropExpectation),
	status='FAILURE',
	)

	yield api.test(
	'no-trigger',
	properties(project_remote=_url('pigweed/pigweed')),
	api.roll_util.forward_roll(),
	commit_data('pigweed', prefix=''),
	api.auto_roller.success(),
	)

	yield api.test(
	'backwards',
	properties(project_remote=_url('pigweed/pigweed')),
	api.roll_util.backward_roll(),
	)