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


# Service account used to obtain OAuth for accessing Android build service.
# We use LUCI Auth Impersonation go/luci-authorization#service-accounts-impersonation
# to get authentication for accessing Android build API. This allows us to keep
# separate Cloud projects for using different Google Cloud APIs.
_service_account = (
    "aemu-artifact-mover@fuchsia-cloud-api-for-test.iam.gserviceaccount.com"
)

# Auth scope used with luci-auth. Note that we need the additional scope
# "https://www.googleapis.com/auth/androidbuild.internal" in order to access
# Android build service.
_scopes = [
    "https://www.googleapis.com/auth/cloud-platform",
    "https://www.googleapis.com/auth/firebase",
    "https://www.googleapis.com/auth/gerritcodereview",
    "https://www.googleapis.com/auth/userinfo.email",
    "https://www.googleapis.com/auth/androidbuild.internal",
]


class AEMUDownloaderAPI(recipe_api.RecipeApi):
    """APIs for running android_downloader."""

    def get_latest_build_id(self, step_name, artifacts, branch, **kwargs):
        """Runs android_downloader tool to get the latest Android build ID
        where all targets we use has been built successfully.

        We currently check for Android build targets:

        - emulator-linux_x64_internal
        - emulator-mac_x64
        - emulator-linux_aarch64
        - emulator-mac_aarch64

        Args:
            step_name (str): Name of the step.
            artifacts (dict<str,array<str>>): Artifacts to check the existence of.
            branch (str): Which Android branch to check for builds.
            kwargs (dict): Passed to self.m.step.
        """
        args = [
            self._luci_auth,
            "context",
            "-scopes",
            " ".join(_scopes),
            "-act-as-service-account",
            _service_account,
            "--",
            self._android_downloader_tool,
            "-find_bid",
            "-branch",
            branch,
        ]

        for target, paths in artifacts.items():
            for path in paths:
                args.append("-artifact")
                args.append(f"{target},{path}")

        step = self._run(step_name, args, **kwargs)
        return step

    def download(
        self, step_name, android_build_id, artifacts, branch, output_root, **kwargs
    ):
        """Runs android_downloader to download artifacts from a specified Android build ID.

        Args:
            step_name (str): Name of the step.
            android_build_id (str): Android build id to use for artifact download.
            artifacts (dict<str,array<str>>): Artifacts to download.
            branch (str): Which Android branch to download artifacts from.
            output_root (Path): Path root to store downloaded artifacts.
            kwargs (dict): Passed to self.m.step.
        """
        args = [
            self._luci_auth,
            "context",
            "-scopes",
            " ".join(_scopes),
            "-act-as-service-account",
            _service_account,
            "--",
            self._android_downloader_tool,
            "-bid",
            android_build_id,
            "-output",
            output_root,
            "-branch",
            branch,
        ]

        for target, paths in artifacts.items():
            for path in paths:
                args.append("-artifact")
                args.append(f"{target},{path}")

        step = self._run(step_name, args, **kwargs)
        return step

    @property
    def _android_downloader_tool(self):
        return self.m.ensure_tool(
            "aemu", self.resource("android_downloader/tool_manifest.json")
        )

    @property
    def _luci_auth(self):
        return self.m.ensure_tool(
            "luci-auth", self.resource("luci_auth/tool_manifest.json")
        )

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