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

 # Copyright 2026 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.
 """Recipe module for rolling properties extracted from CI builds.

 This module provides the `PropertyRollApi` class, which facilitates updating
 a local JSON file by extracting output properties from a passing CI build.
 Key functionalities include:
   - Resolving the latest passing build of a specified builder.
   - Extracting a subset of output properties from that build.
   - Comparing the extracted properties with the local destination file.
   - Updating the local JSON file if its content differs from the extracted
     properties.
   - Generating `PropertyRoll` objects that encapsulate the changes for commit
     message formatting.
 """

 from __future__ import annotations

 import json
 from collections.abc import Sequence

 from google.protobuf import json_format

 from recipe_engine import recipe_api

 from PB.go.chromium.org.luci.buildbucket.proto import common as common_pb2
 from PB.recipe_modules.pigweed.property_roll.property_entry import PropertyEntry

 from RECIPE_MODULES.fuchsia.roll_commit_message import (
     api as roll_commit_message_api,
 )
 from RECIPE_MODULES.pigweed.checkout import api as checkout_api


 class PropertyRoll(roll_commit_message_api.BaseRoll):
     """Represents a roll of properties extracted from a CI build.

     This class holds information about a property roll operation, including
     the source build revision, and the old and new JSON content of the file.
     It's used to generate commit messages for such rolls.
     """

     def __init__(
         self,
         destination_path: str,
         old_value: str | None,
         new_value: str,
         revision: str,
         *args,
         **kwargs,
     ):
         """Initializes a PropertyRoll instance.

         Args:
             destination_path: The path to the destination file in the local
                 checkout.
             old_value: The previous content of the destination file.
             new_value: The new content extracted from the build.
             revision: The commit hash or ID corresponding to the source build.
             *args: Additional arguments for the base class.
             **kwargs: Additional keyword arguments for the base class.
         """
         super().__init__(*args, **kwargs)
         self.destination_path = destination_path
         self.old_value = old_value
         self.new_value = new_value
         self.revision = revision

     def short_name(self) -> str:
         """Returns a short name for the roll, typically the destination path."""
         return self.destination_path

     def message_header(self, force_summary_version: bool = False) -> str:
         """Generates the header for the commit message.

         Args:
             force_summary_version: If True, a more concise header is generated.

         Returns:
             The commit message header string.
         """
         header = self.destination_path
         if force_summary_version:
             return header

         try:
             value_string = self.new_value.strip()
             if '\n' not in value_string:
                 new_header = f'{header}: {value_string}'
                 if len(new_header) <= 62:
                     header = new_header

         except UnicodeDecodeError:  # pragma: no cover
             pass

         return header

     def message_body(
         self,
         *,
         force_summary_version: bool = False,
         escape_tags: Sequence[str] = (),
         filter_tags: Sequence[str] = (),
     ) -> str:
         """Generates the body of the commit message.

         Args:
             force_summary_version: If True, a more concise body is generated.
             escape_tags: A sequence of tags to escape in the message.
             filter_tags: A sequence of tags to filter from the message.

         Returns:
             The commit message body string.
         """
         del force_summary_version, escape_tags, filter_tags  # Unused.
         result = []
         if (
             '\n' not in str(self.old_value).strip()
             and '\n' not in self.new_value.strip()
         ):
             if self.old_value is None:
                 result.append('Initialized to')  # pragma: no cover
             else:
                 result.append('From')
                 old_value = self.old_value
                 if len(old_value.strip()) > 50:
                     old_value = f'{len(old_value)} chars'
                 result.append(f'    {old_value.strip()}')
                 result.append('To')
             new_value = self.new_value
             if len(new_value.strip()) > 50:
                 new_value = f'{len(new_value)} chars'
             result.append(f'    {new_value.strip()}')
             result.append('')

         return '\n'.join(result)

     def message_footer(self, *, send_comment: bool) -> str:
         """Generates the footer for the commit message.

         Args:
             send_comment: If True, indicates a comment should be sent.

         Returns:
             An empty string, as property rolls typically don't have footers.
         """
         del send_comment  # Unused.
         return ''

     def output_property(self) -> dict[str, str]:
         """Generates a dictionary of properties representing the roll.

         Returns:
             A dictionary containing details of the roll, such as path,
             old/new values, and revision.
         """
         return {
             'path': self.destination_path,
             'old': str(self.old_value),
             'new': self.new_value,
             'revision': self.revision,
         }


 class PropertyRollApi(recipe_api.RecipeApi):
     """Provides methods to roll properties extracted from CI builds."""

     PropertyRoll = PropertyRoll

     def update(
         self,
         checkout: checkout_api.CheckoutContext,
         property_entry: PropertyEntry,
     ) -> list[PropertyRoll]:
         """Updates a local JSON file by extracting output properties from a build.

         Args:
             checkout: The checkout context, providing repository information.
             property_entry: A protobuf message defining the property extraction,
                 including project, bucket, builder, properties to extract, and
                 destination path.

         Returns:
             A list containing a `PropertyRoll` object if the file was updated,
             or an empty list if the content was identical.
         """
         with self.m.step.nest(property_entry.destination_path):
             build = self.m.buildbucket_util.last_build(
                 project=property_entry.project or None,
                 bucket=property_entry.bucket,
                 builder=property_entry.builder,
                 status=common_pb2.SUCCESS,
                 fields=['id', 'input', 'output.properties'],
             )
             if not build:
                 raise self.m.step.StepFailure(
                     f'no passing build found for {property_entry.builder}'
                 )

             properties_dict = json_format.MessageToDict(build.output.properties)
             if property_entry.properties:
                 extracted = {}
                 for key in property_entry.properties:
                     if key not in properties_dict:
                         raise self.m.step.StepFailure(
                             f'property {key!r} missing in output properties of build {build.id}'
                         )
                     extracted[key] = properties_dict[key]
             else:
                 extracted = properties_dict

             new_value = json.dumps(extracted, indent=2, sort_keys=True) + '\n'

             destination = checkout.root / property_entry.destination_path
             self.m.file.ensure_directory(
                 f'ensure directory {destination.parent}',
                 destination.parent,
             )
             old_value: str | None = None
             self.m.path.mock_add_file(destination)
             if self.m.path.isfile(destination):
                 old_value = self.m.file.read_text(
                     f'read destination {property_entry.destination_path}',
                     destination,
                 )

             new = self.m.step.empty(f'new {property_entry.destination_path}')
             new.presentation.step_summary_text = repr(new_value)

             revision = str(
                 properties_dict.get('got_revision')
                 or build.input.gitiles_commit.id
                 or build.id
             )

             if old_value == new_value:
                 pres = self.m.step.empty('values are identical').presentation
                 pres.step_summary_text = repr(old_value)

                 pres.properties[property_entry.destination_path] = {
                     'builder': property_entry.builder,
                     'revision': revision,
                 }

                 return []

             self.m.file.write_text(
                 f'write destination {property_entry.destination_path}',
                 destination,
                 new_value,
             )

             return [
                 PropertyRoll(
                     destination_path=property_entry.destination_path,
                     old_value=old_value,
                     new_value=new_value,
                     revision=revision,
                 ),
             ]
	# Copyright 2026 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.
	"""Recipe module for rolling properties extracted from CI builds.

	This module provides the `PropertyRollApi` class, which facilitates updating
	a local JSON file by extracting output properties from a passing CI build.
	Key functionalities include:
	- Resolving the latest passing build of a specified builder.
	- Extracting a subset of output properties from that build.
	- Comparing the extracted properties with the local destination file.
	- Updating the local JSON file if its content differs from the extracted
	properties.
	- Generating `PropertyRoll` objects that encapsulate the changes for commit
	message formatting.
	"""

	from __future__ import annotations

	import json
	from collections.abc import Sequence

	from google.protobuf import json_format

	from recipe_engine import recipe_api

	from PB.go.chromium.org.luci.buildbucket.proto import common as common_pb2
	from PB.recipe_modules.pigweed.property_roll.property_entry import PropertyEntry

	from RECIPE_MODULES.fuchsia.roll_commit_message import (
	api as roll_commit_message_api,
	)
	from RECIPE_MODULES.pigweed.checkout import api as checkout_api


	class PropertyRoll(roll_commit_message_api.BaseRoll):
	"""Represents a roll of properties extracted from a CI build.

	This class holds information about a property roll operation, including
	the source build revision, and the old and new JSON content of the file.
	It's used to generate commit messages for such rolls.
	"""

	def __init__(
	self,
	destination_path: str,
	old_value: str \| None,
	new_value: str,
	revision: str,
	*args,
	**kwargs,
	):
	"""Initializes a PropertyRoll instance.

	Args:
	destination_path: The path to the destination file in the local
	checkout.
	old_value: The previous content of the destination file.
	new_value: The new content extracted from the build.
	revision: The commit hash or ID corresponding to the source build.
	*args: Additional arguments for the base class.
	**kwargs: Additional keyword arguments for the base class.
	"""
	super().__init__(args, *kwargs)
	self.destination_path = destination_path
	self.old_value = old_value
	self.new_value = new_value
	self.revision = revision

	def short_name(self) -> str:
	"""Returns a short name for the roll, typically the destination path."""
	return self.destination_path

	def message_header(self, force_summary_version: bool = False) -> str:
	"""Generates the header for the commit message.

	Args:
	force_summary_version: If True, a more concise header is generated.

	Returns:
	The commit message header string.
	"""
	header = self.destination_path
	if force_summary_version:
	return header

	try:
	value_string = self.new_value.strip()
	if '\n' not in value_string:
	new_header = f'{header}: {value_string}'
	if len(new_header) <= 62:
	header = new_header

	except UnicodeDecodeError: # pragma: no cover
	pass

	return header

	def message_body(
	self,
	*,
	force_summary_version: bool = False,
	escape_tags: Sequence[str] = (),
	filter_tags: Sequence[str] = (),
	) -> str:
	"""Generates the body of the commit message.

	Args:
	force_summary_version: If True, a more concise body is generated.
	escape_tags: A sequence of tags to escape in the message.
	filter_tags: A sequence of tags to filter from the message.

	Returns:
	The commit message body string.
	"""
	del force_summary_version, escape_tags, filter_tags # Unused.
	result = []
	if (
	'\n' not in str(self.old_value).strip()
	and '\n' not in self.new_value.strip()
	):
	if self.old_value is None:
	result.append('Initialized to') # pragma: no cover
	else:
	result.append('From')
	old_value = self.old_value
	if len(old_value.strip()) > 50:
	old_value = f'{len(old_value)} chars'
	result.append(f' {old_value.strip()}')
	result.append('To')
	new_value = self.new_value
	if len(new_value.strip()) > 50:
	new_value = f'{len(new_value)} chars'
	result.append(f' {new_value.strip()}')
	result.append('')

	return '\n'.join(result)

	def message_footer(self, *, send_comment: bool) -> str:
	"""Generates the footer for the commit message.

	Args:
	send_comment: If True, indicates a comment should be sent.

	Returns:
	An empty string, as property rolls typically don't have footers.
	"""
	del send_comment # Unused.
	return ''

	def output_property(self) -> dict[str, str]:
	"""Generates a dictionary of properties representing the roll.

	Returns:
	A dictionary containing details of the roll, such as path,
	old/new values, and revision.
	"""
	return {
	'path': self.destination_path,
	'old': str(self.old_value),
	'new': self.new_value,
	'revision': self.revision,
	}


	class PropertyRollApi(recipe_api.RecipeApi):
	"""Provides methods to roll properties extracted from CI builds."""

	PropertyRoll = PropertyRoll

	def update(
	self,
	checkout: checkout_api.CheckoutContext,
	property_entry: PropertyEntry,
	) -> list[PropertyRoll]:
	"""Updates a local JSON file by extracting output properties from a build.

	Args:
	checkout: The checkout context, providing repository information.
	property_entry: A protobuf message defining the property extraction,
	including project, bucket, builder, properties to extract, and
	destination path.

	Returns:
	A list containing a `PropertyRoll` object if the file was updated,
	or an empty list if the content was identical.
	"""
	with self.m.step.nest(property_entry.destination_path):
	build = self.m.buildbucket_util.last_build(
	project=property_entry.project or None,
	bucket=property_entry.bucket,
	builder=property_entry.builder,
	status=common_pb2.SUCCESS,
	fields=['id', 'input', 'output.properties'],
	)
	if not build:
	raise self.m.step.StepFailure(
	f'no passing build found for {property_entry.builder}'
	)

	properties_dict = json_format.MessageToDict(build.output.properties)
	if property_entry.properties:
	extracted = {}
	for key in property_entry.properties:
	if key not in properties_dict:
	raise self.m.step.StepFailure(
	f'property {key!r} missing in output properties of build {build.id}'
	)
	extracted[key] = properties_dict[key]
	else:
	extracted = properties_dict

	new_value = json.dumps(extracted, indent=2, sort_keys=True) + '\n'

	destination = checkout.root / property_entry.destination_path
	self.m.file.ensure_directory(
	f'ensure directory {destination.parent}',
	destination.parent,
	)
	old_value: str \| None = None
	self.m.path.mock_add_file(destination)
	if self.m.path.isfile(destination):
	old_value = self.m.file.read_text(
	f'read destination {property_entry.destination_path}',
	destination,
	)

	new = self.m.step.empty(f'new {property_entry.destination_path}')
	new.presentation.step_summary_text = repr(new_value)

	revision = str(
	properties_dict.get('got_revision')
	or build.input.gitiles_commit.id
	or build.id
	)

	if old_value == new_value:
	pres = self.m.step.empty('values are identical').presentation
	pres.step_summary_text = repr(old_value)

	pres.properties[property_entry.destination_path] = {
	'builder': property_entry.builder,
	'revision': revision,
	}

	return []

	self.m.file.write_text(
	f'write destination {property_entry.destination_path}',
	destination,
	new_value,
	)

	return [
	PropertyRoll(
	destination_path=property_entry.destination_path,
	old_value=old_value,
	new_value=new_value,
	revision=revision,
	),
	]