xref: /tools/asuite/atest/atest_main.py

#!/usr/bin/env python3
#
# Copyright 2017, The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Command line utility for running Android tests through TradeFederation.

atest helps automate the flow of building test modules across the Android
code base and executing the tests via the TradeFederation test harness.

atest is designed to support any test types that can be ran by TradeFederation.
"""

# pylint: disable=too-many-lines

from __future__ import annotations
from __future__ import print_function

from abc import ABC, abstractmethod
import argparse
import collections
from dataclasses import dataclass
import itertools
import logging
import os
import platform
import sys
import tempfile
import time
from typing import Any, Dict, List, Set, Tuple

from atest import arg_parser
from atest import atest_configs
from atest import atest_execution_info
from atest import atest_utils
from atest import banner
from atest import bazel_mode
from atest import bug_detector
from atest import cli_translator
from atest import constants
from atest import device_update
from atest import module_info
from atest import result_reporter
from atest import test_runner_handler
from atest.atest_enum import DetectType, ExitCode
from atest.coverage import coverage
from atest.metrics import metrics
from atest.metrics import metrics_base
from atest.metrics import metrics_utils
from atest.test_finders import test_finder_utils
from atest.test_finders import test_info
from atest.test_finders.test_info import TestInfo
from atest.test_runner_invocation import TestRunnerInvocation
from atest.tools import indexing
from atest.tools import start_avd as avd

EXPECTED_VARS = frozenset([
    constants.ANDROID_BUILD_TOP,
    'ANDROID_TARGET_OUT_TESTCASES',
    constants.ANDROID_OUT,
])
TEST_RUN_DIR_PREFIX = '%Y%m%d_%H%M%S'
CUSTOM_ARG_FLAG = '--'
OPTION_NOT_FOR_TEST_MAPPING = (
    'Option "{}" does not work for running tests in TEST_MAPPING files'
)

DEVICE_TESTS = 'tests that require device'
HOST_TESTS = 'tests that do NOT require device'
RESULT_HEADER_FMT = '\nResults from %(test_type)s:'
RUN_HEADER_FMT = '\nRunning %(test_count)d %(test_type)s.'
TEST_COUNT = 'test_count'
TEST_TYPE = 'test_type'
END_OF_OPTION = '--'
HAS_IGNORED_ARGS = False
# Conditions that atest should exit without sending result to metrics.
EXIT_CODES_BEFORE_TEST = [
    ExitCode.ENV_NOT_SETUP,
    ExitCode.TEST_NOT_FOUND,
    ExitCode.OUTSIDE_ROOT,
    ExitCode.AVD_CREATE_FAILURE,
    ExitCode.AVD_INVALID_ARGS,
]

# Stdout print prefix for results directory. May be used in integration tests.
_RESULTS_DIR_PRINT_PREFIX = 'Atest results and logs directory: '
# Log prefix for dry-run run command. May be used in integration tests.
_DRY_RUN_COMMAND_LOG_PREFIX = 'Internal run command from dry-run: '


@dataclass
class Steps:
  """A Dataclass that stores steps and shows step assignments."""

  _build: bool
  _device_update: bool
  _install: bool
  _test: bool

  def has_build(self):
    """Return whether build is in steps."""
    return self._build

  def is_build_only(self):
    """Return whether build is the only one in steps."""
    return self._build and not any(
        (self._test, self._install, self._device_update)
    )

  def has_device_update(self):
    """Return whether device update is in steps."""
    return self._device_update

  def has_install(self):
    """Return whether install is in steps."""
    return self._install

  def has_test(self):
    """Return whether install is the only one in steps."""
    return self._test

  def is_test_only(self):
    """Return whether build is not in steps but test."""
    return self._test and not any(
        (self._build, self._install, self._device_update)
    )


def parse_steps(args: arg_parser.AtestArgParser) -> Steps:
  """Return Steps object.

  Args:
      args: an AtestArgParser object.

  Returns:
      Step object that stores the boolean of build, install and test.
  """
  # Implicitly running 'build', 'install' and 'test' when args.steps is None.
  if not args.steps:
    return Steps(True, args.update_device, True, True)
  build = constants.BUILD_STEP in args.steps
  test = constants.TEST_STEP in args.steps
  install = constants.INSTALL_STEP in args.steps
  if install and not test:
    atest_utils.print_and_log_warning(
        'Installing without test step is currently not '
        'supported; Atest will proceed testing!'
    )
    test = True
  return Steps(build, args.update_device, install, test)


def _get_args_from_config():
  """Get customized atest arguments in the config file.

  If the config has not existed yet, atest will initialize an example
  config file for it without any effective options.

  Returns:
      A list read from the config file.
  """
  _config = atest_utils.get_config_folder().joinpath('config')
  if not _config.parent.is_dir():
    _config.parent.mkdir(parents=True)
  args = []
  if not _config.is_file():
    with open(_config, 'w+', encoding='utf8') as cache:
      cache.write(constants.ATEST_EXAMPLE_ARGS)
    return args
  warning = 'Line {} contains {} and will be ignored.'
  print(
      '\n{} {}'.format(
          atest_utils.mark_cyan('Reading config:'),
          atest_utils.mark_yellow(_config),
      )
  )
  # pylint: disable=global-statement:
  global HAS_IGNORED_ARGS
  with open(_config, 'r', encoding='utf8') as cache:
    for entry in cache.readlines():
      # Strip comments.
      arg_in_line = entry.partition('#')[0].strip()
      # Strip test name/path.
      if arg_in_line.startswith('-'):
        # Process argument that contains whitespaces.
        # e.g. ["--serial foo"] -> ["--serial", "foo"]
        if len(arg_in_line.split()) > 1:
          # remove "--" to avoid messing up atest/tradefed commands.
          if END_OF_OPTION in arg_in_line.split():
            HAS_IGNORED_ARGS = True
            print(
                warning.format(
                    atest_utils.mark_yellow(arg_in_line), END_OF_OPTION
                )
            )
          args.extend(arg_in_line.split())
        else:
          if END_OF_OPTION == arg_in_line:
            HAS_IGNORED_ARGS = True
            print(
                warning.format(
                    atest_utils.mark_yellow(arg_in_line), END_OF_OPTION
                )
            )
          args.append(arg_in_line)
  return args


def _parse_args(argv: List[Any]) -> Tuple[argparse.ArgumentParser, List[str]]:
  """Parse command line arguments.

  Args:
      argv: A list of arguments.

  Returns:
      A tuple of an argparse.ArgumentParser class instance holding parsed args
  """
  # Store everything after '--' in custom_args.
  pruned_argv = argv
  custom_args_index = None
  if CUSTOM_ARG_FLAG in argv:
    custom_args_index = argv.index(CUSTOM_ARG_FLAG)
    pruned_argv = argv[:custom_args_index]
  args = arg_parser.create_atest_arg_parser().parse_args(pruned_argv)
  args.custom_args = []
  if custom_args_index is not None:
    for arg in argv[custom_args_index + 1 :]:
      logging.debug('Quoting regex argument %s', arg)
      args.custom_args.append(atest_utils.quote(arg))

  return args


def _configure_logging(verbose: bool, results_dir: str):
  """Configure the logger.

  Args:
      verbose: If true display DEBUG level logs on console.
      results_dir: A directory which stores the ATest execution information.
  """
  log_fmat = '%(asctime)s %(filename)s:%(lineno)s:%(levelname)s: %(message)s'
  date_fmt = '%Y-%m-%d %H:%M:%S'
  log_path = os.path.join(results_dir, 'atest.log')

  logger = logging.getLogger('')
  # Clear the handlers to prevent logging.basicConfig from being called twice.
  logger.handlers = []

  logging.basicConfig(
      filename=log_path, level=logging.DEBUG, format=log_fmat, datefmt=date_fmt
  )

  class _StreamToLogger:
    """A file like class to that redirect writes to a printer and logger."""

    def __init__(self, logger, log_level, printer):
      self._logger = logger
      self._log_level = log_level
      self._printer = printer
      self._buffers = []

    def write(self, buf: str) -> None:
      self._printer.write(buf)

      if len(buf) == 1 and buf[0] == '\n' and self._buffers:
        self._logger.log(self._log_level, ''.join(self._buffers))
        self._buffers.clear()
      else:
        self._buffers.append(buf)

    def flush(self) -> None:
      self._printer.flush()

  stdout_log_level = 25
  stderr_log_level = 45
  logging.addLevelName(stdout_log_level, 'STDOUT')
  logging.addLevelName(stderr_log_level, 'STDERR')
  sys.stdout = _StreamToLogger(logger, stdout_log_level, sys.stdout)
  sys.stderr = _StreamToLogger(logger, stderr_log_level, sys.stderr)


def _missing_environment_variables():
  """Verify the local environment has been set up to run atest.

  Returns:
      List of strings of any missing environment variables.
  """
  missing = list(
      filter(None, [x for x in EXPECTED_VARS if not os.environ.get(x)])
  )
  if missing:
    atest_utils.print_and_log_error(
        "Local environment doesn't appear to have been "
        'initialized. Did you remember to run lunch? Expected '
        'Environment Variables: %s.',
        missing,
    )
  return missing


def make_test_run_dir():
  """Make the test run dir in ATEST_RESULT_ROOT.

  Returns:
      A string of the dir path.
  """
  if not os.path.exists(constants.ATEST_RESULT_ROOT):
    os.makedirs(constants.ATEST_RESULT_ROOT)
  ctime = time.strftime(TEST_RUN_DIR_PREFIX, time.localtime())
  test_result_dir = tempfile.mkdtemp(
      prefix='%s_' % ctime, dir=constants.ATEST_RESULT_ROOT
  )
  print(_RESULTS_DIR_PRINT_PREFIX + test_result_dir)
  return test_result_dir


def get_extra_args(args):
  """Get extra args for test runners.

  Args:
      args: arg parsed object.

  Returns:
      Dict of extra args for test runners to utilize.
  """
  extra_args = {}
  if args.wait_for_debugger:
    extra_args[constants.WAIT_FOR_DEBUGGER] = None
  if not parse_steps(args).has_install():
    extra_args[constants.DISABLE_INSTALL] = None
  # The key and its value of the dict can be called via:
  # if args.aaaa:
  #     extra_args[constants.AAAA] = args.aaaa
  arg_maps = {
      'all_abi': constants.ALL_ABI,
      'annotation_filter': constants.ANNOTATION_FILTER,
      'bazel_arg': constants.BAZEL_ARG,
      'collect_tests_only': constants.COLLECT_TESTS_ONLY,
      'experimental_coverage': constants.COVERAGE,
      'custom_args': constants.CUSTOM_ARGS,
      'device_only': constants.DEVICE_ONLY,
      'disable_teardown': constants.DISABLE_TEARDOWN,
      'disable_upload_result': constants.DISABLE_UPLOAD_RESULT,
      'dry_run': constants.DRY_RUN,
      'host': constants.HOST,
      'instant': constants.INSTANT,
      'iterations': constants.ITERATIONS,
      'request_upload_result': constants.REQUEST_UPLOAD_RESULT,
      'bazel_mode_features': constants.BAZEL_MODE_FEATURES,
      'rerun_until_failure': constants.RERUN_UNTIL_FAILURE,
      'retry_any_failure': constants.RETRY_ANY_FAILURE,
      'serial': constants.SERIAL,
      'sharding': constants.SHARDING,
      'test_filter': constants.TEST_FILTER,
      'test_timeout': constants.TEST_TIMEOUT,
      'tf_debug': constants.TF_DEBUG,
      'tf_template': constants.TF_TEMPLATE,
      'user_type': constants.USER_TYPE,
      'verbose': constants.VERBOSE,
      'use_tf_min_base_template': constants.USE_TF_MIN_BASE_TEMPLATE,
  }
  not_match = [k for k in arg_maps if k not in vars(args)]
  if not_match:
    raise AttributeError(
        '%s object has no attribute %s' % (type(args).__name__, not_match)
    )
  extra_args.update({
      arg_maps.get(k): v for k, v in vars(args).items() if arg_maps.get(k) and v
  })
  return extra_args


def _validate_exec_mode(args, test_infos: list[TestInfo], host_tests=None):
  """Validate all test execution modes are not in conflict.

  Exit the program with INVALID_EXEC_MODE code if the desired is a host-side
  test but the given is a device-side test.

  If the given is a host-side test and not specified `args.host`, forcibly
  set `args.host` to True.

  Args:
      args: parsed args object.
      test_infos: a list of TestInfo objects.
      host_tests: True if all tests should be deviceless, False if all tests
        should be device tests. Default is set to None, which means tests can be
        either deviceless or device tests.
  """
  all_device_modes = {x.get_supported_exec_mode() for x in test_infos}
  err_msg = None
  # In the case of '$atest <device-only> --host', exit.
  if (host_tests or args.host) and constants.DEVICE_TEST in all_device_modes:
    device_only_tests = [
        x.test_name
        for x in test_infos
        if x.get_supported_exec_mode() == constants.DEVICE_TEST
    ]
    err_msg = (
        'Specified --host, but the following tests are device-only:\n  '
        + '\n  '.join(sorted(device_only_tests))
        + '\nPlease remove the  option when running device-only tests.'
    )
  # In the case of '$atest <host-only> <device-only> --host' or
  # '$atest <host-only> <device-only>', exit.
  if (
      constants.DEVICELESS_TEST in all_device_modes
      and constants.DEVICE_TEST in all_device_modes
  ):
    err_msg = 'There are host-only and device-only tests in command.'
  if host_tests is False and constants.DEVICELESS_TEST in all_device_modes:
    err_msg = 'There are host-only tests in command.'
  if err_msg:
    atest_utils.print_and_log_error(err_msg)
    metrics_utils.send_exit_event(ExitCode.INVALID_EXEC_MODE, logs=err_msg)
    sys.exit(ExitCode.INVALID_EXEC_MODE)
  # The 'adb' may not be available for the first repo sync or a clean build;
  # run `adb devices` in the build step again.
  if atest_utils.has_command('adb'):
    _validate_adb_devices(args, test_infos)
  # In the case of '$atest <host-only>', we add --host to run on host-side.
  # The option should only be overridden if `host_tests` is not set.
  if not args.host and host_tests is None:
    logging.debug('Appending "--host" for a deviceless test...')
    args.host = bool(constants.DEVICELESS_TEST in all_device_modes)


def _validate_adb_devices(args, test_infos):
  """Validate the availability of connected devices via adb command.

  Exit the program with error code if have device-only and host-only.

  Args:
      args: parsed args object.
      test_infos: TestInfo object.
  """
  # No need to check device availability if the user does not acquire to test.
  if not parse_steps(args).has_test():
    return
  if args.no_checking_device:
    return
  # No need to check local device availability if the device test is running
  # remotely.
  if args.bazel_mode_features and (
      bazel_mode.Features.EXPERIMENTAL_REMOTE_AVD in args.bazel_mode_features
  ):
    return
  all_device_modes = {x.get_supported_exec_mode() for x in test_infos}
  device_tests = [
      x.test_name
      for x in test_infos
      if x.get_supported_exec_mode() != constants.DEVICELESS_TEST
  ]
  # Only block testing if it is a device test.
  if constants.DEVICE_TEST in all_device_modes:
    if (
        not any((args.host, args.start_avd, args.acloud_create))
        and not atest_utils.get_adb_devices()
    ):
      err_msg = (
          f'Stop running test(s): {", ".join(device_tests)} require a device.'
      )
      atest_utils.colorful_print(err_msg, constants.RED)
      logging.debug(atest_utils.mark_red(constants.REQUIRE_DEVICES_MSG))
      metrics_utils.send_exit_event(ExitCode.DEVICE_NOT_FOUND, logs=err_msg)
      sys.exit(ExitCode.DEVICE_NOT_FOUND)


def _validate_tm_tests_exec_mode(
    args: argparse.Namespace,
    device_test_infos: List[test_info.TestInfo],
    host_test_infos: List[test_info.TestInfo],
):
  """Validate all test execution modes are not in conflict.

  Validate the tests' platform variant setting. For device tests, exit the
  program if any test is found for host-only. For host tests, exit the
  program if any test is found for device-only.

  Args:
      args: parsed args object.
      device_test_infos: TestInfo instances for device tests.
      host_test_infos: TestInfo instances for host tests.
  """

  # No need to verify device tests if atest command is set to only run host
  # tests.
  if device_test_infos and not args.host:
    _validate_exec_mode(args, device_test_infos, host_tests=False)
  if host_test_infos:
    _validate_exec_mode(args, host_test_infos, host_tests=True)


def _has_valid_test_mapping_args(args):
  """Validate test mapping args.

  Not all args work when running tests in TEST_MAPPING files. Validate the
  args before running the tests.

  Args:
      args: parsed args object.

  Returns:
      True if args are valid
  """
  is_test_mapping = atest_utils.is_test_mapping(args)
  if is_test_mapping:
    metrics.LocalDetectEvent(detect_type=DetectType.IS_TEST_MAPPING, result=1)
  else:
    metrics.LocalDetectEvent(detect_type=DetectType.IS_TEST_MAPPING, result=0)
  if not is_test_mapping:
    return True
  options_to_validate = [
      (args.annotation_filter, '--annotation-filter'),
  ]
  for arg_value, arg in options_to_validate:
    if arg_value:
      atest_utils.print_and_log_error(
          atest_utils.mark_red(OPTION_NOT_FOR_TEST_MAPPING.format(arg))
      )
      return False
  return True


def _validate_args(args):
  """Validate setups and args.

  Exit the program with error code if any setup or arg is invalid.

  Args:
      args: parsed args object.
  """
  if _missing_environment_variables():
    sys.exit(ExitCode.ENV_NOT_SETUP)
  if not _has_valid_test_mapping_args(args):
    sys.exit(ExitCode.INVALID_TM_ARGS)


def _print_module_info_from_module_name(mod_info, module_name):
  """print out the related module_info for a module_name.

  Args:
      mod_info: ModuleInfo object.
      module_name: A string of module.

  Returns:
      True if the module_info is found.
  """
  title_mapping = collections.OrderedDict()
  title_mapping[constants.MODULE_COMPATIBILITY_SUITES] = 'Compatibility suite'
  title_mapping[constants.MODULE_PATH] = 'Source code path'
  title_mapping[constants.MODULE_INSTALLED] = 'Installed path'
  target_module_info = mod_info.get_module_info(module_name)
  is_module_found = False
  if target_module_info:
    atest_utils.colorful_print(module_name, constants.GREEN)
    for title_key in title_mapping:
      atest_utils.colorful_print(
          '\t%s' % title_mapping[title_key], constants.CYAN
      )
      for info_value in target_module_info[title_key]:
        print('\t\t{}'.format(info_value))
    is_module_found = True
  return is_module_found


def _print_deprecation_warning(arg_to_deprecate: str):
  """For features that are up for deprecation in the near future, print a message

  to alert the user about the upcoming deprecation.

  Args:
      arg_to_deprecate: the arg with which the to-be-deprecated feature is
        called.
  """
  args_to_deprecation_info = {
      # arg_to_deprecate : (deprecation timeframe, additional info for users)
      '--info': ('is deprecated.', '\nUse CodeSearch or `gomod` instead.')
  }

  warning_message = (
      f'\nWARNING: The `{arg_to_deprecate}` feature '
      + ' '.join(args_to_deprecation_info[arg_to_deprecate])
      + '\nPlease file a bug or feature request to the Atest team if you have'
      ' any concerns.'
  )
  atest_utils.colorful_print(warning_message, constants.RED)


def is_from_test_mapping(test_infos):
  """Check that the test_infos came from TEST_MAPPING files.

  Args:
      test_infos: A set of TestInfos.

  Returns:
      True if the test infos are from TEST_MAPPING files.
  """
  return list(test_infos)[0].from_test_mapping


def _split_test_mapping_tests(test_infos):
  """Split Test Mapping tests into 2 groups: device tests and host tests.

  Args:
      test_infos: A set of TestInfos.

  Returns:
      A tuple of (device_test_infos, host_test_infos), where
      device_test_infos: A set of TestInfos for tests that require device.
      host_test_infos: A set of TestInfos for tests that do NOT require
          device.
  """
  assert is_from_test_mapping(test_infos)
  host_test_infos = {info for info in test_infos if info.host}
  device_test_infos = {info for info in test_infos if not info.host}
  return device_test_infos, host_test_infos


# pylint: disable=too-many-locals
def _run_test_mapping_tests(
    test_type_to_invocations: Dict[str, List[TestRunnerInvocation]],
    extra_args: Dict[str, Any],
) -> ExitCode:
  """Run all tests in TEST_MAPPING files.

  Args:
      test_type_to_invocations: A dict mapping test runner invocations to test
        types.
      extra_args: A dict of extra args for others to utilize.

  Returns:
      Exit code.
  """

  test_results = []
  for test_type, invocations in test_type_to_invocations.items():
    tests = list(
        itertools.chain.from_iterable(i.test_infos for i in invocations)
    )
    if not tests:
      continue
    header = RUN_HEADER_FMT % {TEST_COUNT: len(tests), TEST_TYPE: test_type}
    atest_utils.colorful_print(header, constants.MAGENTA)
    logging.debug('\n'.join([str(info) for info in tests]))

    reporter = result_reporter.ResultReporter(
        collect_only=extra_args.get(constants.COLLECT_TESTS_ONLY),
        wait_for_debugger=atest_configs.GLOBAL_ARGS.wait_for_debugger,
    )
    reporter.print_starting_text()

    tests_exit_code = ExitCode.SUCCESS
    for invocation in invocations:
      tests_exit_code |= invocation.run_all_tests(reporter)

    atest_execution_info.AtestExecutionInfo.result_reporters.append(reporter)
    test_results.append((tests_exit_code, reporter, test_type))

  all_tests_exit_code = ExitCode.SUCCESS
  failed_tests = []
  for tests_exit_code, reporter, test_type in test_results:
    atest_utils.colorful_print(
        RESULT_HEADER_FMT % {TEST_TYPE: test_type}, constants.MAGENTA
    )
    result = tests_exit_code | reporter.print_summary()
    if result:
      failed_tests.append(test_type)
    all_tests_exit_code |= result

  # List failed tests at the end as a reminder.
  if failed_tests:
    atest_utils.colorful_print(
        atest_utils.delimiter('=', 30, prenl=1), constants.YELLOW
    )
    atest_utils.colorful_print('\nFollowing tests failed:', constants.MAGENTA)
    for failure in failed_tests:
      atest_utils.colorful_print(failure, constants.RED)

  return all_tests_exit_code


def _dry_run(results_dir, extra_args, test_infos, mod_info):
  """Only print the commands of the target tests rather than running them in

  actual.

  Args:
      results_dir: Path for saving atest logs.
      extra_args: Dict of extra args for test runners to utilize.
      test_infos: A list of TestInfos.
      mod_info: ModuleInfo object.

  Returns:
      A successful exit code.
  """
  all_run_cmds = []
  for test_runner, tests in test_runner_handler.group_tests_by_test_runners(
      test_infos
  ):
    runner = test_runner(results_dir, mod_info=mod_info, extra_args=extra_args)
    run_cmds = runner.generate_run_commands(tests, extra_args)
    for run_cmd in run_cmds:
      all_run_cmds.append(run_cmd)
      logging.debug(_DRY_RUN_COMMAND_LOG_PREFIX + run_cmd)
      print(
          'Would run test via command: %s' % (atest_utils.mark_green(run_cmd))
      )
  return ExitCode.SUCCESS


def _print_testable_modules(mod_info, suite):
  """Print the testable modules for a given suite.

  Args:
      mod_info: ModuleInfo object.
      suite: A string of suite name.
  """
  testable_modules = mod_info.get_testable_modules(suite)
  print(
      '\n%s'
      % atest_utils.mark_cyan(
          '%s Testable %s modules' % (len(testable_modules), suite)
      )
  )
  print(atest_utils.delimiter('-'))
  for module in sorted(testable_modules):
    print('\t%s' % module)


def _is_inside_android_root():
  """Identify whether the cwd is inside of Android source tree.

  Returns:
      False if the cwd is outside of the source tree, True otherwise.
  """
  build_top = os.getenv(constants.ANDROID_BUILD_TOP, ' ')
  return build_top in os.getcwd()


def _non_action_validator(args: argparse.ArgumentParser):
  """Method for non-action arguments such as --version, --history,

  --latest_result, etc.

  Args:
      args: An argparse.ArgumentParser object.
  """
  if not _is_inside_android_root():
    atest_utils.colorful_print(
        '\nAtest must always work under ${}!'.format(
            constants.ANDROID_BUILD_TOP
        ),
        constants.RED,
    )
    sys.exit(ExitCode.OUTSIDE_ROOT)
  if args.version:
    print(atest_utils.get_atest_version())
    sys.exit(ExitCode.SUCCESS)
  if args.history:
    atest_execution_info.print_test_result(
        constants.ATEST_RESULT_ROOT, args.history
    )
    sys.exit(ExitCode.SUCCESS)
  if args.latest_result:
    atest_execution_info.print_test_result_by_path(constants.LATEST_RESULT_FILE)
    sys.exit(ExitCode.SUCCESS)


def _exclude_modules_in_targets(build_targets):
  """Method that excludes MODULES-IN-* targets.

  Args:
      build_targets: A set of build targets.

  Returns:
      A set of build targets that excludes MODULES-IN-*.
  """
  shrank_build_targets = build_targets.copy()
  logging.debug(
      'Will exclude all "%s*" from the build targets.', constants.MODULES_IN
  )
  for target in build_targets:
    if target.startswith(constants.MODULES_IN):
      logging.debug('Ignore %s.', target)
      shrank_build_targets.remove(target)
  return shrank_build_targets


# pylint: disable=protected-access
def need_rebuild_module_info(args: arg_parser.AtestArgParser) -> bool:
  """Method that tells whether we need to rebuild module-info.json or not.

  Args:
      args: an AtestArgParser object.

  Returns:
      True for forcely/smartly rebuild, otherwise False without rebuilding.
  """
  # +-----------------+
  # | Explicitly pass |  yes
  # |    '--test'     +-------> False (won't rebuild)
  # +--------+--------+
  #          | no
  #          V
  # +-------------------------+
  # | Explicitly pass         |  yes
  # | '--rebuild-module-info' +-------> True (forcely rebuild)
  # +--------+----------------+
  #          | no
  #          V
  # +-------------------+
  # |    Build files    |  no
  # | integrity is good +-------> True (smartly rebuild)
  # +--------+----------+
  #          | yes
  #          V
  #        False (won't rebuild)
  if not parse_steps(args).has_build():
    logging.debug('"--test" mode detected, will not rebuild module-info.')
    return False
  if args.rebuild_module_info:
    msg = (
        f'`{constants.REBUILD_MODULE_INFO_FLAG}` is no longer needed '
        f'since Atest can smartly rebuild {module_info._MODULE_INFO} '
        r'only when needed.'
    )
    atest_utils.colorful_print(msg, constants.YELLOW)
    return True
  logging.debug('Examinating the consistency of build files...')
  if not atest_utils.build_files_integrity_is_ok():
    logging.debug('Found build files were changed.')
    return True
  return False


def need_run_index_targets(args: argparse.ArgumentParser):
  """Method that determines whether Atest need to run index_targets or not.

  The decision flow is as follows: If no build is required, returns False.
  Otherwise, if some index files are missing, returns True. Otherwise, if
  some arguments that doesn't require indexing is present, returns False.
  Otherwise, returns True.

  Args:
      args: An argparse.ArgumentParser object.

  Returns:
      True when none of the above conditions were found.
  """
  has_build_step = parse_steps(args).has_build()
  if not has_build_step:
    logging.debug("Skip indexing because there's no build required.")
    return False

  if not indexing.Indices().has_all_indices():
    logging.debug(
        'Indexing targets is required because some index files do not exist.'
    )
    return True

  no_indexing_args = (
      args.dry_run,
      args.list_modules,
  )
  if any(no_indexing_args):
    logging.debug('Skip indexing for no_indexing_args=%s.', no_indexing_args)
    return False

  return True


def set_build_output_mode(mode: atest_utils.BuildOutputMode):
  """Update environment variable dict accordingly to args.build_output."""
  # Changing this variable does not retrigger builds.
  atest_utils.update_build_env(
      {'ANDROID_QUIET_BUILD': 'true', 'BUILD_OUTPUT_MODE': mode.value}
  )


def get_device_count_config(test_infos, mod_info):
  """Get the amount of desired devices from the test config.

  Args:
      test_infos: A set of TestInfo instances.
      mod_info: ModuleInfo object.

  Returns: the count of devices in test config. If there are more than one
           configs, return the maximum.
  """
  max_count = 0
  for tinfo in test_infos:
    test_config, _ = test_finder_utils.get_test_config_and_srcs(tinfo, mod_info)
    if test_config:
      devices = atest_utils.get_config_device(test_config)
      if devices:
        max_count = max(len(devices), max_count)
  return max_count


def _send_start_event(argv: List[Any], tests: List[str]):
  """Send AtestStartEvent to metrics"""
  os_pyver = (
      f'{platform.platform()}:{platform.python_version()}/'
      f'{atest_utils.get_manifest_branch(True)}:'
      f'{atest_utils.get_atest_version()}'
  )
  metrics.AtestStartEvent(
      command_line=' '.join(argv),
      test_references=tests,
      cwd=os.getcwd(),
      os=os_pyver,
  )


def _get_acloud_proc_and_log(
    args: argparse.ArgumentParser, results_dir: str
) -> Tuple[Any, Any]:
  """Return tuple of acloud process ID and report file."""
  if any((args.acloud_create, args.start_avd)):
    logging.debug('Creating acloud or avd.')
    return avd.acloud_create_validator(results_dir, args)
  return None, None


def has_set_sufficient_devices(
    required_amount: int, serial: List[str] = None
) -> bool:
  """Detect whether sufficient device serial is set for test."""
  given_amount = len(serial) if serial else 0
  # Only check when both given_amount and required_amount are non zero.
  if all((given_amount, required_amount)):
    # Base on TF rules, given_amount can be greater than or equal to
    # required_amount.
    if required_amount > given_amount:
      atest_utils.colorful_print(
          f'The test requires {required_amount} devices, '
          f'but {given_amount} were given.',
          constants.RED,
      )
      return False
  return True


def setup_metrics_tool_name(no_metrics: bool = False):
  """Setup tool_name and sub_tool_name for MetricsBase."""
  if (
      not no_metrics
      and metrics_base.MetricsBase.user_type == metrics_base.INTERNAL_USER
  ):
    metrics_utils.print_data_collection_notice()

    USER_FROM_TOOL = os.getenv(constants.USER_FROM_TOOL)
    metrics_base.MetricsBase.tool_name = (
        USER_FROM_TOOL if USER_FROM_TOOL else constants.TOOL_NAME
    )

    USER_FROM_SUB_TOOL = os.getenv(constants.USER_FROM_SUB_TOOL)
    metrics_base.MetricsBase.sub_tool_name = (
        USER_FROM_SUB_TOOL if USER_FROM_SUB_TOOL else constants.SUB_TOOL_NAME
    )


# pylint: disable=too-many-statements
# pylint: disable=too-many-branches
# pylint: disable=too-many-return-statements
def _main(
    argv: List[Any],
    results_dir: str,
    args: argparse.Namespace,
    banner_printer: banner.BannerPrinter,
):
  """Entry point of atest script.

  Args:
      argv: A list of arguments.
      results_dir: A directory which stores the ATest execution information.
      args: An argparse.Namespace class instance holding parsed args.
      banner_printer: A BannerPrinter object used to collect banners and print
        banners at the end of this invocation.

  Returns:
      Exit code.
  """
  _begin_time = time.time()
  logging.debug(
      'Running atest script with argv %s, results_dir %s, args %s.',
      argv,
      results_dir,
      args,
  )

  # Sets coverage environment variables.
  if args.experimental_coverage:
    atest_utils.update_build_env(coverage.build_env_vars())
  set_build_output_mode(args.build_output)

  _validate_args(args)
  metrics_utils.get_start_time()
  _send_start_event(argv, args.tests)
  _non_action_validator(args)

  proc_acloud, report_file = _get_acloud_proc_and_log(args, results_dir)
  is_clean = not os.path.exists(
      os.environ.get(constants.ANDROID_PRODUCT_OUT, '')
  )

  # Run Test Mapping or coverage by no-bazel-mode.
  if atest_utils.is_test_mapping(args) or args.experimental_coverage:
    logging.debug('Running test mapping or coverage, disabling bazel mode.')
    atest_utils.colorful_print(
        'Not running using bazel-mode.', constants.YELLOW
    )
    args.bazel_mode = False

  proc_idx = atest_utils.start_threading(lambda: print)
  # Do not index targets while the users intend to dry-run tests.
  if need_run_index_targets(args):
    logging.debug('Starting to index targets in a background thread.')
    proc_idx = atest_utils.start_threading(
        indexing.index_targets,
        daemon=True,
    )
  smart_rebuild = need_rebuild_module_info(args)
  logging.debug('need_rebuild_module_info returned %s', smart_rebuild)

  mod_info = module_info.load(
      force_build=smart_rebuild,
      sqlite_module_cache=args.sqlite_module_cache,
  )
  logging.debug('Obtained module info object: %s', mod_info)

  translator = cli_translator.CLITranslator(
      mod_info=mod_info,
      print_cache_msg=not args.clear_cache,
      bazel_mode_enabled=args.bazel_mode,
      host=args.host,
      bazel_mode_features=args.bazel_mode_features,
  )
  if args.list_modules:
    _print_testable_modules(mod_info, args.list_modules)
    return ExitCode.SUCCESS
  test_infos = set()
  # (b/242567487) index_targets may finish after cli_translator; to
  # mitigate the overhead, the main waits until it finished when no index
  # files are available (e.g. fresh repo sync)
  if proc_idx.is_alive() and not indexing.Indices().has_all_indices():
    proc_idx.join()
  find_start = time.time()
  test_infos = translator.translate(args)

  # Only check for sufficient devices if not dry run.
  args.device_count_config = get_device_count_config(test_infos, mod_info)
  if not args.dry_run and not has_set_sufficient_devices(
      args.device_count_config, args.serial
  ):
    return ExitCode.INSUFFICIENT_DEVICES

  find_duration = time.time() - find_start
  if not test_infos:
    return ExitCode.TEST_NOT_FOUND

  test_execution_plan = _create_test_execution_plan(
      test_infos=test_infos,
      results_dir=results_dir,
      mod_info=mod_info,
      args=args,
      dry_run=args.dry_run,
  )

  extra_args = test_execution_plan.extra_args

  build_targets = test_execution_plan.required_build_targets()

  # Remove MODULE-IN-* from build targets by default.
  if not args.use_modules_in:
    build_targets = _exclude_modules_in_targets(build_targets)

  if args.dry_run:
    return _dry_run(results_dir, extra_args, test_infos, mod_info)

  steps = parse_steps(args)
  device_update_method = _configure_update_method(
      steps=steps,
      plan=test_execution_plan,
      update_modules=set(args.update_modules or []),
      banner_printer=banner_printer,
  )

  if build_targets and steps.has_build():
    if args.experimental_coverage:
      build_targets.update(coverage.build_modules())

    # Add module-info.json target to the list of build targets to keep the
    # file up to date.
    build_targets.add(module_info.get_module_info_target())

    build_targets |= device_update_method.dependencies()

    # Add the -jx as a build target if user specify it.
    if args.build_j:
      build_targets.add(f'-j{args.build_j}')

    build_start = time.time()
    success = atest_utils.build(build_targets)
    build_duration = time.time() - build_start
    metrics.BuildFinishEvent(
        duration=metrics_utils.convert_duration(build_duration),
        success=success,
        targets=build_targets,
    )
    metrics.LocalDetectEvent(
        detect_type=DetectType.BUILD_TIME_PER_TARGET,
        result=int(build_duration / len(build_targets)),
    )
    rebuild_module_info = DetectType.NOT_REBUILD_MODULE_INFO
    if is_clean:
      rebuild_module_info = DetectType.CLEAN_BUILD
    elif args.rebuild_module_info:
      rebuild_module_info = DetectType.REBUILD_MODULE_INFO
    elif smart_rebuild:
      rebuild_module_info = DetectType.SMART_REBUILD_MODULE_INFO
    metrics.LocalDetectEvent(
        detect_type=rebuild_module_info, result=int(build_duration)
    )
    if not success:
      return ExitCode.BUILD_FAILURE
    if proc_acloud:
      proc_acloud.join()
      status = avd.probe_acloud_status(
          report_file, find_duration + build_duration
      )
      if status != 0:
        return status
    # After build step 'adb' command will be available, and stop forward to
    # Tradefed if the tests require a device.
    _validate_adb_devices(args, test_infos)

  device_update_method.update(extra_args.get(constants.SERIAL, []))

  tests_exit_code = ExitCode.SUCCESS
  test_start = time.time()
  if steps.has_test():
    # Only send duration to metrics when no --build.
    if not steps.has_build():
      _init_and_find = time.time() - _begin_time
      logging.debug('Initiation and finding tests took %ss', _init_and_find)
      metrics.LocalDetectEvent(
          detect_type=DetectType.INIT_AND_FIND_MS,
          result=int(_init_and_find * 1000),
      )

    tests_exit_code = test_execution_plan.execute()

    if args.experimental_coverage:
      coverage.generate_coverage_report(
          results_dir,
          test_infos,
          mod_info,
          extra_args.get(constants.HOST, False),
          args.code_under_test,
      )

  metrics.RunTestsFinishEvent(
      duration=metrics_utils.convert_duration(time.time() - test_start)
  )
  preparation_time = atest_execution_info.preparation_time(test_start)
  if preparation_time:
    # Send the preparation time only if it's set.
    metrics.RunnerFinishEvent(
        duration=metrics_utils.convert_duration(preparation_time),
        success=True,
        runner_name=constants.TF_PREPARATION,
        test=[],
    )
  if tests_exit_code != ExitCode.SUCCESS:
    tests_exit_code = ExitCode.TEST_FAILURE

  return tests_exit_code


def _configure_update_method(
    *,
    steps: Steps,
    plan: TestExecutionPlan,
    update_modules: set[str],
    banner_printer: banner.BannerPrinter,
) -> device_update.DeviceUpdateMethod:

  requires_device_update = plan.requires_device_update()

  if not steps.has_device_update():
    if requires_device_update:
      banner_printer.register(
          'Tips: If your test requires device update, consider '
          'http://go/atest-single-command to simplify your workflow!'
      )
    return device_update.NoopUpdateMethod()

  if not requires_device_update:
    atest_utils.colorful_print(
        '\nWarning: Device update ignored because it is not required by '
        'tests in this invocation.',
        constants.YELLOW,
    )
    return device_update.NoopUpdateMethod()

  return device_update.AdeviceUpdateMethod(targets=update_modules)


def _create_test_execution_plan(
    *,
    test_infos: List[test_info.TestInfo],
    results_dir: str,
    mod_info: module_info.ModuleInfo,
    args: argparse.Namespace,
    dry_run: bool,
) -> TestExecutionPlan:
  """Creates a plan to execute the tests.

  Args:
      test_infos: A list of instances of TestInfo.
      results_dir: A directory which stores the ATest execution information.
      mod_info: An instance of ModuleInfo.
      args: An argparse.Namespace instance holding parsed args.
      dry_run: A boolean of whether this invocation is a dry run.

  Returns:
      An instance of TestExecutionPlan.
  """

  if is_from_test_mapping(test_infos):
    return TestMappingExecutionPlan.create(
        test_infos=test_infos,
        results_dir=results_dir,
        mod_info=mod_info,
        args=args,
    )

  return TestModuleExecutionPlan.create(
      test_infos=test_infos,
      results_dir=results_dir,
      mod_info=mod_info,
      args=args,
      dry_run=dry_run,
  )


class TestExecutionPlan(ABC):
  """Represents how an Atest invocation's tests will execute."""

  def __init__(
      self,
      *,
      extra_args: Dict[str, Any],
  ):
    self._extra_args = extra_args

  @property
  def extra_args(self) -> Dict[str, Any]:
    return self._extra_args

  @abstractmethod
  def execute(self) -> ExitCode:
    """Executes all test runner invocations in this plan."""

  @abstractmethod
  def required_build_targets(self) -> Set[str]:
    """Returns the list of build targets required by this plan."""

  @abstractmethod
  def requires_device_update(self) -> bool:
    """Checks whether this plan requires device update."""


