api/controller/sdk.py

# Copyright 2019 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""SDK chroot operations."""

import os
from typing import Dict, Union

from chromite.api import controller
from chromite.api import faux
from chromite.api import validate
from chromite.api.controller import controller_util
from chromite.api.gen.chromiumos import common_pb2
from chromite.lib import cros_build_lib
from chromite.service import sdk


def _ChrootVersionResponse(_input_proto, output_proto, _config):
    """Add a fake chroot version to a successful response."""
    output_proto.version.version = 168


def _BinhostCLs(_input_proto, output_proto, _config):
    """Add fake CL identifiers to a successful response."""
    output_proto.cls = [
        "fakecl:1",
        "fakecl:2",
    ]


def _BuildSdkTarballResponse(_input_proto, output_proto, _config):
    """Populate a fake BuildSdkTarballResponse."""
    output_proto.sdk_tarball_path.path = "/fake/sdk/tarball.tar.gz"
    output_proto.sdk_tarball_path.location = common_pb2.Path.OUTSIDE


@faux.success(_BuildSdkTarballResponse)
@validate.require("chroot")
@validate.validation_complete
def BuildSdkTarball(
    input_proto: "BuildSdkTarballRequest",
    output_proto: "BuildSdkTarballResponse",
    _config: "api_config.ApiConfig",
) -> None:
    chroot = controller_util.ParseChroot(input_proto.chroot)
    output_proto.sdk_tarball_path.path = str(sdk.BuildSdkTarball(chroot))
    output_proto.sdk_tarball_path.location = common_pb2.Path.OUTSIDE


def _CreateManifestFromSdkResponse(_input_proto, output_proto, _config):
    """Populate a fake CreateManifestFromSdkResponse."""
    output_proto.manifest_path.path = "/fake/sdk/tarball.tar.gz.Manifest"
    output_proto.manifest_path.location = common_pb2.Path.Location.INSIDE


@faux.success(_CreateManifestFromSdkResponse)
@validate.require("chroot")
@validate.require("sdk_path")
@validate.require("dest_dir")
@validate.validation_complete
def CreateManifestFromSdk(
    input_proto: "CreateManifestFromSdkRequest",
    output_proto: "CreateManifestFromSdkResponse",
    _config: "api_config.ApiConfig",
) -> None:
    """Create a manifest file showing the ebuilds in an SDK."""

    def _assert_path_is_absolute(path: str, name: str):
        """Raise an exception if the given path is not absolute."""
        if not os.path.isabs(path):
            cros_build_lib.Die(f"The {name} must be absolute; got {path}")

    _assert_path_is_absolute(input_proto.chroot.path, "chroot path")
    _assert_path_is_absolute(input_proto.sdk_path.path, "SDK path")
    _assert_path_is_absolute(input_proto.dest_dir.path, "destination directory")

    sdk_path = controller_util.pb2_path_to_pathlib_path(
        input_proto.sdk_path, input_proto.chroot
    )
    dest_dir = controller_util.pb2_path_to_pathlib_path(
        input_proto.dest_dir, input_proto.chroot
    )

    manifest_path = sdk.CreateManifestFromSdk(sdk_path, dest_dir)
    output_proto.manifest_path.path = str(manifest_path)
    output_proto.manifest_path.location = common_pb2.Path.Location.OUTSIDE


@faux.success(_ChrootVersionResponse)
@faux.empty_error
def Create(
    input_proto: "CreateRequest",
    output_proto: "CreateResponse",
    config: "api_config.ApiConfig",
) -> Union[int, None]:
    """Chroot creation, includes support for replacing an existing chroot.

    Args:
        input_proto: The input proto.
        output_proto: The output proto.
        config: The API call config.

    Returns:
        An error code, None otherwise.
    """
    replace = not input_proto.flags.no_replace
    bootstrap = input_proto.flags.bootstrap
    chroot = controller_util.ParseChroot(input_proto.chroot)

    sdk_version = input_proto.sdk_version
    skip_chroot_upgrade = input_proto.skip_chroot_upgrade

    if config.validate_only:
        return controller.RETURN_CODE_VALID_INPUT

    args = sdk.CreateArguments(
        replace=replace,
        bootstrap=bootstrap,
        chroot=chroot,
        sdk_version=sdk_version,
        skip_chroot_upgrade=skip_chroot_upgrade,
    )

    version = sdk.Create(args)

    if version:
        output_proto.version.version = version
    else:
        # This should be very rare, if ever used, but worth noting.
        cros_build_lib.Die(
            "No chroot version could be found. There was likely an"
            "error creating the chroot that was not detected."
        )


