recipe_modules/cipd_util/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


class CIPDPlatformApi(recipe_api.RecipeApi):
    """Utilities for interacting with CIPD."""

    @property
    def platform_name(self):
        """Returns CIPD's name for the current platform.

        This is the value that the CIPD CLI will substitute for ${platform}
        in any package names that it sees.
        """
        os = self.m.platform.name.replace("win", "windows")
        arch = {"intel": {32: "386", 64: "amd64"}, "arm": {32: "armv6", 64: "arm64"}}[
            self.m.platform.arch
        ][self.m.platform.bits]
        return "%s-%s" % (os, arch)

    def upload_package(
        self,
        pkg_name,
        pkg_root,
        pkg_paths,
        search_tag,
        repository=None,
        install_mode="copy",
        refs=("latest",),
        metadata=None,
        add_version_file=True,
        name=None,
        step_name="cipd",
    ):
        """Creates and uploads a CIPD package containing the tool at pkg_dir.

        The tool is published to CIPD under the path pkg_name.

        Args:
            pkg_name (basestr): The CIPD package to publish to.
            pkg_root (Path): The absolute path to the parent directory of the
                package.
            pkg_paths (list(Path)): A list of Path objects which specify the
                paths to directories or files to upload.
            search_tag (dict): The tag to search for the CIPD pin with. This
                should contain one element and be either `git_revision` or
                `version`.
            repository (str or None): The git repository where code for the
                package lives.
            install_mode (str or None): The install mode for the package.
            refs (str): Refs to set on the package.
            metadata (list of pair/Metadata or None): Metadata to add to the
                package.
            add_version_file (bool): Include a .cipd_version file in the
                package.
            name (str or None): Logical name of the package. Defaults to
                second-to-last part of pkg_name.
            step_name (str): Name of the step.

        Returns:
            The CIPDApi.Pin instance_id.
        """
        pkg_def = self.m.cipd.PackageDefinition(
            package_name=pkg_name, package_root=pkg_root, install_mode=install_mode
        )

        # Mock the existence of the package root directory so `isdir()` will
        # return True.
        self.m.path.mock_add_directory(pkg_root)
        for path in pkg_paths:
            if self.m.path.isdir(path):
                pkg_def.add_dir(path)
            else:
                pkg_def.add_file(path)

        if not name:
            # E.g., "fuchsia/go/linux-amd64" -> "go".
            name = str(pkg_name.split("/")[-2])
        if add_version_file:
            pkg_def.add_version_file(".versions/%s.cipd_version" % name)

        with self.m.step.nest(step_name) as step:
            assert (
                len(search_tag) == 1
            ), "search_tag must contain one (key: value) pair to search for."
            search_tag_key = list(search_tag.keys())[0]
            search_tag_value = search_tag[search_tag_key]
            cipd_pins = self.m.cipd.search(
                pkg_name,
                "%s:%s" % (search_tag_key, search_tag_value),
                test_instances=[],
            )
            if cipd_pins:
                self.m.step("package is up-to-date", cmd=None)
                assert len(cipd_pins) == 1, "%s has too many pins" % pkg_name
                return cipd_pins[0].instance_id

            tags = {}
            tags.update(search_tag)

            final_metadata = [
                self.m.cipd.Metadata("bbid", self.m.buildbucket_util.id),
            ]
            if repository:
                final_metadata.append(
                    self.m.cipd.Metadata("git_repository", repository)
                )

            if metadata:
                for metadatum in metadata:
                    if isinstance(metadatum, self.m.cipd.Metadata):
                        final_metadata.append(metadatum)
                    else:
                        key, value = metadatum
                        final_metadata.append(self.m.cipd.Metadata(key, value))
                        # TODO(fxbug.dev/85982) Remove metadata from tags.
                        tags[key] = value

            cipd_pin = self.m.cipd.create_from_pkg(
                pkg_def=pkg_def,
                refs=refs,
                tags=tags,
                metadata=final_metadata,
            )
            step.presentation.properties.update(cipd_pin._asdict())
            if search_tag:
                # Return search_tag to output properties so it can be
                # used by builders like goma_toolchain.
                step.presentation.properties.update(search_tag)

            return cipd_pin.instance_id