class TestMappingExecutionPlan(TestExecutionPlan):
  """A plan to execute Test Mapping tests."""

  def __init__(
      self,
      *,
      test_type_to_invocations: Dict[str, List[TestRunnerInvocation]],
      extra_args: Dict[str, Any],
  ):
    super().__init__(extra_args=extra_args)
    self._test_type_to_invocations = test_type_to_invocations

  @staticmethod
  def create(
      *,
      test_infos: List[test_info.TestInfo],
      results_dir: str,
      mod_info: module_info.ModuleInfo,
      args: argparse.Namespace,
  ) -> TestMappingExecutionPlan:
    """Creates an instance of TestMappingExecutionPlan.

    Args:
        test_infos: A list of instances of TestInfo.
        results_dir: A directory which stores the ATest execution information.
        mod_info: An instance of ModuleInfo.
        args: An argparse.Namespace instance holding parsed args.

    Returns:
        An instance of TestMappingExecutionPlan.
    """

    device_test_infos, host_test_infos = _split_test_mapping_tests(test_infos)
    _validate_tm_tests_exec_mode(args, device_test_infos, host_test_infos)
    extra_args = get_extra_args(args)

    # TODO: change to another approach that put constants.CUSTOM_ARGS in the
    # end of command to make sure that customized args can override default
    # options.
    # For TEST_MAPPING, set timeout to 600000ms.
    custom_timeout = False
    for custom_args in args.custom_args:
      if '-timeout' in custom_args:
        custom_timeout = True

    if args.test_timeout is None and not custom_timeout:
      extra_args.update({constants.TEST_TIMEOUT: 600000})
      logging.debug(
          'Set test timeout to %sms to align it in TEST_MAPPING.',
          extra_args.get(constants.TEST_TIMEOUT),
      )

    def create_invocations(runner_extra_args, runner_test_infos):
      return test_runner_handler.create_test_runner_invocations(
          test_infos=runner_test_infos,
          results_dir=results_dir,
          mod_info=mod_info,
          extra_args=runner_extra_args,
          minimal_build=args.minimal_build,
      )

    test_type_to_invocations = collections.OrderedDict()
    if extra_args.get(constants.DEVICE_ONLY):
      atest_utils.colorful_print(
          'Option `--device-only` specified. Skip running deviceless tests.',
          constants.MAGENTA,
      )
    else:
      # `host` option needs to be set to True to run host side tests.
      host_extra_args = extra_args.copy()
      host_extra_args[constants.HOST] = True
      test_type_to_invocations.setdefault(HOST_TESTS, []).extend(
          create_invocations(host_extra_args, host_test_infos)
      )

    if extra_args.get(constants.HOST):
      atest_utils.colorful_print(
          'Option `--host` specified. Skip running device tests.',
          constants.MAGENTA,
      )
    else:
      test_type_to_invocations.setdefault(DEVICE_TESTS, []).extend(
          create_invocations(extra_args, device_test_infos)
      )

    return TestMappingExecutionPlan(
        test_type_to_invocations=test_type_to_invocations,
        extra_args=extra_args,
    )

  def requires_device_update(self) -> bool:
    return _requires_device_update(
        [i for invs in self._test_type_to_invocations.values() for i in invs]
    )

  def required_build_targets(self) -> Set[str]:
    build_targets = set()
    for invocation in itertools.chain.from_iterable(
        self._test_type_to_invocations.values()
    ):
      build_targets |= invocation.get_test_runner_reqs()

    return build_targets

  def execute(self) -> ExitCode:
    return _run_test_mapping_tests(
        self._test_type_to_invocations, self.extra_args
    )


