recipe_modules/run_script_roll/api.py

# 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.
"""Enable roll logic to be in-tree."""

from __future__ import annotations

from collections.abc import Mapping, Sequence
from typing import Any

from recipe_engine import recipe_api

from PB.recipe_modules.pigweed.run_script_roll.run_script_entry import (
    RunScriptEntry,
)

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 RunScriptRoll(roll_commit_message_api.BaseRoll):
    """Represents a roll created by running an in-tree script."""

    def __init__(
        self,
        name: str,
        header: str,
        body: str,
        properties: Mapping[str, Any],
        *args,
        **kwargs,
    ):
        """Initializes a RunScriptRoll instance.

        Args:
            name: Short name for the roll.
            header: Short description of the roll.
            body: Details of what changed in the roll.
            *args: Additional arguments for the base class.
            **kwargs: Additional keyword arguments for the base class.
        """
        super().__init__(*args, **kwargs)
        self.name = name
        self.header = header
        self.body = body
        self.properties = properties

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

    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.
        """
        if force_summary_version:
            return self.name
        return f'{self.name}: {self.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.
        return self.body

    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 run script 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.
        """
        return self.properties


class RunScriptRollApi(recipe_api.RecipeApi):
    """Allows for in-tree roll logic."""

    RunScriptRoll = RunScriptRoll

    def run(
        self,
        checkout: checkout_api.CheckoutContext,
        run_script_entry: RunScriptEntry,
    ) -> list[RunScriptRoll]:
        """Runs an in-tree script that modifies the checkout.

        Args:
            checkout: The checkout context, providing repository information.
            run_script_entry: A protobuf message defining the script to run.

        Returns:
            A list containing a `RunScriptRoll` object if the tree was updated,
            or an empty list if no updates were made.
        """
        with self.m.step.nest(run_script_entry.script):
            script = checkout.root / run_script_entry.script

            result = self.m.step(
                'run',
                [script],
                stdout=self.m.json.output(),
                step_test_data=lambda: self.m.json.test_api.output_stream(
                    {
                        'name': 'foo',
                        'header': 'changed some files',
                        'body': 'abc: changed\ndef: changed\nghi: changed\n',
                        'properties': {'key': 'value'},
                    }
                ),
            ).stdout

            if not result:
                return []

            return [
                RunScriptRoll(
                    name=result['name'],
                    header=result['header'],
                    body=result['body'],
                    properties=result['properties'],
                ),
            ]