recipe_modules/dpi/api.py

# Copyright 2022 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.recipe_modules.fuchsia.dpi.options import DPIUploadOptions

SCOPES = [
    "https://www.googleapis.com/auth/cloud-platform",
    "https://www.googleapis.com/auth/userinfo.email",
]


class DpiAPI(recipe_api.RecipeApi):
    """APIs for running Decentralized Product Integration operations."""

    Options = DPIUploadOptions

    def fetch(
        self,
        step_name,
        artifact_lock,
        out_dir,
    ):
        """Fetches artifacts through DPI.

        Args:
            artifact_lock (Path): The artifact_lock.json path.
            out_dir (Path): The directory for output.
        """

        args = [
            self._dpi_tool,
            "fetch",
            artifact_lock,
            out_dir,
        ]
        return self._run(step_name, args)

    def upload(
        self,
        step_name,
        build_dir,
        options,
    ):
        """Uploads artifacts through DPI.

        Args:
            build_dir (Path): The build directory containing all the built
             artifacts. All the paths in manifest file are relative to this
             directory.
            options (DPIUploadOptions): The upload options.
        """
        token = self.m.service_account.default().get_access_token(scopes=SCOPES)
        access_token_path = self.m.raw_io.input_text(token)

        args = [
            self._dpi_tool,
            "upload",
            build_dir.join(options.manifest_path),
            access_token_path,
            "--build-dir",
            build_dir,
            "--repo-hostname",
            options.repo_hostname,
            "--gcs-bucket",
            options.gcs_bucket,
        ]
        return self._run(step_name, args)

    @property
    def _dpi_tool(self):
        return self.m.ensure_tool("dpi", self.resource("tool_manifest.json"))

    def _run(self, step_name, args, **kwargs):
        return self.m.step(step_name, args, **kwargs)