recipe_modules/cipd_upload/api.py

# Copyright 2020 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.
"""Utility functions for uploading a cipd package."""

from __future__ import annotations

from typing import Sequence, TYPE_CHECKING

from PB.recipe_modules.fuchsia.cipd_util.upload_manifest import (
    CIPDUploadManifest,
)

from recipe_engine import recipe_api

if TYPE_CHECKING:  # pragma: no cover
    from recipe_engine import config_types
    from RECIPE_MODULES.pigweed.checkout import api as checkout_api


class CipdUploadApi(recipe_api.RecipeApi):
    """Utility functions for uploading a cipd package."""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._platform: str | None = None

    @property
    def platform(self) -> str:
        if self._platform:
            return self._platform
        operating_system = self.m.platform.name.replace('win', 'windows')
        architecture = {
            'intel': {
                32: '386',
                64: 'amd64',
            },
            'arm': {
                32: 'armv6',
                64: 'arm64',
            },
        }[self.m.platform.arch][self.m.platform.bits]
        self._platform = '-'.join((operating_system, architecture))
        return self._platform

    def __call__(
        self,
        cipd_path: str,
        package_root: config_types.Path,
        search_tag: dict[str, str],
        paths: Sequence[config_types.Path] = (),
        repository: str = None,
        install_mode: str | None = 'copy',
        metadata: list[tuple[str, str]] | None = None,
        add_platform: bool = True,
    ) -> str:
        """Create CIPD package and upload it.

        Args:
          cipd_path: CIPD package path
          package_root: path from which relative paths should be calculated,
            if neither package_dirs nor package_files is listed all files within
            this directory are included in the package
          search_tag: primary key for revision of package
          paths: list of files and directories to include in package
          repository: URL of repository, used as a CIPD tag
          install_mode: 'copy', 'symlink', or None
          metadata: metadata to add to package
          add_platform: add platform string to the end of cipd_path

        Returns:
          Instance id of uploaded package.
        """

        name = str(cipd_path.split('/')[-1])

        if add_platform:
            assert not cipd_path.endswith(self.platform)
            cipd_path = '/'.join((cipd_path.rstrip('/'), self.platform))

        if not paths:
            paths = [package_root]

        return self.m.cipd_util.upload_package(
            pkg_name=cipd_path,
            pkg_root=package_root,
            pkg_paths=paths,
            search_tag=search_tag,
            repository=repository,
            install_mode=install_mode,
            metadata=metadata,
            name=name,
        )

    def manifest(
        self,
        manifest_path: config_types.Path,
        build_dir: config_types.Path,
        checkout: checkout_api.CheckoutContext,
        cas_digests: dict[str, str],
        upload_to_cipd: bool,
    ):
        manifest = self.m.file.read_proto(
            f'read cipd manifest {manifest_path}',
            manifest_path,
            CIPDUploadManifest,
            codec='JSONPB',
        )

        self.m.cipd_util.upload_from_manifest(
            cipd_package=manifest.pkg_name,
            cipd_manifest=manifest,
            build_dir=build_dir,
            repository=checkout.options.remote,
            git_revision=checkout.revision(),
            upload_to_cipd=upload_to_cipd,
            cas_digests=cas_digests,
        )