api/controller/test.py

# 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.

"""Test controller.

Handles all testing related functionality, it is not itself a test.
"""

import functools
import os
import string
import subprocess

from chromite.api import controller
from chromite.api import faux
from chromite.api import validate
from chromite.api.metrics import deserialize_metrics_log
from chromite.api.controller import controller_util
from chromite.api.gen.chromite.api import test_pb2
from chromite.api.gen.chromiumos import common_pb2
from chromite.api.gen.chromiumos.build.api import container_metadata_pb2
from chromite.api.gen.chromiumos.test.api import coverage_rule_pb2
from chromite.api.gen.chromiumos.test.api import dut_attribute_pb2
from chromite.api.gen.chromiumos.test.api import test_suite_pb2
from chromite.cbuildbot import goma_util
from chromite.lib import build_target_lib
from chromite.lib import chroot_lib
from chromite.lib import constants
from chromite.lib import cros_build_lib
from chromite.lib import image_lib
from chromite.lib import osutils
from chromite.lib import sysroot_lib
from chromite.lib.parser import package_info
from chromite.scripts import cros_set_lsb_release
from chromite.service import packages as packages_service
from chromite.service import test
from chromite.third_party.google.protobuf import json_format
from chromite.third_party.google.protobuf import text_format
from chromite.utils import key_value_store
from chromite.utils import metrics


@faux.empty_success
@faux.empty_completed_unsuccessfully_error
def DebugInfoTest(input_proto, _output_proto, config):
  """Run the debug info tests."""
  sysroot_path = input_proto.sysroot.path
  target_name = input_proto.sysroot.build_target.name

  if not sysroot_path:
    if target_name:
      sysroot_path = build_target_lib.get_default_sysroot_path(target_name)
    else:
      cros_build_lib.Die("The sysroot path or the sysroot's build target name "
                         'must be provided.')

  # We could get away with out this, but it's a cheap check.
  sysroot = sysroot_lib.Sysroot(sysroot_path)
  if not sysroot.Exists():
    cros_build_lib.Die('The provided sysroot does not exist.')

  if config.validate_only:
    return controller.RETURN_CODE_VALID_INPUT

  if test.DebugInfoTest(sysroot_path):
    return controller.RETURN_CODE_SUCCESS
  else:
    return controller.RETURN_CODE_COMPLETED_UNSUCCESSFULLY


def _BuildTargetUnitTestResponse(input_proto, output_proto, _config):
  """Add tarball path to a successful response."""
  output_proto.tarball_path = os.path.join(input_proto.result_path,
                                           'unit_tests.tar')


def _BuildTargetUnitTestFailedResponse(_input_proto, output_proto, _config):
  """Add failed packages to a failed response."""
  packages = ['foo/bar', 'cat/pkg']
  for pkg in packages:
    pkg_info = package_info.parse(pkg)
    pkg_info_msg = output_proto.failed_packages.add()
    controller_util.serialize_package_info(pkg_info, pkg_info_msg)


