recipe_modules/ensure_tool/api.py

# Copyright 2021 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.

from recipe_engine import recipe_api

from PB.infra.tool_metadata import ToolMetadata


class EnsureToolApi(recipe_api.RecipeApi):
    """APIs for ensuring prebuilt tools are fetched."""

    def __init__(self, **kwargs):
        super(EnsureToolApi, self).__init__(**kwargs)
        self._installed_tools = {}
        self._lock = None

    def initialize(self):
        self._lock = self.m.futures.make_bounded_semaphore()

    def get_tool_metadata(self, tool_name, manifest_path):
        """
        Read a tool manifest and return its path and version.

        Args:
            tool_name (str): Name of tool to pass to test data.
            manifest (Path): Path to tool manifest jsonpb which can be parsed
              into a ToolMetadata proto message.

        Returns:
            ToolMetadata: Tool metadata from manifest.
        """
        return self.m.file.read_proto(
            "read manifest",
            manifest_path,
            ToolMetadata,
            "JSONPB",
            test_proto=ToolMetadata(
                path="path/to/%s" % tool_name,
                version="version:pinned-version",
            ),
        )

    def write_tool_metadata(self, manifest_path, path, version, do_not_autoroll=False):
        """
        Write a tool metadata to a manifest.

        Args:
            manifest_path (Path): Path to tool manifest jsonpb to write.
            path (str): Tool CIPD path.
            version (str): Tool CIPD version.
            do_not_autoroll (bool): Whether to instruct autorollers to ignore
              the tool.
        """
        tool_metadata = ToolMetadata(
            path=path,
            version=version,
            do_not_autoroll=do_not_autoroll,
        )
        self.m.file.write_proto(
            "write manifest",
            manifest_path,
            tool_metadata,
            "JSONPB",
        )

    def __call__(self, tool_name, manifest_path, executable_path=None):
        """
        Given a tool manifest, ensure the tool and return its Path.

        Args:
            tool_name (str): Human-readable tool name to embed in step name.
              Also passed to get_tool_metadata().
            manifest_path (Path): Passed to get_tool_metadata().
            executable_path (Path): Passed to api.cipd.ensure_tool().

        Returns:
            Path: Path to tool.
        """

        cache_key = (tool_name, manifest_path)

        with self._lock:
            if cache_key not in self._installed_tools:
                with self.m.step.nest("ensure %s" % tool_name):
                    tool_metadata = self.get_tool_metadata(tool_name, manifest_path)
                    self._installed_tools[cache_key] = self.m.cipd.ensure_tool(
                        tool_metadata.path,
                        tool_metadata.version,
                        executable_path=executable_path,
                    )

        return self._installed_tools[cache_key]