class TestModuleExecutionPlan(TestExecutionPlan):
  """A plan to execute the test modules explicitly passed on the command-line."""

  def __init__(
      self,
      *,
      test_runner_invocations: List[TestRunnerInvocation],
      extra_args: Dict[str, Any],
  ):
    super().__init__(extra_args=extra_args)
    self._test_runner_invocations = test_runner_invocations

  @staticmethod
  def create(
      *,
      test_infos: List[test_info.TestInfo],
      results_dir: str,
      mod_info: module_info.ModuleInfo,
      args: argparse.Namespace,
      dry_run: bool,
  ) -> TestModuleExecutionPlan:
    """Creates an instance of TestModuleExecutionPlan.

    Args:
        test_infos: A list of instances of TestInfo.
        results_dir: A directory which stores the ATest execution information.
        mod_info: An instance of ModuleInfo.
        args: An argparse.Namespace instance holding parsed args.
        dry_run: A boolean of whether this invocation is a dry run.

    Returns:
        An instance of TestModuleExecutionPlan.
    """

    if not dry_run:
      _validate_exec_mode(args, test_infos)

    # _validate_exec_mode appends --host automatically when pure
    # host-side tests, so re-parsing extra_args is a must.
    extra_args = get_extra_args(args)

    invocations = test_runner_handler.create_test_runner_invocations(
        test_infos=test_infos,
        results_dir=results_dir,
        mod_info=mod_info,
        extra_args=extra_args,
        minimal_build=args.minimal_build,
    )

    return TestModuleExecutionPlan(
        test_runner_invocations=invocations,
        extra_args=extra_args,
    )

  def requires_device_update(self) -> bool:
    return _requires_device_update(self._test_runner_invocations)

  def required_build_targets(self) -> Set[str]:
    build_targets = set()
    for test_runner_invocation in self._test_runner_invocations:
      build_targets |= test_runner_invocation.get_test_runner_reqs()

    return build_targets

  def execute(self) -> ExitCode:

    reporter = result_reporter.ResultReporter(
        collect_only=self.extra_args.get(constants.COLLECT_TESTS_ONLY),
        wait_for_debugger=atest_configs.GLOBAL_ARGS.wait_for_debugger,
    )
    reporter.print_starting_text()

    exit_code = ExitCode.SUCCESS
    for invocation in self._test_runner_invocations:
      exit_code |= invocation.run_all_tests(reporter)

    atest_execution_info.AtestExecutionInfo.result_reporters.append(reporter)
    return reporter.print_summary() | exit_code