@faux.success(_BuildTargetUnitTestResponse)
@faux.error(_BuildTargetUnitTestFailedResponse)
@validate.require('build_target.name')
@validate.exists('result_path')
@validate.require_each('packages', ['category', 'package_name'])
@validate.validation_complete
@metrics.collect_metrics
def BuildTargetUnitTest(input_proto, output_proto, _config):
  """Run a build target's ebuild unit tests."""
  # Required args.
  result_path = input_proto.result_path

  # Method flags.
  # An empty sysroot means build packages was not run. This is used for
  # certain boards that need to use prebuilts (e.g. grunt's unittest-only).
  was_built = not input_proto.flags.empty_sysroot

  # Packages to be tested.
  packages_package_info = input_proto.packages
  packages = []
  for package_info_msg in packages_package_info:
    cpv = controller_util.PackageInfoToCPV(package_info_msg)
    packages.append(cpv.cp)

  # Skipped tests.
  blocklisted_package_info = input_proto.package_blocklist
  blocklist = []
  for package_info_msg in blocklisted_package_info:
    blocklist.append(controller_util.PackageInfoToString(package_info_msg))

  # Allow call to filter out non-cros_workon packages from the input packages.
  filter_only_cros_workon = input_proto.flags.filter_only_cros_workon

  # Allow call to succeed if no tests were found.
  testable_packages_optional = input_proto.flags.testable_packages_optional

  build_target = controller_util.ParseBuildTarget(input_proto.build_target)
  chroot = controller_util.ParseChroot(input_proto.chroot)

  code_coverage = input_proto.flags.code_coverage

  result = test.BuildTargetUnitTest(
      build_target,
      chroot,
      packages=packages,
      blocklist=blocklist,
      was_built=was_built,
      code_coverage=code_coverage,
      testable_packages_optional=testable_packages_optional,
      filter_only_cros_workon=filter_only_cros_workon)

  if not result.success:
    # Failed to run tests or some tests failed.
    # Record all failed packages.
    for pkg_info in result.failed_pkgs:
      package_info_msg = output_proto.failed_packages.add()
      controller_util.serialize_package_info(pkg_info, package_info_msg)
    if result.failed_pkgs:
      return controller.RETURN_CODE_UNSUCCESSFUL_RESPONSE_AVAILABLE
    else:
      return controller.RETURN_CODE_COMPLETED_UNSUCCESSFULLY

  sysroot = sysroot_lib.Sysroot(build_target.root)
  tarball = test.BuildTargetUnitTestTarball(chroot, sysroot, result_path)
  if tarball:
    output_proto.tarball_path = tarball
  deserialize_metrics_log(output_proto.events, prefix=build_target.name)


SRC_DIR = os.path.join(constants.SOURCE_ROOT, 'src')
PLATFORM_DEV_DIR = os.path.join(SRC_DIR, 'platform/dev')
TEST_SERVICE_DIR = os.path.join(PLATFORM_DEV_DIR, 'src/chromiumos/test')
TEST_CONTAINER_BUILD_SCRIPTS = {
    'cros-provision':
        os.path.join(TEST_SERVICE_DIR, 'provision/docker/build-dockerimage.sh'),
    'cros-dut':
        os.path.join(TEST_SERVICE_DIR, 'dut/docker/build-dockerimage.sh'),
    'cros-test':
        os.path.join(
            PLATFORM_DEV_DIR,
            'test/container/utils/build-dockerimage.sh'
        ),
    'cros-test-finder':
        os.path.join(
            TEST_SERVICE_DIR,
            'test_finder/docker/build-dockerimage.sh',
        ),
}


def _BuildTestServiceContainersResponse(input_proto, output_proto, _config):
  """Fake success response"""
  # pylint: disable=unused-argument
  output_proto.results.append(test_pb2.TestServiceContainerBuildResult(
      success=test_pb2.TestServiceContainerBuildResult.Success()
  ))


def _BuildTestServiceContainersFailedResponse(
    _input_proto, output_proto, _config):
  """Fake failure response"""

  # pylint: disable=unused-argument
  output_proto.results.append(test_pb2.TestServiceContainerBuildResult(
      failure=test_pb2.TestServiceContainerBuildResult.Failure(
          error_message='fake error'
      )
  ))


@validate.constraint('valid docker tag')
def _ValidDockerTag(tag):
  """Check that a string meets requirements for Docker tag naming."""
  # Tags can't start with period or dash
  if tag[0] in '.-':
    return "tag can't begin with '.' or '-'"

  # Tags can only consist of [a-zA-Z0-9-_.]
  allowed_chars = set(string.ascii_letters+string.digits+'-_.')
  invalid_chars = set(tag) - allowed_chars
  if invalid_chars:
    return f'saw one or more invalid characters: [{"".join(invalid_chars)}]'

  # Finally, max tag length is 128 characters
  if len(tag) > 128:
    return 'maximum tag length is 128 characters'


@validate.constraint('valid docker label key')
def _ValidDockerLabelKey(key):
  """Check that a string meets requirements for Docker tag naming."""

  # Label keys should start and end with a lowercase letter
  lowercase = set(string.ascii_lowercase)
  if not (key[0] in lowercase and key[-1] in lowercase):
    return "label key doesn't start and end with lowercase letter"

  # Label keys can have lower-case alphanumeric characters, period and dash
  allowed_chars = set(string.ascii_lowercase+string.digits+'-.')
  invalid_chars = set(key) - allowed_chars
  if invalid_chars:
    return f'saw one or more invalid characters: [{"".join(invalid_chars)}]'

  # Repeated . and - aren't allowed
  for char in '.-':
    if char * 2 in key:
      return f"'{char}' can\'t be repeated in label key" % char


