api/controller/packages.py

# -*- coding: utf-8 -*-
# Copyright 2019 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Package related functionality."""

from __future__ import print_function

from chromite.api import faux
from chromite.api import validate
from chromite.api.controller import controller_util
from chromite.api.gen.chromite.api import binhost_pb2
from chromite.api.gen.chromite.api import packages_pb2
from chromite.api.gen.chromiumos import common_pb2
from chromite.lib import constants
from chromite.lib import cros_build_lib
from chromite.lib import cros_logging as logging
from chromite.lib import portage_util
from chromite.lib.uprev_lib import GitRef
from chromite.lib.parser import package_info
from chromite.service import packages


_OVERLAY_TYPE_TO_NAME = {
    binhost_pb2.OVERLAYTYPE_PUBLIC: constants.PUBLIC_OVERLAYS,
    binhost_pb2.OVERLAYTYPE_PRIVATE: constants.PRIVATE_OVERLAYS,
    binhost_pb2.OVERLAYTYPE_BOTH: constants.BOTH_OVERLAYS,
}

def _UprevResponse(_input_proto, output_proto, _config):
  """Add fake paths to a successful uprev response."""
  output_proto.modified_ebuilds.add().path = '/fake/path1'
  output_proto.modified_ebuilds.add().path = '/fake/path2'

@faux.success(_UprevResponse)
@faux.empty_error
@validate.require('overlay_type')
@validate.is_in('overlay_type', _OVERLAY_TYPE_TO_NAME)
@validate.validation_complete
def Uprev(input_proto, output_proto, _config):
  """Uprev all cros workon ebuilds that have changes."""
  build_targets = controller_util.ParseBuildTargets(input_proto.build_targets)
  overlay_type = _OVERLAY_TYPE_TO_NAME[input_proto.overlay_type]
  chroot = controller_util.ParseChroot(input_proto.chroot)
  output_dir = input_proto.output_dir or None

  try:
    uprevved = packages.uprev_build_targets(build_targets, overlay_type, chroot,
                                            output_dir)
  except packages.Error as e:
    # Handle module errors nicely, let everything else bubble up.
    cros_build_lib.Die(e)

  for path in uprevved:
    output_proto.modified_ebuilds.add().path = path


def _UprevVersionedPackageResponse(_input_proto, output_proto, _config):
  """Add fake paths to a successful uprev versioned package response."""
  uprev_response = output_proto.responses.add()
  uprev_response.modified_ebuilds.add().path = '/uprev/response/path'


@faux.success(_UprevVersionedPackageResponse)
@faux.empty_error
@validate.require('versions')
@validate.require('package_info.package_name', 'package_info.category')
@validate.validation_complete
def UprevVersionedPackage(input_proto, output_proto, _config):
  """Uprev a versioned package.

  See go/pupr-generator for details about this endpoint.
  """
  chroot = controller_util.ParseChroot(input_proto.chroot)
  build_targets = controller_util.ParseBuildTargets(input_proto.build_targets)
  package = controller_util.PackageInfoToCPV(input_proto.package_info)
  refs = []
  for ref in input_proto.versions:
    refs.append(GitRef(path=ref.repository, ref=ref.ref, revision=ref.revision))

  try:
    result = packages.uprev_versioned_package(package, build_targets, refs,
                                              chroot)
  except packages.Error as e:
    # Handle module errors nicely, let everything else bubble up.
    cros_build_lib.Die(e)

  if not result.uprevved:
    # No uprevs executed, skip the output population.
    return

  for modified in result.modified:
    uprev_response = output_proto.responses.add()
    uprev_response.version = modified.new_version
    for path in modified.files:
      uprev_response.modified_ebuilds.add().path = path


def _GetBestVisibleResponse(_input_proto, output_proto, _config):
  """Add fake paths to a successful GetBestVisible response."""
  pkg_info_msg = common_pb2.PackageInfo(
      category='category',
      package_name='name',
      version='1.01',
  )
  output_proto.package_info.CopyFrom(pkg_info_msg)


@faux.success(_GetBestVisibleResponse)
@faux.empty_error
@validate.require('atom')
@validate.validation_complete
def GetBestVisible(input_proto, output_proto, _config):
  """Returns the best visible PackageInfo for the indicated atom."""
  build_target = None
  if input_proto.build_target.name:
    build_target = controller_util.ParseBuildTarget(input_proto.build_target)

  cpv = packages.get_best_visible(input_proto.atom, build_target=build_target)
  pkg_info_msg = common_pb2.PackageInfo()
  controller_util.CPVToPackageInfo(cpv, pkg_info_msg)
  output_proto.package_info.CopyFrom(pkg_info_msg)


def _ChromeVersionResponse(_input_proto, output_proto, _config):
  """Add a fake chrome version to a successful response."""
  output_proto.version = '78.0.3900.0'


@faux.success(_ChromeVersionResponse)
@faux.empty_error
@validate.require('build_target.name')
@validate.validation_complete
def GetChromeVersion(input_proto, output_proto, _config):
  """Returns the chrome version."""
  build_target = controller_util.ParseBuildTarget(input_proto.build_target)
  chrome_version = packages.determine_chrome_version(build_target)
  if chrome_version:
    output_proto.version = chrome_version


