recipe_modules/variable_roll/api.py

# Copyright 2025 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 updating version pins stored as variables in text files.

This module provides the `VariableRollApi` for scenarios where a dependency's
commit hash (or other version string) is embedded directly within a text-based
file as a variable assignment. For example:

    MY_VARIABLE = "commit_hash_or_version_string"
    # or
    OTHER_VARIABLE = 'another_hash'

The API can:
  - Locate such variable assignments within a file.
  - Resolve the latest commit hash from a specified remote Git repository and branch.
  - Update the variable's value to this new hash.
  - Add or update metadata comments near the variable to indicate it's
    automatically managed.
  - Generate `Roll` objects for standardized commit messages.
"""

from __future__ import annotations

import dataclasses
import re
from collections.abc import Callable, Sequence

from recipe_engine import config_types, recipe_api

from PB.recipe_modules.pigweed.variable_roll.variable_entry import (
    CipdVariableEntry,
    VariableEntry,
)

from RECIPE_MODULES.fuchsia.git_roll_util import api as git_roll_util_api
from RECIPE_MODULES.pigweed.checkout import api as checkout_api


@dataclasses.dataclass
class RevisionChange:
    """Represents a change in a revision string, typically a commit hash.

    Attributes:
        old: The old revision string.
        new: The new revision string.
        finalize: A callable that, when invoked, performs the actual update
                  (e.g., writing the new revision to a file). This allows
                  deferring the write operation.
    """

    old: str
    new: str
    finalize: Callable[[], None] = dataclasses.field(repr=False)


class VariableRollApi(recipe_api.RecipeApi):
    """API for updating version pins stored as variables in text files.

    This module handles scenarios where a dependency's commit hash is stored
    in a file like:

        MY_VARIABLE = "commit_hash_here"

    It can find such variables, update them to the latest commit from a
    specified remote and branch, and add metadata comments.
    """

    def _regex(
        self,
        variable: str,
        value_regex: str,
    ) -> re.Pattern:
        return re.compile(
            r'^'
            r'(?P<prefix>\s*)'
            rf'(?P<variable>{variable})'
            r'(?P<equal>\s*=\s*)'
            r'(?P<quote>[\'"]?)'
            rf'(?P<value>{value_regex})'
            r'(?P=quote)'
            r'(?P<suffix>\s*)'
            r'$',
        )

    def _get_variable_value(
        self,
        *,
        path: config_types.Path,
        variable: str,
        value_regex: str,
        test_data: str | None = None,
    ) -> str | None:
        lines = self.m.file.read_text(
            'read',
            path,
            test_data=test_data,
        ).splitlines()

        for line in lines:
            regex = self._regex(
                variable=variable,
                value_regex=value_regex,
            )
            if match := regex.search(line):
                return match.group('value')

        return None  # pragma: no cover

    def _update_variable(
        self,
        *,
        path: config_types.Path,
        variable: str,
        new_revision: str,
        value_regex: str,
        comment_prefix: str,
        test_data: str | None = None,
    ) -> RevisionChange | None:
        """Internal helper to find and prepare an update for a variable in a file.

        Reads the specified file, searches for a line matching the variable
        assignment pattern (e.g., `VAR = 'hash'`), and if found, prepares
        the lines to be written with the new revision and metadata comments.

        Args:
            path: The absolute path to the file to update.
            variable: The name of the variable to search for.
            new_revision: The new commit hash to set for the variable.
            value_regex: The pattern the current variable value is expected to
                match.
            comment_prefix: The string to use for starting comment lines
                (e.g., '#').
            test_data: Optional test data to use for file content in tests.

        Returns:
            A `RevisionChange` object if the variable was found and an update
            is pending, or None if the variable was not found.
        """
        lines = self.m.file.read_text(
            'read',
            path,
            test_data=test_data,
        ).splitlines()

        for i, line in enumerate(lines):
            if line.strip().startswith(f'{comment_prefix} ROLL:'):
                lines[i] = None
                continue

            match = self._regex(
                variable=variable,
                value_regex=value_regex,
            ).search(line)
            if not match:
                continue

            metadata_prefix = f'{match.group("prefix")}{comment_prefix} ROLL: '
            now = self.m.time.utcnow().strftime('%Y-%m-%d')
            new_lines = [
                f'{metadata_prefix}Warning: this variable is automatically '
                'updated.',
                f'{metadata_prefix}Last updated {now}.',
                f'{metadata_prefix}By {self.m.buildbucket.build_url()}.',
            ]

            new_lines.append(
                ''.join(
                    [
                        match.group('prefix'),
                        match.group('variable'),
                        match.group('equal'),
                        match.group('quote'),
                        new_revision,
                        match.group('quote'),
                        match.group('suffix'),
                    ],
                ),
            )

            lines[i] = '\n'.join(new_lines)

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

            return RevisionChange(
                old=match.group('value'),
                new=new_revision,
                finalize=update_file,
            )

    def update(
        self,
        checkout: checkout_api.CheckoutContext,
        variable_entry: VariableEntry,
        test_data: str | None = None,
    ) -> list[git_roll_util_api.Roll]:
        """Updates a variable in a file to a new commit hash from a remote.

        This function orchestrates the update of a version pin. It resolves the
        latest revision from the specified remote and branch, then uses
        `_update_variable` to find and modify the variable in the target file.
        If an update occurs, it generates a `Roll` object for commit message
        formatting.

        Args:
            checkout: The checkout context.
            variable_entry: A protobuf message defining the variable to update,
                            including file path, variable name, remote URL,
                            branch, and comment prefix.
            test_data: Optional test data for file content, used during testing.

        Returns:
            A list containing a `Roll` object if the variable was updated,
            or an empty list if no update was needed (e.g., already up-to-date,
            backwards roll, or variable not found).
        """
        variable_entry.comment_prefix = variable_entry.comment_prefix or '#'

        with self.m.step.nest(
            f'{variable_entry.path}[{variable_entry.variable}]',
        ):
            branch = variable_entry.branch or 'main'

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

            full_path = checkout.root / variable_entry.path

            if test_data is None:
                test_data = (
                    f"BAR = '{'1' * 40}'\n"
                    '# ROLL: Comment\n'
                    f'FOO = "{"0" * 40}"\n'
                    f'BAZ = "{"2" * 40}"\n'
                )

            change = self._update_variable(
                path=full_path,
                variable=variable_entry.variable,
                new_revision=new_revision,
                value_regex=r'[0-9a-fA-F]{40}',
                comment_prefix=variable_entry.comment_prefix,
                test_data=test_data,
            )
            if not change:
                self.m.step.empty(
                    f'variable {variable_entry.variable} not found',
                )
                return []

            short_name = f'{variable_entry.path}[{variable_entry.variable}]'

            try:
                roll = self.m.git_roll_util.get_roll(
                    repo_url=variable_entry.remote,
                    repo_short_name=short_name,
                    old_rev=change.old,
                    new_rev=new_revision,
                )

            except self.m.git_roll_util.BackwardsRollError:
                props = self.m.step.empty('output property').presentation
                props.properties[short_name] = {
                    'remote': variable_entry.remote,
                    'revision': new_revision,
                }

                return []

            change.finalize()

            return [roll]

    def update_cipd(
        self,
        checkout: checkout_api.CheckoutContext,
        variable_entry: CipdVariableEntry,
        platforms: Sequence[str] | None = None,
        test_data: str | None = None,
    ):
        with self.m.step.nest(
            f'{variable_entry.path}[{variable_entry.variable}]',
        ):
            variable_entry.comment_prefix = variable_entry.comment_prefix or '#'
            if not variable_entry.name:
                variable_entry.name = self.m.cipd_roll.package_name(
                    variable_entry.spec,
                )

            chars = r'@.,\w-'
            value_regex = rf'[{chars}]+(?::[{chars}]+)?'

            # TODO: b/234874582 - Support Windows.
            if platforms is None:
                platforms = ('mac-amd64', 'mac-arm64', 'linux-amd64')

            if test_data is None:
                test_data = (
                    f"BAR = 'git_revision:{'1' * 40}'\n"
                    '# ROLL: Comment\n'
                    f'FOO = "git_revision:{"0" * 40}"\n'
                    f'BAZ = "git_revision:{"2" * 40}"\n'
                )

            path = checkout.root / variable_entry.path
            old_version = self._get_variable_value(
                path=path,
                variable=variable_entry.variable,
                value_regex=value_regex,
                test_data=test_data,
            )
            if not old_version:  # pragma: no cover
                self.m.step.empty(f"couldn't find {variable_entry.variable}")
                return []

            new_version = self.m.cipd_roll.select_version(
                spec=variable_entry.spec,
                ref=variable_entry.ref,
                tag=variable_entry.tag,
                allow_mismatched_refs=variable_entry.allow_mismatched_refs,
                platforms=platforms,
                old_version=old_version,
            )
            if not new_version:
                # select_version() failing is adequately covered by the
                # cipd_roll module.
                return []  # pragma: no cover

            change = self._update_variable(
                path=path,
                variable=variable_entry.variable,
                new_revision=new_version,
                value_regex=value_regex,
                comment_prefix=variable_entry.comment_prefix,
                test_data=test_data,
            )
            assert change, 'should never get here: already found variable'

            roll = self.m.cipd_roll.Roll(
                package_name=variable_entry.name,
                package_spec=variable_entry.spec,
                old_version=old_version,
                new_version=new_version,
            )

            change.finalize()

            return [roll]