| # 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. |
| """Recipe module for rolling files copied from remote Git repositories. |
| |
| This module provides the `CopyRollApi` class, which facilitates updating |
| a local file by copying its content from a specified source file in a |
| remote Git repository. Key functionalities include: |
| - Resolving the latest revision of the source file's repository. |
| - Fetching the content of the source file at that revision. |
| - Comparing the fetched content with the local destination file. |
| - Updating the local file if its content differs from the source. |
| - Generating `CopyRoll` objects that encapsulate the changes for commit |
| message formatting. |
| """ |
| |
| from __future__ import annotations |
| |
| import urllib.parse |
| from collections.abc import Sequence |
| |
| from recipe_engine import recipe_api |
| |
| from PB.recipe_modules.pigweed.copy_roll.copy_entry import CopyEntry |
| |
| 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 CopyRoll(roll_commit_message_api.BaseRoll): |
| """Represents a roll of a file copied from a remote repository. |
| |
| This class holds information about a file copy operation, including the |
| source revision, and the old and new content of the file. It's used |
| to generate commit messages for such rolls. |
| """ |
| |
| def __init__( |
| self, |
| destination_path: str, |
| old_value: str, |
| new_value: str, |
| revision: str, |
| *args, |
| **kwargs, |
| ): |
| """Initializes a CopyRoll 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 fetched from the source. |
| revision: The commit hash from the source repository corresponding |
| to the new_value. |
| *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 copy 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': self.old_value, |
| 'new': self.new_value, |
| 'revision': self.revision, |
| } |
| |
| |
| class CopyRollApi(recipe_api.RecipeApi): |
| """Provides methods to roll files copied from remote Git repositories.""" |
| |
| CopyRoll = CopyRoll |
| |
| def update( |
| self, |
| checkout: checkout_api.CheckoutContext, |
| copy_entry: CopyEntry, |
| ) -> list[CopyRoll]: |
| """Updates a local file by copying its content from a remote source. |
| |
| This method fetches the content of a specified file (`source_path`) |
| from a remote Git repository at its latest revision (or a specified |
| branch). It then compares this content with the local file |
| (`destination_path`). If the content differs, the local file is |
| updated. |
| |
| Args: |
| checkout: The checkout context, providing repository information. |
| copy_entry: A protobuf message defining the copy operation, |
| including remote URL, source path, destination path, and |
| branch. |
| |
| Returns: |
| A list containing a `CopyRoll` object if the file was updated, |
| or an empty list if the content was identical. |
| """ |
| with self.m.step.nest(copy_entry.destination_path): |
| new_revision = self.m.git_roll_util.resolve_new_revision( |
| copy_entry.remote, |
| copy_entry.branch or 'main', |
| checkout.remotes_equivalent, |
| ) |
| |
| parsed_remote = urllib.parse.urlparse(copy_entry.remote) |
| |
| destination = checkout.root / copy_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 {copy_entry.destination_path}', |
| destination, |
| ) |
| |
| new_value = self.m.gitiles.fetch( |
| host=parsed_remote.netloc, |
| project=parsed_remote.path.lstrip('/'), |
| ref=new_revision, |
| path=copy_entry.source_path, |
| step_name=f'read source {copy_entry.source_path}', |
| ).decode() |
| |
| new = self.m.step.empty(f'new {copy_entry.source_path}') |
| new.presentation.step_summary_text = repr(new_value) |
| |
| if old_value == new_value: |
| pres = self.m.step.empty('values are identical').presentation |
| pres.step_summary_text = repr(old_value) |
| |
| pres.properties[copy_entry.destination_path] = { |
| 'remote': copy_entry.remote, |
| 'revision': new_revision, |
| } |
| |
| return [] |
| |
| self.m.file.write_text( |
| f'write destination {copy_entry.destination_path}', |
| destination, |
| new_value, |
| ) |
| |
| return [ |
| CopyRoll( |
| destination_path=copy_entry.destination_path, |
| old_value=old_value, |
| new_value=new_value, |
| revision=new_revision, |
| ), |
| ] |
