build/bazel/scripts/gn_targets_utils.py

# Copyright 2025 The Fuchsia Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Helper classes to deal with @gn_targets dependencies in the Fuchsia workspace."""

import dataclasses
import json
import subprocess
import sys
import typing as T
from pathlib import Path


@dataclasses.dataclass
class BazelBuildActionInfo(object):
    """Model an entry in the bazel_build_action_targets.json file.

    See //BUILD.gn for schema description.
    """

    # LINT.IfChange
    gn_target: str
    bazel_targets: list[str]
    no_sdk: bool
    gn_targets_dir: str
    gn_targets_manifest: str
    bazel_command_file: str = ""
    build_events_log_json: str = ""
    path_mapping: str = ""
    # LINT.ThenChange(//BUILD.gn:bazel_build_actions)


class BazelBuildActionsMap(object):
    """A class used to model a map of Bazel top-level targets to the GN bazel_action() labels that builds them.

    This is built from the content of the //:bazel_build_action_targets generated_file() output.
    This does not provide information about the dependencies of the top-level Bazel actions,
    for these, an actual Bazel query is needed, see BazelBuildActionQuery below.
    """

    def __init__(self, json_content: list[dict[str, T.Any]]) -> None:
        self._targets: dict[str, BazelBuildActionInfo] = {}
        self._bazel_to_gn: dict[str, str] = {}
        for entry in json_content:
            info = BazelBuildActionInfo(**entry)
            self._targets[info.gn_target] = info
            for bazel_target in info.bazel_targets:
                self._bazel_to_gn[bazel_target] = info.gn_target

    @staticmethod
    def FromBuildDir(build_dir: Path) -> "BazelBuildActionsMap":
        """Create instance from content of Ninja build directory.

        Args:
            build_dir: Ninja build directory, populated by `fx gen`.
        Returns:
            New BazelBuildActionsMap
        Raises:
            FileNotFoundError if file is missing.
        """
        with (build_dir / "bazel_build_action_targets.json").open("rb") as f:
            content = json.load(f)
        return BazelBuildActionsMap(content)

    @property
    def bazel_targets(self) -> list[str]:
        return sorted(self._bazel_to_gn.keys())

    @property
    def gn_targets(self) -> list[str]:
        return sorted(set(self._bazel_to_gn.values()))

    def get_info(self, gn_label: str) -> T.Optional[BazelBuildActionInfo]:
        """Retrieve BazelBuildActionInfo matching a given GN label."""
        return self._targets.get(gn_label)

    def find_gn_target_for(self, bazel_target: str) -> str:
        """Retrieve the GN target label used to build a given top-level Bazel target.

        Args:
            bazel_target: A top-level Bazel target that is built by one of
               the GN bazel_action() targets.

        Returns:
            The corresponding GN target label, or an empty string if there is no match.
        """
        return self._bazel_to_gn.get(bazel_target, "")


class BazelBuildActionQuery(object):
    """Convenience class to wrap Bazel query operations when trying to find
    the wrapping GN bazel_action() target for a given Bazel target.
          GN target, or in case of failure, a message explaining its reason.
    """

    def __init__(
        self, bazel_target: str, actions_map: BazelBuildActionsMap
    ) -> None:
        """Create instance."""
        self._bazel_target = bazel_target
        self._actions_map = actions_map

    def make_query_command(self, bazel_pre_cmd_args: list[str]) -> list[str]:
        """Return the query command to be performed.

        Note that this forces @gn_targets to be empty, which will generate errors
        that are ignored through the use of --keep_going. The error output *must*
        be filtered with filter_query_errors() to detect real errors, before
        calling process_query_output().

        Args:
            bazel_pre_cmd_args: List of command-line arguments that appear before the 'query'
               command. At a minimum, this would be a list with a single item with the path
               of the Bazel binary / launcher script.
        Returns:
            A string list representing a command to be passed to a function
            like subprocess.run().
        """
        query_expr = "allpaths(set(%s), %s)" % (
            " ".join(self._actions_map.bazel_targets),
            self._bazel_target,
        )

        return bazel_pre_cmd_args + [
            "query",
            "--config=no_gn_targets",
            "--config=quiet",
            "--keep_going",
            query_expr,
        ]

    @staticmethod
    def filter_query_errors(errors: str) -> str:
        """Filter the errors generated by the invocation of a make_query_command() result.

        This removes all errors that are due to the empty @gn_targets repository,
        but leaves any others, to catch unexpected breakages due to developer changes.

        Args:
            errors: The query's command stderr output as a string.
        Returns:
            The error output without expected errors related to  @gn_targets.
            A non-empty value means that real errors were detected during the query.
        """
        real_errors = []
        for error in errors.splitlines():
            if (
                error.startswith("ERROR: ")
                and "no such package '@@gn_targets//" in error
            ):
                continue
            if error.startswith(
                "Starting local Bazel server and connecting to it..."
            ):
                continue
            if error.startswith('ERROR: Evaluation of query "allpaths(set('):
                continue
            if error.startswith(
                "WARNING: --keep_going specified, ignoring errors."
            ):
                continue
            real_errors.append(error)

        return "\n".join(real_errors)

    def process_query_output(self, query_result: str) -> list[str]:
        """Parse the result of a Bazel query to get a list of GN  bazel_action() labels.

        Args:
            query_result: The stdout of running the Bazel cquery command generated
                by make_query_command() as a string.
        Returns:
            A list of GN target strings, each pointing to a target definition
            using bazel_action() or one of its wrappers, that depend on this
            instance's bazel target.
        """
        lines = query_result.splitlines()
        gn_targets = set()

        for line in lines:
            gn_target = self._actions_map.find_gn_target_for(line)
            if gn_target:
                gn_targets.add(gn_target)

        return sorted(gn_targets)