def _requires_device_update(invocations: List[TestRunnerInvocation]) -> bool:
  """Checks if any invocation requires device update."""
  return any(i.requires_device_update() for i in invocations)


if __name__ == '__main__':
  results_dir = make_test_run_dir()
  if END_OF_OPTION in sys.argv:
    end_position = sys.argv.index(END_OF_OPTION)
    final_args = [
        *sys.argv[1:end_position],
        *_get_args_from_config(),
        *sys.argv[end_position:],
    ]
  else:
    final_args = [*sys.argv[1:], *_get_args_from_config()]
  if final_args != sys.argv[1:]:
    print(
        'The actual cmd will be: \n\t{}\n'.format(
            atest_utils.mark_cyan('atest ' + ' '.join(final_args))
        )
    )
    metrics.LocalDetectEvent(detect_type=DetectType.ATEST_CONFIG, result=1)
    if HAS_IGNORED_ARGS:
      atest_utils.colorful_print(
          'Please correct the config and try again.', constants.YELLOW
      )
      sys.exit(ExitCode.EXIT_BEFORE_MAIN)
  else:
    metrics.LocalDetectEvent(detect_type=DetectType.ATEST_CONFIG, result=0)

  args = _parse_args(final_args)
  atest_configs.GLOBAL_ARGS = args
  _configure_logging(args.verbose, results_dir)

  logging.debug(
      'Start of atest run. sys.argv: %s, final_args: %s', sys.argv, final_args
  )

  banner_printer = banner.BannerPrinter.create()

  with atest_execution_info.AtestExecutionInfo(
      final_args, results_dir, atest_configs.GLOBAL_ARGS
  ) as result_file:
    setup_metrics_tool_name(atest_configs.GLOBAL_ARGS.no_metrics)

    exit_code = _main(
        final_args,
        results_dir,
        atest_configs.GLOBAL_ARGS,
        banner_printer,
    )
    detector = bug_detector.BugDetector(final_args, exit_code)
    if exit_code not in EXIT_CODES_BEFORE_TEST:
      metrics.LocalDetectEvent(
          detect_type=DetectType.BUG_DETECTED, result=detector.caught_result
      )
      if result_file:
        print("Run 'atest --history' to review test result history.")

  banner_printer.print()

  sys.exit(exit_code)