@faux.success(_ChrootVersionResponse)
@faux.empty_error
@validate.require_each("toolchain_targets", ["name"])
@validate.validation_complete
def Update(
    input_proto: "UpdateRequest",
    output_proto: "UpdateResponse",
    _config: "api_config.ApiConfig",
):
    """Update the chroot.

    Args:
        input_proto: The input proto.
        output_proto: The output proto.
        _config: The API call config.
    """
    build_source = input_proto.flags.build_source
    targets = [target.name for target in input_proto.toolchain_targets]
    toolchain_changed = input_proto.flags.toolchain_changed

    args = sdk.UpdateArguments(
        build_source=build_source,
        toolchain_targets=targets,
        toolchain_changed=toolchain_changed,
    )

    version = sdk.Update(args)

    if version:
        output_proto.version.version = version
    else:
        # This should be very rare, if ever used, but worth noting.
        cros_build_lib.Die(
            "No chroot version could be found. There was likely an"
            "error creating the chroot that was not detected."
        )


@faux.all_empty
@validate.require("binhost_gs_bucket")
@validate.validation_complete
def Uprev(input_proto, output_proto, _config):
    """Update SDK version file and prebuilt files to point to the latest SDK.

    Files will be changed locally, but not committed.
    """
    # If the UprevRequest did not specify a target version,
    # check the remote SDK version file on Google Cloud Storage for the latest
    # uprev target.
    target_version = (
        input_proto.version or sdk.get_latest_uprev_target_version()
    )

    # The main uprev logic occurs in service/sdk.py.
    modified_files = sdk.uprev_sdk_and_prebuilts(
        binhost_gs_bucket=input_proto.binhost_gs_bucket,
        version=target_version,
    )

    # Populate the UprevResponse object with the modified files.
    for modified_file in modified_files:
        proto_path = output_proto.modified_files.add()
        proto_path.path = str(modified_file)
        proto_path.location = common_pb2.Path.OUTSIDE
    output_proto.version = target_version


@faux.all_empty
@validate.validation_complete
def Delete(input_proto, _output_proto, _config):
    """Delete a chroot."""
    chroot = controller_util.ParseChroot(input_proto.chroot)
    sdk.Delete(chroot, force=True)


@faux.all_empty
@validate.validation_complete
def Unmount(_input_proto, _output_proto, _config):
    """Unmount a chroot"""
    # Deprecated. Do nothing.


@faux.all_empty
@validate.require("path.path")
@validate.validation_complete
def UnmountPath(input_proto, _output_proto, _config):
    """Unmount a path"""
    sdk.UnmountPath(input_proto.path.path)


@faux.all_empty
@validate.validation_complete
def Clean(input_proto, _output_proto, _config):
    """Clean unneeded files from a chroot."""
    chroot = controller_util.ParseChroot(input_proto.chroot)

    # Default (flagless) call sets 'safe' and 'sysroots'.
    if not (
        input_proto.safe
        or input_proto.images
        or input_proto.sysroots
        or input_proto.tmp
        or input_proto.cache
        or input_proto.logs
        or input_proto.workdirs
        or input_proto.incrementals
    ):
        sdk.Clean(chroot, safe=True, sysroots=True)
    else:
        sdk.Clean(
            chroot,
            safe=input_proto.safe,
            images=input_proto.images,
            sysroots=input_proto.sysroots,
            tmp=input_proto.tmp,
            cache=input_proto.cache,
            logs=input_proto.logs,
            workdirs=input_proto.workdirs,
            incrementals=input_proto.incrementals,
        )


@faux.all_empty
@validate.validation_complete
def BuildPrebuilts(input_proto, _output_proto, _config):
    """Build the binary packages that comprise the Chromium OS SDK."""
    chroot = controller_util.ParseChroot(input_proto.chroot)
    sdk.BuildPrebuilts(chroot, board=input_proto.build_target.name)


@faux.success(_BinhostCLs)
@faux.empty_error
@validate.require(
    "prepend_version", "version", "upload_location", "sdk_tarball_template"
)
@validate.validation_complete
def CreateBinhostCLs(
    input_proto: "CreateBinhostCLsRequest",
    output_proto: "CreateBinhostCLsResponse",
    _config: "api_config.ApiConfig",
) -> None:
    """Create CLs to update the binhost to point at uploaded prebuilts."""
    output_proto.cls.extend(
        sdk.CreateBinhostCLs(
            input_proto.prepend_version,
            input_proto.version,
            input_proto.upload_location,
            input_proto.sdk_tarball_template,
        )
    )


@faux.all_empty
@validate.require("prepend_version", "version", "upload_location")
@validate.validation_complete
def UploadPrebuiltPackages(input_proto, _output_proto, _config):
    """Upload prebuilt packages."""
    sdk.UploadPrebuiltPackages(
        controller_util.ParseChroot(input_proto.chroot),
        input_proto.prepend_version,
        input_proto.version,
        input_proto.upload_location,
    )


@faux.all_empty
@validate.require("chroot")
@validate.validation_complete
def BuildSdkToolchain(input_proto, output_proto, _config):
    """Build cross-compiler packages for the SDK."""
    chroot = controller_util.ParseChroot(input_proto.chroot)
    extra_env: Dict[str, str] = {}
    if input_proto.use_flags:
        extra_env["USE"] = " ".join(use.flag for use in input_proto.use_flags)
    generated_files = sdk.BuildSdkToolchain(chroot, extra_env=extra_env)
    output_proto.generated_files.extend(generated_files)