blob: 9d03ff9515fe8d4fce500185c933f5e3be422f80 [file] [log] [blame]
#!/usr/bin/env python3
# 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.
"""Evaluate ChromeOS tast tests."""
from __future__ import annotations
import json
import logging
import os
import re
import shutil
import subprocess
import sys
from bisect_kit import bisector_cli
from bisect_kit import catapult_util
from bisect_kit import cli
from bisect_kit import common
from bisect_kit import core
from bisect_kit import cros_lab_util
from bisect_kit import cros_util
from bisect_kit import errors
from bisect_kit import util
logger = logging.getLogger(__name__)
OLD = 'old'
NEW = 'new'
SKIP = 'skip'
FATAL = 'fatal'
EXIT_CODE_MAP = {
OLD: cli.EXIT_CODE_OLD,
NEW: cli.EXIT_CODE_NEW,
SKIP: cli.EXIT_CODE_SKIP,
FATAL: cli.EXIT_CODE_FATAL,
}
def create_argument_parser():
parents = [cli.create_session_optional_parser()]
parser = cli.ArgumentParser(description=__doc__, parents=parents)
parser.add_argument(
'--rich-result',
action='store_true',
help='Instead of mere exit code, output detailed information in json',
)
parser.add_argument(
'dut',
nargs='?',
type=cli.argtype_notempty,
metavar='DUT',
default=os.environ.get('DUT', ''),
)
parser.add_argument(
'--chromeos-root',
type=cli.argtype_dir_path,
metavar='CHROMEOS_ROOT',
default=os.environ.get('CHROMEOS_ROOT', ''),
help='ChromeOS tree root',
)
parser.add_argument(
'--prebuilt',
action='store_true',
help='Run tast using existing server prebuilt package if specified; '
'otherwise use the default one',
)
parser.add_argument(
'--tast-build',
action='store_true',
help='Build tast test bundle (-build=true) if specified; '
'default is using prebuilt bundle on the DUT',
)
parser.add_argument(
'--with-private-bundles',
action='store_true',
help='Whether search tests in private bundles or not',
)
parser.add_argument(
'--reboot-before-test',
action='store_true',
help='Reboot before test run',
)
parser.add_argument(
'--args',
help='Extra variables passed to `tast -var`, as "key=value"',
type=cli.argtype_key_value,
action='append',
default=[],
)
group = parser.add_argument_group(title='Options for normal autotest tests')
group.add_argument(
'--test-name',
required=True,
help='Test name, like "video_VideoDecodeAccelerator.h264"',
)
group.add_argument(
'--fail-to-pass',
action='store_true',
help='For functional tests: old behavior is FAIL and new behavior is '
'PASS; If not specified, default = old behavior is PASS and new '
'behavior is FAIL',
)
group.add_argument(
'--consider-test-crash-as-failure',
action='store_true',
help='Consider test crashes as failure instead of skip',
)
group.add_argument(
'--metric',
help='Metric name of performance test; example: '
'"cheets_SystemRawImageSize"',
)
return parser
def prepare_to_run_test(opts):
if opts.reboot_before_test:
cros_util.reboot(
opts.dut, force_reboot_callback=cros_lab_util.reboot_via_servo
)
def to_forwarded_host(host: str) -> tuple[str, str]:
# Returns (dut address in chroot, dut address needs to be forwarded)
if cros_lab_util.is_dut_needs_forwarded(host):
return cros_util.FORWARDED_DUT_HOST, host
return host, ''
def tast_cmd(opts, cmd, *args):
flags = []
if opts.prebuilt:
tast_dir = os.path.join(
cros_util.chromeos_root_inside_chroot, cros_util.prebuilt_tast_dir
)
tast_bin = os.path.join(tast_dir, 'tast')
flags += [
'-remotebundledir',
os.path.join(tast_dir, 'bundles', 'remote'),
]
flags += ['-remotedatadir', os.path.join(tast_dir, 'data')]
flags += ['-remoterunner', os.path.join(tast_dir, 'remote_test_runner')]
flags += [
'-defaultvarsdir',
os.path.join(tast_dir, 'vars', 'private'),
'-defaultvarsdir',
os.path.join(tast_dir, 'vars', 'public'),
]
else:
tast_bin = 'tast'
if cmd == 'run':
flags += ['-var=%s=%s' % x for x in opts.args]
# TODO(zjchang): ensure the tast version for buildbucket builds is correct
if not opts.tast_build:
flags.append('-build=false')
if opts.with_private_bundles:
flags.append('-downloadprivatebundles=true')
return [tast_bin, '-verbose', cmd] + flags + list(args)
def get_tast_bundle_info(opts, pattern=None):
dut, forward_host = to_forwarded_host(opts.dut)
bundles = ['cros']
if opts.with_private_bundles:
bundles.append('crosint')
args = [dut]
if pattern:
args.append(pattern)
result = {}
for bundle in bundles:
cmd = tast_cmd(opts, 'list', '-json', '-buildbundle=' + bundle, *args)
try:
json_text = cros_util.cros_sdk(
opts.chromeos_root,
*cmd,
log_stdout=False,
forward_host=forward_host,
)
except subprocess.CalledProcessError as e:
raise errors.ExternalError(
'failed to get tast bundle info, assume it is temporary: %s' % e
)
for entry in json.loads(json_text):
result[entry['name']] = bundle
return result
def run_test(opts, bundle):
"""Runs an autotest test.
Args:
opts: An argparse.Namespace to hold command line arguments.
bundle: tast's test bundle
Returns:
path of test result (outside chroot)
"""
# Set results dir inside source tree, so it's easier to access them outside
# chroot.
results_dir = os.path.join(
cros_util.chromeos_root_inside_chroot, 'tmp/tast_results_tmp'
)
results_dir_output_chroot = results_dir.replace(
cros_util.chromeos_root_inside_chroot, opts.chromeos_root
)
# Don't reuse existing results dir, otherwise tast may rename output files.
if os.path.exists(results_dir_output_chroot):
shutil.rmtree(results_dir_output_chroot)
# TODO(kcwu): add -timeout
flags = ['-resultsdir', results_dir]
if opts.tast_build:
flags.append('-buildbundle=' + bundle)
dut, forward_host = to_forwarded_host(opts.dut)
cmd = tast_cmd(opts, 'run', *flags, dut, opts.test_name)
cros_util.cros_sdk(opts.chromeos_root, *cmd, forward_host=forward_host)
return results_dir_output_chroot
# Be careful that both tauto and tast generate results.json, but their format
# is totally different.
def parse_results_json(result_dir, test_name) -> tuple[bool, str | None]:
passed = None
reason = None
results_path = os.path.join(result_dir, 'results.json')
util.copy_file_to_log_folder(results_path)
with open(results_path) as f:
for result in json.load(f):
if result['name'] != test_name:
logger.warning('unexpected test ran: %s', result['name'])
continue
if not result['errors']:
passed = True
else:
passed = False
# We capture the reason of the last error because usually the last one
# is fatal.
for error in reversed(result['errors']):
reason = error.get('reason')
if reason:
break
if passed is None:
raise errors.ExternalError('no test result for "%s"?' % test_name)
return passed, reason
def gather_test_result(opts, result_dir) -> core.StepResult:
error_path = os.path.join(result_dir, 'run_error.txt')
if os.path.exists(error_path):
with open(error_path) as f:
message = f.read()
raise errors.ExternalError('tast global error: %s' % message)
passed, reason = parse_results_json(result_dir, opts.test_name)
if opts.metric:
chart_path = os.path.join(
result_dir, 'tests', opts.test_name, 'results-chart.json'
)
util.copy_file_to_log_folder(chart_path)
try:
values = catapult_util.get_benchmark_values(chart_path, opts.metric)
except Exception as e:
if not passed:
raise errors.BisectionTemporaryError(
'test failed to generate metric values, reason: %s' % reason
) from e
raise
return core.StepResult('value', values=values)
if opts.fail_to_pass:
if passed:
logger.info('passed')
return core.StepResult('new', reason)
logger.info('failed')
return core.StepResult('old', reason)
if passed:
logger.info('passed')
return core.StepResult('old', reason)
logger.info('failed')
return core.StepResult('new', reason)
def step_main(args: tuple[str] | None) -> core.StepResult:
parser = create_argument_parser()
opts = parser.parse_args(args)
common.config_logging(opts)
if cros_lab_util.is_satlab_dut(opts.dut):
cros_lab_util.write_satlab_ssh_config(opts.dut)
if not cros_util.is_dut(opts.dut):
raise errors.BrokenDutException(
'%r is not a valid DUT address' % opts.dut
)
# Private bundles are not included in the OS image, make sure they are
# available.
if opts.with_private_bundles and not opts.tast_build:
if not cros_util.query_dut_is_by_official_builder(opts.dut):
# Official builders uploaded private tast bundles and tast can download
# them with -downloadprivatebundles=true, so --tast-build is not
# required. Otherwise, we must build the bundles by ourselves.
raise errors.ArgumentError(
'--tast-build',
'for non-official chromeos image, --tast-build must be specified',
)
# Verify command line options.
if opts.metric:
if opts.fail_to_pass:
raise errors.ArgumentError(
'--fail-to-pass',
'--fail-to-pass is not for benchmark test (--metric)',
)
# Remove the "tast." prefix prepended by autotest.
opts.test_name = re.sub(r'^tast\.', '', opts.test_name)
cros_util.prepare_chroot(opts.chromeos_root)
tast_bundle_info = get_tast_bundle_info(opts, opts.test_name)
if not tast_bundle_info:
tast_bundle_info = get_tast_bundle_info(opts)
util.report_similar_candidates(
'test name', opts.test_name, list(tast_bundle_info)
)
assert 0 # unreachable
if len(tast_bundle_info) != 1 or opts.test_name not in tast_bundle_info:
# For example, tast in chroot after 12205.0.0 is IPC incompatible with
# tast on DUT earlier than 12028.0.0 (crbug/932307)
raise errors.ExecutionFatalError(
'"tast list" returns unexpected tests; '
'incompatible tast on DUT and in chroot?'
)
try:
prepare_to_run_test(opts)
except Exception as e:
raise errors.BisectionTemporaryError(
'failed when prepare, assume it is temporary: %s' % e
)
bundle = tast_bundle_info[opts.test_name]
try:
result_dir = run_test(opts, bundle)
except subprocess.CalledProcessError as e:
reason = 'failed to run tast; maybe build, ssh, or setup failures'
# TODO(njrafi): Differentiate between ssh failure and test crash
if opts.consider_test_crash_as_failure:
return util.get_test_crash_step_result(reason, opts.fail_to_pass)
raise errors.BisectionTemporaryError(reason) from e
return gather_test_result(opts, result_dir)
def action() -> bisector_cli.EvalAction:
return bisector_cli.EvalAction.WITH_DUT
def main(args: tuple[str] | None = None) -> int:
return bisector_cli.step_main_wrapper(step_main, args)
if __name__ == '__main__':
sys.exit(main())