recipe_modules/bazel_roll/api.py - infra/recipes - Git at Google

 # Copyright 2024 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.

 from __future__ import annotations

 import configparser
 import dataclasses
 import io
 import re
 from typing import Callable, Generator, Sequence, TYPE_CHECKING

 from PB.recipe_modules.pigweed.bazel.options import Options as BazelOptions
 from recipe_engine import recipe_api

 if TYPE_CHECKING:  # pragma: no cover
     from PB.recipe_modules.pigweed.bazel_roll.git_repository import (
         GitRepository,
     )
     from recipe_engine import config_types
     from RECIPE_MODULES.fuchsia.git_roll_util import api as git_roll_util_api
     from RECIPE_MODULES.pigweed.checkout import api as checkout_api


 def nwise(iterable, n):
     # nwise('ABCDEFG', 3) → ABC BCD CDE DEF EFG
     # See also
     # https://docs.python.org/3/library/itertools.html#itertools.pairwise
     iterator = iter(iterable)
     initial_items = [None]
     for i in range(1, n):
         initial_items.append(next(iterator, None))
     items = tuple(initial_items)
     for x in iterator:
         items = (*items[1:], x)
         yield items


 def trim_nearby_lines(lines: Sequence[LineProxy]) -> list[LineProxy]:
     # If there's a blank line in the second half, remove any following lines.
     for i in range(len(lines) // 2, len(lines)):
         text = lines[i].text or ''
         if not text.strip():
             lines = lines[0:i]
             break

     # If there's a blank line in the first half, remove any preceding lines.
     for i in range(len(lines) // 2, 0, -1):
         text = lines[i].text or ''
         if not text.strip():
             lines = lines[i:]
             break

     return lines


 def proximity_sort_nearby_lines(
     lines: Sequence[LineProxy],
     center: int,
 ) -> list[LineProxy]:
     # Shift the order to be center-out instead of ascending.
     lines = [(abs(center - x.idx), x.idx, x) for x in lines]
     return [x[-1] for x in sorted(lines)]


 @dataclasses.dataclass
 class UpdateResult:
     old_value: str
     project_name: str | None
     finalize: Callable[[], None] | None = None


 GIT_HASH_PATTERN = '[0-9a-fA-F]{40}'


 class LineProxy:
     def __init__(self, lines, idx):
         self._lines = lines
         self._idx = idx

     @property
     def idx(self):
         return self._idx

     @property
     def text(self):
         return self._lines[self._idx]

     @text.setter
     def text(self, value):
         self._lines[self._idx] = value

     def __repr__(self):
         return f'{self.idx} {self.text}'


 def proxy(lines):
     return [LineProxy(lines, i) for i in range(len(lines))]


 class BazelRollApi(recipe_api.RecipeApi):

     UpdateResult = UpdateResult
     GIT_HASH_PATTERN = GIT_HASH_PATTERN

     def workspace_path(
         self,
         root: config_types.Path,
         value: str,
     ) -> config_types.Path:
         """Figure out the location of the WORKSPACE or MODULE.bazel file.

         If value is '', look for root / 'WORKSPACE' and root / 'MODULE.bazel'.
         If exactly one exists, return it. If not, error out.

         If root / value is a file, return it.

         If root / value is a directory, set root = root / value and apply the
         above logic. This enables applying this logic to subdirectories.

         Args:
             api: Recipe API object.
             root: Checkout root.
             value: Relative path specified in properties.

         Returns:
             Path to the WORKSPACE or MODULE.bazel file.
         """
         if value:
             value_path = root / value
             self.m.path.mock_add_file(value_path)
             if self.m.path.isfile(value_path):
                 return value_path
             elif self.m.path.isdir(value_path):  # pragma: no cover
                 root = value_path
             else:
                 self.m.step.empty(  # pragma: no cover
                     f'{value_path} does not exist',
                     status='FAILURE',
                 )

         workspace = root / 'WORKSPACE'
         module_bazel = root / 'MODULE.bazel'

         self.m.path.mock_add_file(workspace)

         if self.m.path.isfile(module_bazel) and self.m.path.isfile(workspace):
             self.m.step.empty(  # pragma: no cover
                 f'{module_bazel} and {workspace} both exist',
                 status='FAILURE',
             )

         if self.m.path.isfile(module_bazel):
             return module_bazel  # pragma: no cover

         if self.m.path.isfile(workspace):
             return workspace

         self.m.step.empty(  # pragma: no cover
             'no WORKSPACE or MODULE.bazel file found',
             status='FAILURE',
         )

     def update_git_repository(
         self,
         checkout: checkout_self.m.CheckoutContext,
         git_repository: GitRepository,
     ) -> list[git_roll_util_api.Roll]:
         project_name = git_repository.name
         if not project_name:
             project_name = git_repository.remote.split('/')[-1]
             project_name = project_name.removesuffix('.git')

         with self.m.step.nest(project_name):
             workspace_path = self.workspace_path(
                 checkout.root,
                 git_repository.workspace_path,
             )
             git_repository.branch = git_repository.branch or 'main'

             new_revision = self.m.git_roll_util.resolve_new_revision(
                 git_repository.remote,
                 git_repository.branch,
                 checkout.remotes_equivalent,
             )

             update_result = self.update_value(
                 project_value=git_repository.remote,
                 new_value=new_revision,
                 path=workspace_path,
                 delay_write=True,
                 search_key='remote',
                 value_key='commit',
                 value_pattern=GIT_HASH_PATTERN,
                 matches=checkout.remotes_equivalent,
             )
             if not update_result:
                 self.m.step.empty(
                     'failed to update commit hash',
                     status='FAILURE',
                 )

             try:
                 roll = self.m.git_roll_util.get_roll(
                     repo_url=git_repository.remote,
                     repo_short_name=project_name,
                     old_rev=update_result.old_value,
                     new_rev=new_revision,
                 )

             except self.m.git_roll_util.BackwardsRollError:
                 return []

             update_result.finalize()

             with self.m.step.nest('update MODULE.bazel.lock'):
                 runner = self.m.bazel.new_runner(
                     checkout=checkout,
                     options=BazelOptions(),
                 )

                 with self.m.context(cwd=checkout.root):
                     self.m.step(
                         'bazelisk mod deps --lockfile_mode=update',
                         [
                             runner.ensure(),
                             'mod',
                             'deps',
                             '--lockfile_mode=update',
                         ],
                     )

             return [roll]

     def _read(self, path: config_types.Path, num_nearby_lines: int):
         lines = [''] * num_nearby_lines
         lines.extend(
             self.m.file.read_text(
                 f'read old {path.name}',
                 path,
                 test_data=self.m.bazel_roll.test_api.TEST_WORKSPACE_FILE,
             )
             .strip()
             .splitlines()
         )
         lines.extend([''] * num_nearby_lines)
         return lines

     def _get_matching_groups(
         self,
         lines: Sequence[str],
         num_nearby_lines: int,
         project_value: str,
         search_key: str,
         replace: bool = False,
         matches: Callable[[str, str], bool] | None = None,
     ) -> list[tuple[LineProxy, list[LineProxy]]]:
         matching_groups: list[tuple[LineProxy, list[LineProxy]]] = []

         # This is more constraining than necessary, but ensures there aren't
         # special characters that confuse the regex.
         assert search_key.isidentifier(), search_key

         for nearby_lines in nwise(proxy(lines), num_nearby_lines * 2 + 1):
             curr = nearby_lines[len(nearby_lines) // 2]
             match = re.search(
                 rf'^\s*{search_key}\s*=\s*"(?P<search>[^"]+)",?\s*$',
                 curr.text,
             )
             if not match:
                 continue

             match_value = match.group('search')

             if not matches:
                 # TODO: b/376915917 - Remove no cover pragma.
                 matches = lambda x, y: x == y  # pragma: no cover

             if matches(match_value, project_value):
                 pres = self.m.step.empty(
                     f'found matching {search_key} {match_value!r}'
                 ).presentation
                 pres.logs['lines'] = [repr(x) for x in nearby_lines]
                 matching_groups.append((curr, nearby_lines))

                 if replace and match_value != project_value:
                     curr.text = curr.text.replace(match_value, project_value)

             else:
                 pres = self.m.step.empty(
                     f'found other {search_key} {match_value!r}'
                 ).presentation
                 pres.logs['lines'] = [repr(x) for x in nearby_lines]

         return matching_groups

     def _process_nearby_lines(self, nearby_lines):
         pres = self.m.step.empty('lines').presentation
         center = nearby_lines[len(nearby_lines) // 2].idx
         pres.logs['0_center'] = [str(center)]

         pres.logs['1_orig'] = [repr(x) for x in nearby_lines]

         nearby_lines = trim_nearby_lines(nearby_lines)
         pres.logs['2_trimmed'] = [repr(x) for x in nearby_lines]

         nearby_lines = proximity_sort_nearby_lines(nearby_lines, center)
         pres.logs['3_sorted'] = [repr(x) for x in nearby_lines]

         return nearby_lines

     def retrieve_attributes(
         self,
         project_value: str,
         path: config_types.Path,
         num_nearby_lines: int = 10,
         title_keys: str | Sequence[str] = ('name', 'module_name'),
         matches: Callable[[str, str], bool] | None = None,
     ) -> list[dict[str, str]]:
         lines = self._read(path, num_nearby_lines)

         assert isinstance(title_keys, (list, tuple)), repr(title_keys)

         matching_groups = self._get_matching_groups(
             lines=lines,
             num_nearby_lines=num_nearby_lines,
             project_value=project_value,
             search_key='remote',
             matches=matches,
         )

         results: list[dict[str, str]] = []
         for _, nearby_lines in matching_groups:
             entry = {}
             nearby_lines = self._process_nearby_lines(nearby_lines)

             for line in nearby_lines:
                 if match := re.search(
                     r'^\s*([\w_]+)\s*=\s*(\S.*),\s*$',
                     line.text,
                 ):
                     entry[match.group(1)] = match.group(2).strip('"')
                     if match.group(1) in title_keys:
                         entry['name'] = match.group(2).strip('"')

             if entry:
                 results.append(entry)

         return results

     def update_value(
         self,
         *,
         project_value: str,
         new_value: str,
         path: config_types.Path,
         search_key: str,
         value_key: str,
         num_nearby_lines: int = 10,
         replace: bool = False,
         delay_write: bool = False,
         title_keys: Sequence[str] = ('name', 'module_name'),
         value_pattern: str | None = None,
         matches: Callable[[str, str], bool] | None = None,
     ) -> UpdateResult | None:
         lines = self._read(path, num_nearby_lines)

         matching_groups = self._get_matching_groups(
             lines=lines,
             num_nearby_lines=num_nearby_lines,
             project_value=project_value,
             replace=replace,
             search_key=search_key,
             matches=matches,
         )

         if not matching_groups:
             self.m.step.empty(
                 f'could not find {search_key} {project_value} in {path}',
             )
             return None

         project_names: list[str] = []
         if value_pattern is None:
             # TODO: b/376915917 - Remove no cover pragma.
             value_pattern = '[^"]+'  # pragma: no cover

         for matching_line, nearby_lines in matching_groups:
             nearby_lines = self._process_nearby_lines(nearby_lines)

             value_rx = re.compile(
                 rf'^(?P<prefix>\s*{value_key}\s*=\s*")'
                 rf'(?P<value>{value_pattern})'
                 r'(?P<suffix>",?\s*)$'
             )

             for line in nearby_lines:
                 if match := value_rx.search(line.text):
                     idx = line.idx
                     break
             else:
                 pres = self.m.step.empty(
                     f'could not find {value_key} line adjacent to '
                     f'{matching_line.text!r} in {path}',
                 ).presentation
                 pres.step_summary_text = repr(nearby_lines)
                 return None

             old_value = match.group('value')

             prefix = match.group("prefix")
             suffix = match.group("suffix")
             lines[idx] = f'{prefix}{new_value}{suffix}'

             # Remove all existing metadata lines in this git_repository() entry.
             idx2 = idx - 1
             while lines[idx2].strip().startswith('# ROLL: '):
                 lines[idx2] = None
                 idx2 -= 1

             ws_prefix = re.search(r'^\s*', prefix).group(0)
             comment_prefix = f'{ws_prefix}# ROLL: '

             now = self.m.time.utcnow().strftime('%Y-%m-%d')
             comment_lines = (
                 f'{comment_prefix}Warning: this entry is automatically '
                 'updated.',
                 f'{comment_prefix}Last updated {now}.',
                 f'{comment_prefix}By {self.m.buildbucket.build_url()}.',
             )

             lines[idx] = '\n'.join(comment_lines + (lines[idx],))

             assert isinstance(title_keys, (list, tuple)), repr(title_keys)

             # This is more restrictive than necessary but helps keep the regex
             # simple.
             for key in title_keys:
                 assert isinstance(key, str), repr(key)
                 assert key.isidentifier(), repr(key)

             for line in nearby_lines:
                 keys = "|".join(title_keys)
                 if match := re.search(
                     rf'^\s*(?:{keys})\s*=\s*"(?P<value>[^"]+)",?\s*$',
                     line.text or '',
                 ):
                     project_names.append(match.group('value'))
                     break

         def write():
             self.m.file.write_text(
                 f'write new {path.name}',
                 path,
                 ''.join(
                     f'{x}\n'
                     for x in lines[num_nearby_lines:-num_nearby_lines]
                     if x is not None
                 ),
             )

         if delay_write:
             return UpdateResult(
                 old_value=old_value,
                 project_name=', '.join(project_names),
                 finalize=write,
             )

         else:
             write()

             return UpdateResult(
                 old_value=old_value,
                 project_name=', '.join(project_names),
             )
	# Copyright 2024 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.

	from __future__ import annotations

	import configparser
	import dataclasses
	import io
	import re
	from typing import Callable, Generator, Sequence, TYPE_CHECKING

	from PB.recipe_modules.pigweed.bazel.options import Options as BazelOptions
	from recipe_engine import recipe_api

	if TYPE_CHECKING: # pragma: no cover
	from PB.recipe_modules.pigweed.bazel_roll.git_repository import (
	GitRepository,
	)
	from recipe_engine import config_types
	from RECIPE_MODULES.fuchsia.git_roll_util import api as git_roll_util_api
	from RECIPE_MODULES.pigweed.checkout import api as checkout_api


	def nwise(iterable, n):
	# nwise('ABCDEFG', 3) → ABC BCD CDE DEF EFG
	# See also
	# https://docs.python.org/3/library/itertools.html#itertools.pairwise
	iterator = iter(iterable)
	initial_items = [None]
	for i in range(1, n):
	initial_items.append(next(iterator, None))
	items = tuple(initial_items)
	for x in iterator:
	items = (*items[1:], x)
	yield items


	def trim_nearby_lines(lines: Sequence[LineProxy]) -> list[LineProxy]:
	# If there's a blank line in the second half, remove any following lines.
	for i in range(len(lines) // 2, len(lines)):
	text = lines[i].text or ''
	if not text.strip():
	lines = lines[0:i]
	break

	# If there's a blank line in the first half, remove any preceding lines.
	for i in range(len(lines) // 2, 0, -1):
	text = lines[i].text or ''
	if not text.strip():
	lines = lines[i:]
	break

	return lines


	def proximity_sort_nearby_lines(
	lines: Sequence[LineProxy],
	center: int,
	) -> list[LineProxy]:
	# Shift the order to be center-out instead of ascending.
	lines = [(abs(center - x.idx), x.idx, x) for x in lines]
	return [x[-1] for x in sorted(lines)]


	@dataclasses.dataclass
	class UpdateResult:
	old_value: str
	project_name: str \| None
	finalize: Callable[[], None] \| None = None


	GIT_HASH_PATTERN = '[0-9a-fA-F]{40}'


	class LineProxy:
	def __init__(self, lines, idx):
	self._lines = lines
	self._idx = idx

	@property
	def idx(self):
	return self._idx

	@property
	def text(self):
	return self._lines[self._idx]

	@text.setter
	def text(self, value):
	self._lines[self._idx] = value

	def __repr__(self):
	return f'{self.idx} {self.text}'


	def proxy(lines):
	return [LineProxy(lines, i) for i in range(len(lines))]


	class BazelRollApi(recipe_api.RecipeApi):

	UpdateResult = UpdateResult
	GIT_HASH_PATTERN = GIT_HASH_PATTERN

	def workspace_path(
	self,
	root: config_types.Path,
	value: str,
	) -> config_types.Path:
	"""Figure out the location of the WORKSPACE or MODULE.bazel file.

	If value is '', look for root / 'WORKSPACE' and root / 'MODULE.bazel'.
	If exactly one exists, return it. If not, error out.

	If root / value is a file, return it.

	If root / value is a directory, set root = root / value and apply the
	above logic. This enables applying this logic to subdirectories.

	Args:
	api: Recipe API object.
	root: Checkout root.
	value: Relative path specified in properties.

	Returns:
	Path to the WORKSPACE or MODULE.bazel file.
	"""
	if value:
	value_path = root / value
	self.m.path.mock_add_file(value_path)
	if self.m.path.isfile(value_path):
	return value_path
	elif self.m.path.isdir(value_path): # pragma: no cover
	root = value_path
	else:
	self.m.step.empty( # pragma: no cover
	f'{value_path} does not exist',
	status='FAILURE',
	)

	workspace = root / 'WORKSPACE'
	module_bazel = root / 'MODULE.bazel'

	self.m.path.mock_add_file(workspace)

	if self.m.path.isfile(module_bazel) and self.m.path.isfile(workspace):
	self.m.step.empty( # pragma: no cover
	f'{module_bazel} and {workspace} both exist',
	status='FAILURE',
	)

	if self.m.path.isfile(module_bazel):
	return module_bazel # pragma: no cover

	if self.m.path.isfile(workspace):
	return workspace

	self.m.step.empty( # pragma: no cover
	'no WORKSPACE or MODULE.bazel file found',
	status='FAILURE',
	)

	def update_git_repository(
	self,
	checkout: checkout_self.m.CheckoutContext,
	git_repository: GitRepository,
	) -> list[git_roll_util_api.Roll]:
	project_name = git_repository.name
	if not project_name:
	project_name = git_repository.remote.split('/')[-1]
	project_name = project_name.removesuffix('.git')

	with self.m.step.nest(project_name):
	workspace_path = self.workspace_path(
	checkout.root,
	git_repository.workspace_path,
	)
	git_repository.branch = git_repository.branch or 'main'

	new_revision = self.m.git_roll_util.resolve_new_revision(
	git_repository.remote,
	git_repository.branch,
	checkout.remotes_equivalent,
	)

	update_result = self.update_value(
	project_value=git_repository.remote,
	new_value=new_revision,
	path=workspace_path,
	delay_write=True,
	search_key='remote',
	value_key='commit',
	value_pattern=GIT_HASH_PATTERN,
	matches=checkout.remotes_equivalent,
	)
	if not update_result:
	self.m.step.empty(
	'failed to update commit hash',
	status='FAILURE',
	)

	try:
	roll = self.m.git_roll_util.get_roll(
	repo_url=git_repository.remote,
	repo_short_name=project_name,
	old_rev=update_result.old_value,
	new_rev=new_revision,
	)

	except self.m.git_roll_util.BackwardsRollError:
	return []

	update_result.finalize()

	with self.m.step.nest('update MODULE.bazel.lock'):
	runner = self.m.bazel.new_runner(
	checkout=checkout,
	options=BazelOptions(),
	)

	with self.m.context(cwd=checkout.root):
	self.m.step(
	'bazelisk mod deps --lockfile_mode=update',
	[
	runner.ensure(),
	'mod',
	'deps',
	'--lockfile_mode=update',
	],
	)

	return [roll]

	def _read(self, path: config_types.Path, num_nearby_lines: int):
	lines = [''] * num_nearby_lines
	lines.extend(
	self.m.file.read_text(
	f'read old {path.name}',
	path,
	test_data=self.m.bazel_roll.test_api.TEST_WORKSPACE_FILE,
	)
	.strip()
	.splitlines()
	)
	lines.extend([''] * num_nearby_lines)
	return lines

	def _get_matching_groups(
	self,
	lines: Sequence[str],
	num_nearby_lines: int,
	project_value: str,
	search_key: str,
	replace: bool = False,
	matches: Callable[[str, str], bool] \| None = None,
	) -> list[tuple[LineProxy, list[LineProxy]]]:
	matching_groups: list[tuple[LineProxy, list[LineProxy]]] = []

	# This is more constraining than necessary, but ensures there aren't
	# special characters that confuse the regex.
	assert search_key.isidentifier(), search_key

	for nearby_lines in nwise(proxy(lines), num_nearby_lines * 2 + 1):
	curr = nearby_lines[len(nearby_lines) // 2]
	match = re.search(
	rf'^\s{search_key}\s=\s"(?P<search>[^"]+)",?\s$',
	curr.text,
	)
	if not match:
	continue

	match_value = match.group('search')

	if not matches:
	# TODO: b/376915917 - Remove no cover pragma.
	matches = lambda x, y: x == y # pragma: no cover

	if matches(match_value, project_value):
	pres = self.m.step.empty(
	f'found matching {search_key} {match_value!r}'
	).presentation
	pres.logs['lines'] = [repr(x) for x in nearby_lines]
	matching_groups.append((curr, nearby_lines))

	if replace and match_value != project_value:
	curr.text = curr.text.replace(match_value, project_value)

	else:
	pres = self.m.step.empty(
	f'found other {search_key} {match_value!r}'
	).presentation
	pres.logs['lines'] = [repr(x) for x in nearby_lines]

	return matching_groups

	def _process_nearby_lines(self, nearby_lines):
	pres = self.m.step.empty('lines').presentation
	center = nearby_lines[len(nearby_lines) // 2].idx
	pres.logs['0_center'] = [str(center)]

	pres.logs['1_orig'] = [repr(x) for x in nearby_lines]

	nearby_lines = trim_nearby_lines(nearby_lines)
	pres.logs['2_trimmed'] = [repr(x) for x in nearby_lines]

	nearby_lines = proximity_sort_nearby_lines(nearby_lines, center)
	pres.logs['3_sorted'] = [repr(x) for x in nearby_lines]

	return nearby_lines

	def retrieve_attributes(
	self,
	project_value: str,
	path: config_types.Path,
	num_nearby_lines: int = 10,
	title_keys: str \| Sequence[str] = ('name', 'module_name'),
	matches: Callable[[str, str], bool] \| None = None,
	) -> list[dict[str, str]]:
	lines = self._read(path, num_nearby_lines)

	assert isinstance(title_keys, (list, tuple)), repr(title_keys)

	matching_groups = self._get_matching_groups(
	lines=lines,
	num_nearby_lines=num_nearby_lines,
	project_value=project_value,
	search_key='remote',
	matches=matches,
	)

	results: list[dict[str, str]] = []
	for _, nearby_lines in matching_groups:
	entry = {}
	nearby_lines = self._process_nearby_lines(nearby_lines)

	for line in nearby_lines:
	if match := re.search(
	r'^\s([\w_]+)\s=\s(\S.),\s*$',
	line.text,
	):
	entry[match.group(1)] = match.group(2).strip('"')
	if match.group(1) in title_keys:
	entry['name'] = match.group(2).strip('"')

	if entry:
	results.append(entry)

	return results

	def update_value(
	self,
	*,
	project_value: str,
	new_value: str,
	path: config_types.Path,
	search_key: str,
	value_key: str,
	num_nearby_lines: int = 10,
	replace: bool = False,
	delay_write: bool = False,
	title_keys: Sequence[str] = ('name', 'module_name'),
	value_pattern: str \| None = None,
	matches: Callable[[str, str], bool] \| None = None,
	) -> UpdateResult \| None:
	lines = self._read(path, num_nearby_lines)

	matching_groups = self._get_matching_groups(
	lines=lines,
	num_nearby_lines=num_nearby_lines,
	project_value=project_value,
	replace=replace,
	search_key=search_key,
	matches=matches,
	)

	if not matching_groups:
	self.m.step.empty(
	f'could not find {search_key} {project_value} in {path}',
	)
	return None

	project_names: list[str] = []
	if value_pattern is None:
	# TODO: b/376915917 - Remove no cover pragma.
	value_pattern = '[^"]+' # pragma: no cover

	for matching_line, nearby_lines in matching_groups:
	nearby_lines = self._process_nearby_lines(nearby_lines)

	value_rx = re.compile(
	rf'^(?P<prefix>\s{value_key}\s=\s*")'
	rf'(?P<value>{value_pattern})'
	r'(?P<suffix>",?\s*)$'
	)

	for line in nearby_lines:
	if match := value_rx.search(line.text):
	idx = line.idx
	break
	else:
	pres = self.m.step.empty(
	f'could not find {value_key} line adjacent to '
	f'{matching_line.text!r} in {path}',
	).presentation
	pres.step_summary_text = repr(nearby_lines)
	return None

	old_value = match.group('value')

	prefix = match.group("prefix")
	suffix = match.group("suffix")
	lines[idx] = f'{prefix}{new_value}{suffix}'

	# Remove all existing metadata lines in this git_repository() entry.
	idx2 = idx - 1
	while lines[idx2].strip().startswith('# ROLL: '):
	lines[idx2] = None
	idx2 -= 1

	ws_prefix = re.search(r'^\s*', prefix).group(0)
	comment_prefix = f'{ws_prefix}# ROLL: '

	now = self.m.time.utcnow().strftime('%Y-%m-%d')
	comment_lines = (
	f'{comment_prefix}Warning: this entry is automatically '
	'updated.',
	f'{comment_prefix}Last updated {now}.',
	f'{comment_prefix}By {self.m.buildbucket.build_url()}.',
	)

	lines[idx] = '\n'.join(comment_lines + (lines[idx],))

	assert isinstance(title_keys, (list, tuple)), repr(title_keys)

	# This is more restrictive than necessary but helps keep the regex
	# simple.
	for key in title_keys:
	assert isinstance(key, str), repr(key)
	assert key.isidentifier(), repr(key)

	for line in nearby_lines:
	keys = "\|".join(title_keys)
	if match := re.search(
	rf'^\s(?:{keys})\s=\s"(?P<value>[^"]+)",?\s$',
	line.text or '',
	):
	project_names.append(match.group('value'))
	break

	def write():
	self.m.file.write_text(
	f'write new {path.name}',
	path,
	''.join(
	f'{x}\n'
	for x in lines[num_nearby_lines:-num_nearby_lines]
	if x is not None
	),
	)

	if delay_write:
	return UpdateResult(
	old_value=old_value,
	project_name=', '.join(project_names),
	finalize=write,
	)

	else:
	write()

	return UpdateResult(
	old_value=old_value,
	project_name=', '.join(project_names),
	)