def _GetTargetVersionsResponse(_input_proto, output_proto, _config):
  """Add fake target version fields to a successful response."""
  output_proto.android_version = '5812377'
  output_proto.android_branch_version = 'git_nyc-mr1-arc'
  output_proto.android_target_version = 'cheets'
  output_proto.chrome_version = '78.0.3900.0'
  output_proto.platform_version = '12438.0.0'
  output_proto.milestone_version = '78'
  output_proto.full_version = 'R78-12438.0.0'


@faux.success(_GetTargetVersionsResponse)
@faux.empty_error
@validate.require('build_target.name')
@validate.require_each('packages', ['category', 'package_name'])
@validate.validation_complete
def GetTargetVersions(input_proto, output_proto, _config):
  """Returns the target versions."""
  build_target = controller_util.ParseBuildTarget(input_proto.build_target)
  # Android version.
  android_version = packages.determine_android_version([build_target.name])
  logging.info('Found android version: %s', android_version)
  if android_version:
    output_proto.android_version = android_version
  # Android branch version.
  android_branch_version = packages.determine_android_branch(build_target.name)
  logging.info('Found android branch version: %s', android_branch_version)
  if android_branch_version:
    output_proto.android_branch_version = android_branch_version
  # Android target version.
  android_target_version = packages.determine_android_target(build_target.name)
  logging.info('Found android target version: %s', android_target_version)
  if android_target_version:
    output_proto.android_target_version = android_target_version

  # TODO(crbug/1019770): Investigate cases where builds_chrome is true but
  # chrome_version is None.

  # If input_proto.packages is empty, then the default set of packages will
  # be used as defined in dependency.GetBuildDependency.
  package_list = None
  if input_proto.packages:
    package_list = [
        controller_util.PackageInfoToCPV(x) for x in input_proto.packages
    ]
  builds_chrome = packages.builds(constants.CHROME_CP, build_target,
                                  packages=package_list)
  if builds_chrome:
    # Chrome version fetch.
    chrome_version = packages.determine_chrome_version(build_target)
    logging.info('Found chrome version: %s', chrome_version)
    if chrome_version:
      output_proto.chrome_version = chrome_version

  # The ChromeOS version info.
  output_proto.platform_version = packages.determine_platform_version()
  output_proto.milestone_version = packages.determine_milestone_version()
  output_proto.full_version = packages.determine_full_version()


def _GetBuilderMetadataResponse(input_proto, output_proto, _config):
  """Add fake metadata fields to a successful response."""
  # Populate only a few fields to validate faux testing.
  build_target_metadata = output_proto.build_target_metadata.add()
  build_target_metadata.build_target = input_proto.build_target.name
  build_target_metadata.android_container_branch = 'git_pi-arc'
  model_metadata = output_proto.model_metadata.add()
  model_metadata.model_name = 'astronaut'
  model_metadata.ec_firmware_version = 'coral_v1.1.1234-56789f'


@faux.success(_GetBuilderMetadataResponse)
@faux.empty_error
@validate.require('build_target.name')
@validate.validation_complete
def GetBuilderMetadata(input_proto, output_proto, _config):
  """Returns the target builder metadata."""
  build_target = controller_util.ParseBuildTarget(input_proto.build_target)
  build_target_metadata = output_proto.build_target_metadata.add()
  build_target_metadata.build_target = build_target.name
  # Android version.
  android_version = packages.determine_android_version([build_target.name])
  logging.info('Found android version: %s', android_version)
  if android_version:
    build_target_metadata.android_container_version = android_version
  # Android branch version.
  android_branch_version = packages.determine_android_branch(build_target.name)
  logging.info('Found android branch version: %s', android_branch_version)
  if android_branch_version:
    build_target_metadata.android_container_branch = android_branch_version
  # Android target version.
  android_target_version = packages.determine_android_target(build_target.name)
  logging.info('Found android target version: %s', android_target_version)
  if android_target_version:
    build_target_metadata.android_container_target = android_target_version

  build_target_metadata.arc_use_set = 'arc' in portage_util.GetBoardUseFlags(
      build_target.name)

  fw_versions = packages.determine_firmware_versions(build_target)
  build_target_metadata.main_firmware_version = fw_versions.main_fw_version
  build_target_metadata.ec_firmware_version = fw_versions.ec_fw_version
  build_target_metadata.kernel_version = packages.determine_kernel_version(
      build_target)
  fingerprints = packages.find_fingerprints(build_target)
  build_target_metadata.fingerprints.extend(fingerprints)

  models = packages.get_models(build_target)
  if models:
    all_fw_versions = packages.get_all_firmware_versions(build_target)
    for model in models:
      if model in all_fw_versions:
        fw_versions = all_fw_versions[model]
        ec = fw_versions.ec_rw or fw_versions.ec
        main_ro = fw_versions.main
        main_rw = fw_versions.main_rw or main_ro
        # Get the firmware key-id for the current board and model.
        key_id = packages.get_key_id(build_target, model)
        model_metadata = output_proto.model_metadata.add()
        model_metadata.model_name = model
        model_metadata.ec_firmware_version = ec
        model_metadata.firmware_key_id = key_id
        model_metadata.main_readonly_firmware_version = main_ro
        model_metadata.main_readwrite_firmware_version = main_rw