# A LogSinkType models a callable that takes an log message and returns None.
LogSinkType: T.TypeAlias = T.Callable[[str], None]


def _stderr_log_sink(msg: str) -> None:
    print(msg, file=sys.stderr)


# A CommandRunnerType is a callable that takes a list of command strings,
# and returns a (returncode, stdout, stderr) tuple from its execution.
CommandRunnerType: T.TypeAlias = T.Callable[[list[str]], tuple[int, str, str]]


def _subprocess_command_runner(args: list[str]) -> tuple[int, str, str]:
    """A CommandRunnerType implementation that uses subprocess.run()."""
    ret = subprocess.run(
        args, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE
    )
    return ret.returncode, ret.stdout, ret.stderr


def find_gn_bazel_action_infos_for(
    bazel_target: str,
    actions_map: BazelBuildActionsMap,
    bazel_cmd_args: list[str],
    log_step: T.Optional[LogSinkType] = None,
    log_err: LogSinkType = _stderr_log_sink,
    command_runner: CommandRunnerType = _subprocess_command_runner,
) -> list[BazelBuildActionInfo]:
    """Find the BazelBuildActionInfo instances matching a given Bazel target.

    Find all GN bazel_action() target whose top-level bazel targets
    depend on a given |bazel_target|, and return the corresponding
    BazelBuildActionInfo items in a list.

    This is the main logic around `fx bazel-tool set_gn_targets`.

    Args:
        bazel_target: Bazel target label to look for. Must start
           with // or @.

        actions_map: A BazelBuildActionsMap instance used as input.

        bazel_cmd_args: A list of strings making up a command-line
           invocation for Bazel, before the "query" command.
           E.g.: ["bazel"]

        log_step: An optional LogSink callable used to print log messages.

        log_err: An optional LogSink callable used to send error messages.
           Defaults to _stderr_log_sink which simply sends errors to
           sys.stderr.

        command_runner: An optional CommandRunner to run the final
           Bazel query command. Defaults to _subprocess_command_runner
           that uses subprocess.run.

    Returns:
        A list of BazelBuildActionInfo items. In case of failure,
        errors will be sent to |log_err| and the function will return
        an empty list.

        An empty list without errors means the target is not a known
        dependency of the actions_map's GN and Bazel targets.

    Raises:
        AssertionError if |bazel_target| has invalid format.
    """
    # Check inputs.
    if not bazel_target.startswith(("@", "//")):
        log_err(f"Target label must start with // or @: {bazel_target}")
        return []

    if "(" in bazel_target:
        log_err(
            f"Target label cannot include GN toolchain suffix: {bazel_target}"
        )
        return []

    gn_target = actions_map.find_gn_target_for(bazel_target)
    if gn_target:
        if log_step:
            log_step(
                f"Bazel target {bazel_target} maps directly to GN target {gn_target}"
            )
        info = actions_map.get_info(gn_target)
        assert info  # Appease mypy
        return [info]

    action_query = BazelBuildActionQuery(bazel_target, actions_map)
    query_command = action_query.make_query_command(bazel_cmd_args)
    returncode, stdout, stderr = command_runner(query_command)
    if returncode != 0:
        stderr = action_query.filter_query_errors(stderr)
        if stderr:
            # Report unexpected errors directly.
            log_err(f"Bazel query returned unexpected errors:\n%s\n" % stderr)
            return []

    gn_targets = action_query.process_query_output(stdout)

    if log_step:
        log_step(f"Bazel query result: {gn_targets}")

    result = []
    for gn_target in gn_targets:
        info = actions_map.get_info(gn_target)
        assert info  # Appease mypy
        result.append(info)

    return result