@faux.success(_BuildTestServiceContainersResponse)
@faux.error(_BuildTestServiceContainersFailedResponse)
@validate.require('build_target.name')
@validate.require('chroot.path')
@validate.check_constraint('tags', _ValidDockerTag)
@validate.check_constraint('labels', _ValidDockerLabelKey)
@validate.validation_complete
def BuildTestServiceContainers(
    input_proto: test_pb2.BuildTestServiceContainersRequest,
    output_proto: test_pb2.BuildTestServiceContainersResponse, _config):
  """Builds docker containers for all test services and pushes them to gcr.io"""
  build_target = controller_util.ParseBuildTarget(input_proto.build_target)
  chroot = controller_util.ParseChroot(input_proto.chroot)
  sysroot = sysroot_lib.Sysroot(build_target.root)

  tags = ','.join(input_proto.tags)
  labels = (
      f'{key}={value}' for key, value in input_proto.labels.items()
  )

  for human_name, build_script in TEST_CONTAINER_BUILD_SCRIPTS.items():
    with osutils.TempDir(prefix='test_container') as tempdir:
      output_path = os.path.join(tempdir, 'metadata.jsonpb')

      # Note that we use an output file instead of stdout to avoid any issues
      # with maintaining stdout hygiene.  Stdout and stderr are combined to
      # form the error log in response to any errors.
      cmd = [build_script, chroot.path, sysroot.path]

      if input_proto.HasField('repository'):
        cmd += ['--host', input_proto.repository.hostname]
        cmd += ['--project', input_proto.repository.project]

      cmd += ['--tags', tags]
      cmd += ['--output', output_path]
      cmd += labels

      result = test_pb2.TestServiceContainerBuildResult()
      result.name = human_name

      cmd_result = cros_build_lib.run(cmd, check=False,
                                      stderr=subprocess.STDOUT,
                                      stdout=True)
      if cmd_result.returncode == 0:
        # Read the ContainerImageInfo message produced by the container build.
        image_info = container_metadata_pb2.ContainerImageInfo()
        json_format.Parse(osutils.ReadFile(output_path), image_info)

        result.success.CopyFrom(
            test_pb2.TestServiceContainerBuildResult.Success(
                image_info=image_info
            )
        )
      else:
        result.failure.CopyFrom(
            test_pb2.TestServiceContainerBuildResult.Failure(
                error_message=cmd_result.stdout
            )
        )

      output_proto.results.append(result)


@faux.empty_success
@faux.empty_completed_unsuccessfully_error
@validate.validation_complete
def ChromiteUnitTest(_input_proto, _output_proto, _config):
  """Run the chromite unit tests."""
  if test.ChromiteUnitTest():
    return controller.RETURN_CODE_SUCCESS
  else:
    return controller.RETURN_CODE_COMPLETED_UNSUCCESSFULLY


@faux.empty_success
@faux.empty_completed_unsuccessfully_error
@validate.validation_complete
def ChromitePytest(_input_proto, _output_proto, _config):
  """Run the chromite unit tests."""
  # TODO(vapier): Delete this stub.
  return controller.RETURN_CODE_SUCCESS


@faux.all_empty
@validate.require('sysroot.path', 'sysroot.build_target.name', 'chrome_root')
@validate.validation_complete
def SimpleChromeWorkflowTest(input_proto, _output_proto, _config):
  """Run SimpleChromeWorkflow tests."""
  if input_proto.goma_config.goma_dir:
    chromeos_goma_dir = input_proto.goma_config.chromeos_goma_dir or None
    goma = goma_util.Goma(
        input_proto.goma_config.goma_dir,
        input_proto.goma_config.goma_client_json,
        stage_name='BuildApiTestSimpleChrome',
        chromeos_goma_dir=chromeos_goma_dir)
  else:
    goma = None
  return test.SimpleChromeWorkflowTest(input_proto.sysroot.path,
                                       input_proto.sysroot.build_target.name,
                                       input_proto.chrome_root,
                                       goma)