def _HasPrebuiltSuccess(_input_proto, output_proto, _config):
  """The mock success case for HasChromePrebuilt."""
  output_proto.has_prebuilt = True


@faux.success(_HasPrebuiltSuccess)
@faux.empty_error
@validate.require('build_target.name')
@validate.validation_complete
def HasChromePrebuilt(input_proto, output_proto, _config):
  """Checks if the most recent version of Chrome has a prebuilt."""
  build_target = controller_util.ParseBuildTarget(input_proto.build_target)
  useflags = 'chrome_internal' if input_proto.chrome else None
  exists = packages.has_prebuilt(constants.CHROME_CP, build_target=build_target,
                                 useflags=useflags)

  output_proto.has_prebuilt = exists


@faux.success(_HasPrebuiltSuccess)
@faux.empty_error
@validate.require('build_target.name', 'package_info.category',
                  'package_info.package_name')
@validate.validation_complete
def HasPrebuilt(input_proto, output_proto, _config):
  """Checks if the most recent version of Chrome has a prebuilt."""
  build_target = controller_util.ParseBuildTarget(input_proto.build_target)
  package = controller_util.PackageInfoToCPV(input_proto.package_info).cp
  useflags = 'chrome_internal' if input_proto.chrome else None
  exists = packages.has_prebuilt(
      package, build_target=build_target, useflags=useflags)

  output_proto.has_prebuilt = exists


def _BuildsChromeSuccess(_input_proto, output_proto, _config):
  """Mock success case for BuildsChrome."""
  output_proto.builds_chrome = True


@faux.success(_BuildsChromeSuccess)
@faux.empty_error
@validate.require('build_target.name')
@validate.require_each('packages', ['category', 'package_name'])
@validate.validation_complete
def BuildsChrome(input_proto, output_proto, _config):
  """Check if the board builds chrome."""
  build_target = controller_util.ParseBuildTarget(input_proto.build_target)
  cpvs = [controller_util.PackageInfoToCPV(pi) for pi in input_proto.packages]
  builds_chrome = packages.builds(constants.CHROME_CP, build_target, cpvs)
  output_proto.builds_chrome = builds_chrome


def _NeedsChromeSourceSuccess(_input_proto, output_proto, _config):
  """Mock success case for NeedsChromeSource."""
  output_proto.needs_chrome_source = True
  output_proto.builds_chrome = True

  output_proto.reasons.append(
      packages_pb2.NeedsChromeSourceResponse.NO_PREBUILT)
  pkg_info_msg = output_proto.packages.add()
  pkg_info_msg.category = constants.CHROME_CN
  pkg_info_msg.package_name = constants.CHROME_PN

  output_proto.reasons.append(
      packages_pb2.NeedsChromeSourceResponse.FOLLOWER_LACKS_PREBUILT)
  for pkg in constants.OTHER_CHROME_PACKAGES:
    pkg_info_msg = output_proto.packages.add()
    pkg_info = package_info.parse(pkg)
    controller_util.serialize_package_info(pkg_info, pkg_info_msg)


@faux.success(_NeedsChromeSourceSuccess)
@faux.empty_error
@validate.require('install_request.sysroot.build_target.name')
@validate.exists('install_request.sysroot.path')
@validate.validation_complete
def NeedsChromeSource(input_proto, output_proto, _config):
  """Check if the build will need the chrome source."""
  # Input parsing.
  build_target = controller_util.ParseBuildTarget(
      input_proto.install_request.sysroot.build_target)
  compile_source = input_proto.install_request.flags.compile_source
  pkgs = [controller_util.deserialize_package_info(pi) for pi in
          input_proto.install_request.packages]
  use_flags = [f.flag for f in input_proto.install_request.use_flags]

  result = packages.needs_chrome_source(
      build_target,
      compile_source=compile_source,
      packages=pkgs,
      useflags=use_flags)

  # Record everything in the response.
  output_proto.needs_chrome_source = result.needs_chrome_source
  output_proto.builds_chrome = result.builds_chrome

  # Compile source reason.
  if compile_source:
    output_proto.reasons.append(
        packages_pb2.NeedsChromeSourceResponse.COMPILE_SOURCE)

  # No chrome prebuilt reason.
  if result.missing_chrome_prebuilt:
    output_proto.reasons.append(
        packages_pb2.NeedsChromeSourceResponse.NO_PREBUILT)

  # Follower package(s) lack prebuilt reason.
  if result.missing_follower_prebuilt:
    output_proto.reasons.append(
        packages_pb2.NeedsChromeSourceResponse.FOLLOWER_LACKS_PREBUILT)

  for pkg in result.packages:
    pkg_info = output_proto.packages.add()
    controller_util.serialize_package_info(pkg, pkg_info)