@faux.all_empty
@validate.require('build_target.name', 'vm_path.path', 'test_harness',
                  'vm_tests')
@validate.validation_complete
def VmTest(input_proto, _output_proto, _config):
  """Run VM tests."""
  build_target_name = input_proto.build_target.name
  vm_path = input_proto.vm_path.path

  test_harness = input_proto.test_harness

  vm_tests = input_proto.vm_tests

  cmd = ['cros_run_test', '--debug', '--no-display', '--copy-on-write',
         '--board', build_target_name, '--image-path', vm_path,
         '--%s' % test_pb2.VmTestRequest.TestHarness.Name(test_harness).lower()]
  cmd.extend(vm_test.pattern for vm_test in vm_tests)

  if input_proto.ssh_options.port:
    cmd.extend(['--ssh-port', str(input_proto.ssh_options.port)])

  if input_proto.ssh_options.private_key_path:
    cmd.extend(['--private-key', input_proto.ssh_options.private_key_path.path])

  # TODO(evanhernandez): Find a nice way to pass test_that-args through
  # the build API. Or obviate them.
  if test_harness == test_pb2.VmTestRequest.AUTOTEST:
    cmd.append('--test_that-args=--allow-chrome-crashes')

  with osutils.TempDir(prefix='vm-test-results.') as results_dir:
    cmd.extend(['--results-dir', results_dir])
    cros_build_lib.run(cmd, kill_timeout=10 * 60)


@faux.all_empty
@validate.require('image_payload.path.path', 'cache_payloads')
@validate.require_each('cache_payloads', ['path.path'])
@validate.validation_complete
def MoblabVmTest(input_proto, _output_proto, _config):
  """Run Moblab VM tests."""
  chroot = controller_util.ParseChroot(input_proto.chroot)
  image_payload_dir = input_proto.image_payload.path.path
  cache_payload_dirs = [cp.path.path for cp in input_proto.cache_payloads]

  # Autotest and Moblab depend on the builder path, so we must read it from
  # the image.
  image_file = os.path.join(image_payload_dir, constants.TEST_IMAGE_BIN)
  with osutils.TempDir() as mount_dir:
    with image_lib.LoopbackPartitions(image_file, destination=mount_dir) as lp:
      # The file we want is /etc/lsb-release, which lives in the ROOT-A
      # disk partition.
      partition_paths = lp.Mount([constants.PART_ROOT_A])
      assert len(partition_paths) == 1, (
          'expected one partition path, got: %r' % partition_paths)
      partition_path = partition_paths[0]
      lsb_release_file = os.path.join(partition_path,
                                      constants.LSB_RELEASE_PATH.strip('/'))
      lsb_release_kvs = key_value_store.LoadFile(lsb_release_file)
      builder = lsb_release_kvs.get(cros_set_lsb_release.LSB_KEY_BUILDER_PATH)

  if not builder:
    cros_build_lib.Die('Image did not contain key %s in %s',
                       cros_set_lsb_release.LSB_KEY_BUILDER_PATH,
                       constants.LSB_RELEASE_PATH)

  # Now we can run the tests.
  with chroot.tempdir() as workspace_dir, chroot.tempdir() as results_dir:
    # Convert the results directory to an absolute chroot directory.
    chroot_results_dir = '/%s' % os.path.relpath(results_dir, chroot.path)
    vms = test.CreateMoblabVm(workspace_dir, chroot.path, image_payload_dir)
    cache_dir = test.PrepareMoblabVmImageCache(vms, builder, cache_payload_dirs)
    test.RunMoblabVmTest(chroot, vms, builder, cache_dir, chroot_results_dir)
    test.ValidateMoblabVmTest(results_dir)


@faux.all_empty
@validate.validation_complete
def CrosSigningTest(_input_proto, _output_proto, _config):
  """Run the cros-signing unit tests."""
  test_runner = os.path.join(constants.SOURCE_ROOT, 'cros-signing', 'signer',
                             'run_tests.py')
  result = cros_build_lib.run([test_runner], check=False)

  return result.returncode


def GetArtifacts(in_proto: common_pb2.ArtifactsByService.Test,
                 chroot: chroot_lib.Chroot, sysroot_class: sysroot_lib.Sysroot,
                 build_target: build_target_lib.BuildTarget,
                 output_dir: str) -> list:
  """Builds and copies test artifacts to specified output_dir.

  Copies test artifacts to output_dir, returning a list of (output_dir: str)
  paths to the desired files.

  Args:
    in_proto: Proto request defining reqs.
    chroot: The chroot class used for these artifacts.
    sysroot_class: The sysroot class used for these artifacts.
    build_target: The build target used for these artifacts.
    output_dir: The path to write artifacts to.

  Returns:
    A list of dictionary mappings of ArtifactType to list of paths.
  """
  generated = []

  artifact_types = {
      in_proto.ArtifactType.UNIT_TESTS: test.BuildTargetUnitTestTarball,
      in_proto.ArtifactType.CODE_COVERAGE_LLVM_JSON:
          test.BundleCodeCoverageLlvmJson,
      in_proto.ArtifactType.HWQUAL: functools.partial(
          test.BundleHwqualTarball,
          build_target.name, packages_service.determine_full_version()),
  }

  for output_artifact in in_proto.output_artifacts:
    for artifact_type, func in artifact_types.items():
      if artifact_type in output_artifact.artifact_types:
        paths = func(chroot, sysroot_class, output_dir)
        if paths:
          generated.append({
              'paths': [paths] if isinstance(paths, str) else paths,
              'type': artifact_type,
          })

  return generated


def _GetCoverageRulesResponseSuccess(
    _input_proto, output_proto: test_pb2.GetCoverageRulesResponse, _config):
  output_proto.coverage_rules.append(
      coverage_rule_pb2.CoverageRule(
          name='kernel:4.4',
          test_suites=[
              test_suite_pb2.TestSuite(
                  test_case_tag_criteria=test_suite_pb2.TestSuite
                  .TestCaseTagCriteria(tags=['kernel']))
          ],
          dut_criteria=[
              dut_attribute_pb2.DutCriterion(
                  attribute_id=dut_attribute_pb2.DutAttribute.Id(
                      value='system_build_target'),
                  values=['overlayA'],
              )
          ],
      ),)


@faux.success(_GetCoverageRulesResponseSuccess)
@faux.empty_error
@validate.require('source_test_plans')
@validate.exists('dut_attribute_list.path', 'build_metadata_list.path',
                 'flat_config_list.path')
@validate.validation_complete
def GetCoverageRules(input_proto: test_pb2.GetCoverageRulesRequest,
                     output_proto: test_pb2.GetCoverageRulesResponse, _config):
  """Call the testplan tool to generate CoverageRules."""
  source_test_plans = input_proto.source_test_plans
  dut_attribute_list = input_proto.dut_attribute_list
  build_metadata_list = input_proto.build_metadata_list
  flat_config_list = input_proto.flat_config_list

  cmd = [
      'testplan', 'generate', '-dutattributes', dut_attribute_list.path,
      '-buildmetadata', build_metadata_list.path, '-flatconfiglist',
      flat_config_list.path, '-logtostderr', '-v', '2'
  ]

  with osutils.TempDir(prefix='get_coverage_rules_input') as tempdir:
    # Write all input files required by testplan, and read the output file
    # containing CoverageRules.
    for i, plan in enumerate(source_test_plans):
      plan_path = os.path.join(tempdir, 'source_test_plan_%d.textpb' % i)
      osutils.WriteFile(plan_path, text_format.MessageToString(plan))
      cmd.extend(['-plan', plan_path])

    out_path = os.path.join(tempdir, 'out.jsonpb')
    cmd.extend(['-out', out_path])

    cros_build_lib.run(cmd)

    out_text = osutils.ReadFile(out_path)

  # The output file contains CoverageRules as jsonpb, separated by newlines.
  coverage_rules = []
  for out_line in out_text.splitlines():
    coverage_rule = coverage_rule_pb2.CoverageRule()
    json_format.Parse(out_line, coverage_rule)
    coverage_rules.append(coverage_rule)

  output_proto.coverage_rules.extend(coverage_rules)