| # Copyright 2018 The ChromiumOS Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| from __future__ import absolute_import, division, print_function |
| |
| import datetime |
| import json |
| import logging |
| import os |
| import tempfile |
| from collections import OrderedDict |
| |
| import dateutil.parser |
| import six |
| import yaml |
| from autotest_lib.client.common_lib import (base_job, config_vars, error) |
| from autotest_lib.client.common_lib.cros import dev_server, tpm_utils |
| from autotest_lib.server import test, utils |
| from autotest_lib.server.cros.network import wifi_test_context_manager |
| from autotest_lib.server.hosts import cros_host, servo_constants, servo_host |
| from autotest_lib.site_utils.rpm_control_system import rpm_constants |
| from autotest_lib.utils import labellib |
| from six.moves import urllib |
| |
| # A datetime.DateTime representing the Unix epoch in UTC. |
| _UNIX_EPOCH = dateutil.parser.parse('1970-01-01T00:00:00Z') |
| |
| # Keywords that are used in result json file. |
| _KEY_NAME = 'name' |
| _KEY_START = 'start' |
| _KEY_END = 'end' |
| _KEY_ERRORS = 'errors' |
| _KEY_SKIP_REASON = 'skipReason' |
| _KEY_REASON = 'reason' |
| _KEY_TIME = 'time' |
| _KEY_MISSING_REASON = 'missingReason' |
| |
| |
| def split_arguments(args): |
| """Splits arguments into the autotest and tast variable assignments. |
| Use the results as command_args and varslist respectively. |
| |
| @param args: List of strings passed to test_that --args |
| |
| @returns Array of Tauto args, Array of TAST variable assignments. |
| """ |
| |
| auto_args = [] |
| tast_vars = [] |
| for a in args: |
| if a.startswith("tast."): |
| tast_vars.append(a[5:]) |
| else: |
| auto_args.append(a) |
| return auto_args, tast_vars |
| |
| |
| def _encode_text(text): |
| """Takes an unicode string into utf-8 string |
| (bytes for python 2 and text for python 3). |
| """ |
| if six.PY2: |
| return text.encode('utf-8') |
| return text |
| |
| |
| def _encode_json(j): |
| """Takes JSON object parsed by json.load() family, and encode each unicode |
| strings into str. |
| """ |
| if isinstance(j, six.text_type): |
| return _encode_text(j) |
| if isinstance(j, list): |
| return [_encode_json(x) for x in j] |
| if isinstance(j, dict): |
| return dict((_encode_json(k), _encode_json(v)) |
| for k, v in six.iteritems(j)) |
| return j |
| |
| |
| def _server_arg(command_args, server_str): |
| """Return server arg out of command_args if found.""" |
| server_arg = None |
| server = None |
| for arg in command_args: |
| if f'{server_str}=' in arg: |
| server_arg = arg |
| break |
| if not server_arg: |
| return |
| server = server_arg.split('=')[1] |
| |
| # In case servers are passed... |
| # e.g.: f"{server_str}=localhost:1343,localhost:5678" |
| if ',' in server: |
| # Currently only support the first |
| server = server.split(',')[0] |
| |
| return server |
| |
| |
| def _dut_server_arg(command_args): |
| """Return dut_server arg out of command_args if found.""" |
| return _server_arg(command_args, 'dut_servers') |
| |
| |
| def _cache_server_arg(command_args): |
| """Return cache_server arg out of command_args if found.""" |
| return _server_arg(command_args, 'cache_endpoint') |
| |
| |
| class TastConfigError(error.AutotestError): |
| """Indicates a problem with configuration files.""" |
| |
| |
| class tast(test.test): |
| """Autotest server test that runs a Tast test suite. |
| |
| Tast is an integration-testing framework analagous to the test-running |
| portion of Autotest. See |
| https://chromium.googlesource.com/chromiumos/platform/tast/ for more |
| information. |
| |
| This class runs the "tast" command locally to execute a Tast test suite on a |
| remote DUT. |
| """ |
| version = 1 |
| |
| # Maximum time to wait for various tast commands to complete, in seconds. |
| _VERSION_TIMEOUT_SEC = 10 |
| _DOWNLOAD_TIMEOUT_SEC = 120 |
| _LIST_TIMEOUT_SEC = 120 |
| _LIST_TRIES = 3 |
| |
| # Additional time to add to the run timeout (e.g. for collecting crashes and |
| # logs). |
| _RUN_OVERHEAD_SEC = 20 |
| # Additional time given to the run command to exit before it's killed. |
| _RUN_EXIT_SEC = 5 |
| |
| # Number of times to retry SSH connection attempts to the DUT. |
| _SSH_CONNECT_RETRIES = 2 |
| |
| # File written by the tast command containing test results, as |
| # newline-terminated JSON TestResult objects. |
| _STREAMED_RESULTS_FILENAME = 'streamed_results.jsonl' |
| |
| # Text file written by the tast command if a global error caused the test |
| # run to fail (e.g. SSH connection to DUT was lost). |
| _RUN_ERROR_FILENAME = 'run_error.txt' |
| |
| # Maximum number of failing and missing tests to include in error messages. |
| _MAX_TEST_NAMES_IN_ERROR = 3 |
| |
| # Default paths where Tast files are installed by Portage packages. |
| _PORTAGE_TAST_PATH = '/usr/bin/tast' |
| |
| # Alternate locations for Tast files when using Server-Side Packaging. |
| # These files are installed from autotest_server_package.tar.bz2. |
| _SSP_ROOT = '/usr/local/tast' |
| _SSP_TAST_PATH = os.path.join(_SSP_ROOT, 'tast') |
| _SSP_REMOTE_BUNDLE_DIR = os.path.join(_SSP_ROOT, 'bundles/remote') |
| _SSP_REMOTE_DATA_DIR = os.path.join(_SSP_ROOT, 'data') |
| _SSP_REMOTE_TEST_RUNNER_PATH = os.path.join(_SSP_ROOT, 'remote_test_runner') |
| _SSP_DEFAULT_VARS_DIR_PATH = os.path.join(_SSP_ROOT, 'vars') |
| |
| _F20_CONTAINER_BREADCRUMB = '/usr/local/f20container' |
| # Prefix added to Tast test names when writing their results to TKO |
| # status.log files. |
| _TEST_NAME_PREFIX = 'tast.' |
| |
| # Prefixes of keyval keys recorded for missing tests. |
| _MISSING_TEST_KEYVAL_PREFIX = 'tast_missing_test.' |
| |
| # Job start/end TKO event status codes from base_client_job._rungroup in |
| # client/bin/job.py. |
| _JOB_STATUS_START = 'START' |
| _JOB_STATUS_END_GOOD = 'END GOOD' |
| _JOB_STATUS_END_FAIL = 'END FAIL' |
| _JOB_STATUS_END_NOSTATUS = 'END NOSTATUS' |
| _JOB_STATUS_END_SKIP = 'END TEST_NA' |
| |
| # In-job TKO event status codes from base_client_job._run_test_base in |
| # client/bin/job.py and client/common_lib/error.py. |
| _JOB_STATUS_GOOD = 'GOOD' |
| _JOB_STATUS_FAIL = 'FAIL' |
| _JOB_STATUS_NOSTATUS = 'NOSTATUS' |
| _JOB_STATUS_SKIP = 'TEST_NA' |
| |
| # Status reason used when an individual Tast test doesn't finish running. |
| _TEST_DID_NOT_FINISH_MSG = 'Test did not finish' |
| # Status reason used when an individual Tast test doesn't start running. |
| _TEST_DID_NOT_RUN_MSG = 'Test did not run' |
| |
| # Default Max System Message Log Size 20MB |
| _DEFAULT_MAX_SYS_MSG_LOG_SIZE = 20 * 1024 * 1024 |
| |
| def initialize(self, |
| host, |
| test_exprs, |
| ignore_test_failures=False, |
| max_run_sec=3600, |
| command_args=[], |
| install_root='/', |
| ssp=None, |
| build=None, |
| build_bundle='cros', |
| run_private_tests=True, |
| varsfiles=[], |
| download_data_lazily=True, |
| clear_tpm=True, |
| totalshards=1, |
| shardindex=0, |
| shardmethod=None, |
| companion_duts={}, |
| varslist=[], |
| maybemissingvars='', |
| use_camera_box=False, |
| vars_gs_path='', |
| retries=0, |
| ephemeraldevserver=None, |
| is_cft=False, |
| exclude_missing=False, |
| test_filter_files=[], |
| extrauseflags=None, |
| report_skipped=False, |
| max_sys_msg_log_size=_DEFAULT_MAX_SYS_MSG_LOG_SIZE): |
| """ |
| @param host: remote.RemoteHost instance representing DUT. |
| @param test_exprs: Array of strings describing tests to run. |
| @param ignore_test_failures: If False, this test will fail if individual |
| Tast tests report failure. If True, this test will only fail in |
| response to the tast command failing to run successfully. This |
| should generally be False when the test is running inline and True |
| when it's running asynchronously. |
| @param max_run_sec: Integer maximum running time for the "tast run" |
| command in seconds. |
| @param command_args: List of arguments passed on the command line via |
| test_that's --args flag, i.e. |args| in control file. |
| @param install_root: Root directory under which Tast binaries are |
| installed. Alternate values may be passed by unit tests. |
| @param ssp: Whether to use SSP files. Default is to auto-detect. |
| @param build: Whether to build test runners and test bundles. |
| Default is to build if and only if SSP is unavailable |
| (i.e. build = not ssp). |
| @param build_bundle: Test bundle name to build. Effective only when |
| build=True. |
| @param run_private_tests: Download and run private tests. Effective |
| only when build=False. When build=True, build_bundle can be |
| specified to build and run a private bundle. |
| @param varsfiles: list of names of yaml files containing variables set |
| in |-varsfile| arguments. |
| @param download_data_lazily: If True, external data files are downloaded |
| lazily between tests. If false, external data files are downloaded |
| in a batch before running tests. |
| @param clear_tpm: clear the TPM first before running the tast tests. |
| @param totalshards: Total number of shards. |
| @param shardindex: The shard index to be run. |
| @param shardmethod: The shard method how to distribute the tests. |
| @param companion_duts: A map of role to DUT name to tast run command as |
| |-companiondut| arguments. Each entry in the map will be formatted |
| as "role:dut" for each -companiondut argument. |
| @param varslist: list of strings to pass to tast run command as |-vars| |
| arguments. Each string should be formatted as "name=value". |
| @param maybemissingvars: a regex to pass to tast run command as |
| |-maybemissingvars| arguments. |
| @param vars_gs_path: gs path to load vars from. The vars are loaded |
| from gs in json format (key = value), then stored in a local |
| yaml file. The local file name is then appended to |-varsfiles|. |
| @param use_camera_box: Bring the IP address of chart device in CameraBox |
| to tast tests. |
| @param ephemeraldevserver: A value to pass to -ephemeraldevserver |
| @param exclude_missing: This option will exclude tests that are requested, but not found in |
| `tast list` command |
| @param test_filter_files: This option includes a list of files containing names |
| of test to be disabled. |
| @param extrauseflags: Extra use flags to control tast behavior. This value will be passed to |
| the `tast` command as -extrauseflags command line flag, |
| @param report_skipped: If true then skipped tests will be reported in |
| the status.log |
| @param max_sys_msg_log_size: Max size for the downloaded system message log after |
| each test (default to 20MB) |
| |
| When the F20 breadcrumb is detected, it is assumed we are running in |
| the F20 container, meaning we will force disable SSP (though the |
| SSP flag should be false in this case). The F20 container is fully |
| build versioned and matches the chroot paths, so we do not want to |
| take the SSP logic. |
| |
| @raises error.TestFail if the Tast installation couldn't be found. |
| """ |
| f20_container = False |
| if os.path.exists(self._F20_CONTAINER_BREADCRUMB): |
| ssp = False |
| f20_container = True |
| if ssp is None: |
| ssp = os.path.exists(self._SSP_TAST_PATH) |
| if build is None: |
| build = not ssp |
| |
| self._host = host |
| self._test_exprs = test_exprs |
| self._ignore_test_failures = ignore_test_failures |
| self._max_run_sec = max_run_sec |
| self._command_args = command_args |
| self._install_root = install_root |
| self._ssp = ssp |
| self._build = build |
| self._build_bundle = build_bundle |
| self._run_private_tests = run_private_tests |
| self._fake_now = None |
| self._varsfiles = varsfiles |
| self._varslist = varslist |
| self._download_data_lazily = download_data_lazily |
| self._clear_tpm = clear_tpm |
| self._totalshards = totalshards |
| self._shardindex = shardindex |
| self._shardmethod = shardmethod |
| self._companion_duts = companion_duts |
| self._maybemissingvars = maybemissingvars |
| self._vars_gs_path = vars_gs_path |
| self._use_camera_box = use_camera_box |
| self._retries = retries |
| self._f20_container = f20_container or is_cft |
| self._ephemeraldevserver = ephemeraldevserver |
| self._exclude_missing = exclude_missing |
| self._test_filter_files = test_filter_files |
| self._extrauseflags = extrauseflags |
| self._report_skipped = report_skipped |
| self._max_sys_msg_log_size = max_sys_msg_log_size |
| |
| # Need to pass in dut_servers for every test in CFT. |
| # But only add it if not already in varslist. |
| self._cache_server = _cache_server_arg(command_args) |
| dut_serversFound = self._find_dutservers(command_args) |
| |
| # Without dut_servers or cache_server, |
| # trying to download private will to fail in CFT. |
| if self._f20_container and ( |
| not dut_serversFound and not self._cache_server): |
| self._run_private_tests = False |
| |
| # List of JSON objects describing tests that will be run. See Test in |
| # src/platform/tast/src/chromiumos/tast/testing/test.go for details. |
| self._tests_to_run = [] |
| |
| # List of JSON objects corresponding to tests from |
| # _STREAMED_RESULTS_FILENAME. See TestResult in |
| # src/platform/tast/src/chromiumos/cmd/tast/run/results.go for details. |
| self._test_results = [] |
| |
| # Error message read from _RUN_ERROR_FILENAME, if any. |
| self._run_error = None |
| |
| self._tast_path = self._get_path( |
| self._SSP_TAST_PATH if ssp else self._PORTAGE_TAST_PATH) |
| |
| # Register a hook to write the results of individual Tast tests as |
| # top-level entries in the TKO status.log file. |
| self.job.add_post_run_hook(self._log_all_unique_tests) |
| |
| def run_once(self): |
| """Runs a single iteration of the test.""" |
| |
| if self._clear_tpm and self._host_supports_clear_tpm_owner(): |
| tpm_utils.ClearTPMOwnerRequest(self._host, wait_for_ready=True) |
| |
| self._log_version() |
| if self._f20_container and self._cache_server: |
| # CFT, use cacheservers. |
| self._find_cachesevers() |
| else: |
| self._find_devservers() |
| |
| # Shortcut if no test belongs to the specified test_exprs. |
| list_test_exception = None |
| has_tests_to_run = False |
| for i in range(self._LIST_TRIES): |
| try: |
| if i > 0: |
| logging.info('Retrying to get which tests to run') |
| has_tests_to_run = bool(self._get_tests_to_run()) |
| list_test_exception = None |
| break |
| except Exception as e: |
| list_test_exception = e |
| if list_test_exception: |
| raise error.TestFail('Failed to get list of tests to run: %s' % |
| str(list_test_exception)) |
| if not has_tests_to_run: |
| return |
| |
| # TODO(b/221333999): There are no devservers in CFT (F20), so this |
| # would likely error. Once full CFT is done tast.py will be deprecated |
| # and this won't be needed. |
| if not self._f20_container: |
| self._pull_varsfile_from_gs() |
| |
| run_failed = False |
| run_failed_msg = None |
| try: |
| self._run_tests() |
| except Exception as e: |
| run_failed = True |
| run_failed_msg = str(e).split('\n', 1)[0] |
| raise |
| finally: |
| self._read_run_error() |
| # Parse partial results even if the tast command didn't finish. |
| self._parse_results(run_failed, run_failed_msg) |
| |
| def _find_dutservers(self, command_args): |
| """Find Dutsever if there is no cache-server and in CFT. |
| |
| Results is saved in self._varlists. |
| |
| Returns bool if the dutserver is found or not. |
| """ |
| if self._f20_container and not self._cache_server: |
| dut_server = _dut_server_arg(command_args) |
| if dut_server: |
| self._varslist.append('servers.dut=:%s' % dut_server) |
| return True |
| return False |
| |
| def set_fake_now_for_testing(self, now): |
| """Sets a fake timestamp to use in place of time.time() for unit tests. |
| |
| @param now Numeric timestamp as would be returned by time.time(). |
| """ |
| self._fake_now = now |
| |
| def _pull_varsfile_from_gs(self): |
| """Pulls varsfiles from GS, does dynamic values transformation, stores |
| it as a local file and appends the file name to varsfiles. |
| |
| Has to be called after _get_tests_to_run since it's using _tests_to_run. |
| |
| @param varsgspath Path to varsfiles in GS e.g. |
| 'config/perf_cuj/perf_cuj.config'. |
| |
| @raises TastConfigError for config errors. |
| """ |
| if not self._vars_gs_path: |
| return |
| |
| devservers = dev_server.ImageServer.get_available_devservers() |
| devserver_url = devservers[0][0] |
| if not devserver_url: |
| raise TastConfigError('No devserver_url') |
| |
| logging.info('Using devserver: %s', devserver_url) |
| labels = self._host.host_info_store.get().labels |
| build = labellib.LabelsMapping(labels).get(labellib.Key.CROS_VERSION) |
| if not build: |
| raise TastConfigError( |
| 'Not able to detect build, means not running on Moblab.') |
| |
| ds = dev_server.ImageServer(devserver_url) |
| gs_bucket = dev_server._get_image_storage_server() |
| if not gs_bucket: |
| raise TastConfigError('No image storage server gs bucket') |
| |
| config_path, config_file = os.path.split(self._vars_gs_path) |
| archive_url = os.path.join(gs_bucket, config_path.strip('/')) |
| logging.info('Staging configuration from %s.', gs_bucket) |
| try: |
| ds.stage_artifacts(build, |
| archive_url=archive_url, |
| files=[config_file]) |
| except Exception as e: |
| raise TastConfigError('Staging artifacts failed: %s', str(e)) |
| |
| logging.info('Parsing configuration from %s.', archive_url) |
| config_url = os.path.join(devserver_url, 'static', |
| self._vars_gs_path.strip('/')) |
| response = urllib.request.urlopen(config_url) |
| vars = json.loads(response.read()) |
| test_args = dict() |
| for key in vars: |
| test_args[key] = vars[key] |
| logging.info('Read %d values from remote configuration.', len(vars)) |
| |
| extvars = self._fill_config_extvars() |
| test_args = config_vars.TransformConfig(test_args, extvars) |
| |
| with tempfile.NamedTemporaryFile(suffix='.yaml', |
| mode='w+', |
| encoding='utf-8', |
| delete=False) as temp_file: |
| yaml.safe_dump(test_args, |
| stream=temp_file, |
| default_flow_style=False, |
| allow_unicode=True) |
| self._varsfiles.append(temp_file.name) |
| |
| def _fill_config_extvars(self): |
| """Fill in external variables map for conditional config processing. |
| |
| The sources used (in order of precedence low to high): |
| * --varsfiles. |
| * --varslist. |
| * list of tests to run. |
| * command_args: List of arguments passed on the command line via |
| test_that's --args flag, i.e. |args| in control file. |
| * DUT labels (with and without a value). |
| |
| @returns external variables map. |
| """ |
| # The latter overwrites the former. |
| extvars = {} |
| |
| # Load varsfiles |
| for varsfile in self._varsfiles: |
| with open(varsfile, 'r') as f: |
| for key, val in yaml.safe_load(f).items(): |
| if 'var:' + key in extvars: |
| logging.info('var:%s overwritten', key) |
| extvars['var:' + key] = val |
| |
| # Load vars |
| for var in self._varslist: |
| key, val = var.split('=', 1) |
| if 'var:' + key in extvars: |
| logging.info('var:%s overwritten', key) |
| extvars['var:' + key] = val |
| |
| # Load tests_to_run |
| extvars['tests:'] = '\n'.join([test['name'] for test in self._tests_to_run]) |
| for test_to_run in self._tests_to_run: |
| extvars['test:' + test_to_run['name']] = '' |
| |
| # Load command_args |
| extvars['args:'] = '\n'.join(self._command_args) |
| for key, val in utils.args_to_dict(self._command_args).items(): |
| extvars['arg:' + key] = val |
| for command_arg in self._command_args: |
| if '=' not in command_arg and ':' not in command_arg: |
| extvars['arg:' + command_arg] = '' |
| |
| # Load labels |
| labels = self._host.host_info_store.get().labels |
| extvars['labels:'] = '\n'.join(labels) |
| for label in labels: |
| key, val = (label.split(':', 1) + [''])[0:2] |
| extvars['label:' + key] = val |
| |
| return extvars |
| |
| def _get_path(self, path): |
| """Returns the path to an installed Tast-related file or directory. |
| |
| @param path: Absolute paths in root filesystem, e.g. "/usr/bin/tast". |
| |
| @returns Absolute path within install root, e.g. |
| "/usr/local/tast/usr/bin/tast". |
| """ |
| return os.path.join(self._install_root, os.path.relpath(path, '/')) |
| |
| def _get_servo_args(self): |
| """Gets servo-related arguments to pass to "tast run". |
| |
| @returns List of command-line flag strings that should be inserted in |
| the command line after "tast run". |
| """ |
| # Start with information provided by the Autotest database. |
| merged_args = {} |
| host_args = servo_host.get_servo_args_for_host(self._host) |
| if host_args: |
| merged_args.update(host_args) |
| |
| # Incorporate information that was passed manually. |
| args_dict = utils.args_to_dict(self._command_args) |
| merged_args.update(cros_host.CrosHost.get_servo_arguments(args_dict)) |
| |
| logging.info('Autotest servo-related args: %s', merged_args) |
| host_arg = merged_args.get(servo_constants.SERVO_HOST_ATTR) |
| port_arg = merged_args.get(servo_constants.SERVO_PORT_ATTR) |
| if not host_arg or not port_arg: |
| return [] |
| return ['-var=servo=%s:%s' % (host_arg, port_arg)] |
| |
| def _get_firmware_args(self): |
| """Gets firmware-related arguments to pass to "tast run". |
| |
| @returns List of command-line flag strings that should be inserted in |
| the command line after "tast run". |
| """ |
| # Incorporate information that was passed manually. |
| args_dict = utils.args_to_dict(self._command_args) |
| |
| args = [] |
| no_ec_sync = args_dict.get("no_ec_sync") |
| if no_ec_sync: |
| args += ['-var=firmware.no_ec_sync=' + no_ec_sync] |
| logging.info('Firmware args: %s', args) |
| return args |
| |
| def _get_rpm_args(self): |
| """Gets rpm-related arguments to pass to "tast run". |
| |
| @returns List of command-line flag strings that should be inserted in |
| the command line after "tast run". |
| """ |
| info = self._host.host_info_store.get() |
| args = [] |
| forward_args = [ |
| (rpm_constants.POWERUNIT_HOSTNAME_KEY, 'powerunitHostname=%s'), |
| (rpm_constants.POWERUNIT_OUTLET_KEY, 'powerunitOutlet=%s'), |
| (rpm_constants.HYDRA_HOSTNAME_KEY, 'hydraHostname=%s'), |
| ] |
| for key, var_arg in forward_args: |
| if key in info.attributes: |
| args += ['-var=' + var_arg % info.attributes[key]] |
| logging.info('RPM args: %s', args) |
| return args |
| |
| def _get_wificell_args(self): |
| """Gets wificell-related (router, pcap) arguments to pass to "tast run". |
| |
| @returns List of command-line flag strings that should be inserted in |
| the command line after "tast run". |
| """ |
| # Incorporate information that was passed manually. |
| args_dict = utils.args_to_dict(self._command_args) |
| args = [] |
| # Alias of WiFiTestContextManager. |
| WiFiManager = wifi_test_context_manager.WiFiTestContextManager |
| # TODO(crbug.com/1065601): plumb other WiFi test specific arguments, |
| # e.g. pcap address. See: WiFiTestContextManager's constants. |
| forward_args = [ |
| (WiFiManager.CMDLINE_ROUTER_ADDR, 'router=%s'), |
| (WiFiManager.CMDLINE_PCAP_ADDR, 'pcap=%s'), |
| ] |
| for key, var_arg in forward_args: |
| if key in args_dict: |
| args += ['-var=' + var_arg % args_dict[key]] |
| # Append "routers" var for supporting multi-router tests with current |
| # two-AP fixture setup (with specified router_addr and pcap_addr args). |
| # TODO(b/171949862): remove this when a new multi-router fixture is |
| # defined and rolled out to the lab. |
| if (WiFiManager.CMDLINE_ROUTER_ADDR in args_dict |
| and WiFiManager.CMDLINE_PCAP_ADDR in args_dict): |
| args += [ |
| '-var=routers=%s,%s' % |
| (args_dict[WiFiManager.CMDLINE_ROUTER_ADDR], |
| args_dict[WiFiManager.CMDLINE_PCAP_ADDR]) |
| ] |
| logging.info('Autotest wificell-related args: %s', args) |
| return args |
| |
| def _get_cloud_storage_info(self): |
| """Gets the cloud storage bucket URL to pass to tast. |
| |
| @returns Cloud storage bucket URL that should be inserted in |
| the command line after "tast run". |
| """ |
| gs_bucket = dev_server._get_image_storage_server() |
| args_dict = utils.args_to_dict(self._command_args) |
| gcs_override = args_dict.get('buildartifactsurl') |
| if gcs_override: |
| logging.info('Cloud storage bucket: %s', gcs_override) |
| return ['-buildartifactsurl=%s' % gcs_override] |
| build = args_dict.get('build') |
| if not build: |
| labels = self._host.host_info_store.get().labels |
| build = labellib.LabelsMapping(labels).get( |
| labellib.Key.CROS_VERSION) |
| |
| if not gs_bucket or not build: |
| return [] |
| gs_path = os.path.join(gs_bucket, build) |
| if not gs_path.endswith('/'): |
| gs_path += '/' |
| logging.info('Cloud storage bucket: %s', gs_path) |
| return ['-buildartifactsurl=%s' % gs_path] |
| |
| def _find_devservers(self): |
| """Finds available devservers. |
| |
| The result is saved as self._devserver_args. |
| """ |
| logging.info('All devservers: %s', |
| ', '.join(dev_server.ImageServer.servers())) |
| devservers, can_retry = dev_server.ImageServer.get_available_devservers( |
| self._host.hostname, prefer_local_devserver=True) |
| if not devservers and can_retry and (self._host.is_satlab() |
| or 'MOBLAB' in os.environ): |
| devservers, can_retry = dev_server.ImageServer.get_available_devservers( |
| self._host.hostname, prefer_local_devserver=False) |
| logging.info('Using devservers: %s', ', '.join(devservers)) |
| self._devserver_args = ['-devservers=%s' % ','.join(devservers)] |
| if self._ephemeraldevserver is not None: |
| self._devserver_args.append('-ephemeraldevserver=%s' % |
| self._ephemeraldevserver) |
| |
| def _find_cachesevers(self): |
| """Finds available cacheserver to act as the devserver. |
| |
| The result is saved as self._devserver_args. |
| """ |
| if not self._cache_server: |
| return |
| if not self._cache_server.startswith('http://'): |
| self._cache_server = f'http://{self._cache_server}' |
| logging.info('Using devserver: %s', self._cache_server) |
| self._devserver_args = ['-devservers=%s' % self._cache_server] |
| if self._ephemeraldevserver is not None: |
| self._devserver_args.append('-ephemeraldevserver=%s' % |
| self._ephemeraldevserver) |
| |
| def _log_version(self): |
| """Runs the tast command locally to log its version.""" |
| try: |
| utils.run([self._tast_path, '-version'], |
| timeout=self._VERSION_TIMEOUT_SEC, |
| stdout_tee=utils.TEE_TO_LOGS, |
| stderr_tee=utils.TEE_TO_LOGS, |
| stderr_is_expected=True, |
| stdout_level=logging.INFO, |
| stderr_level=logging.ERROR) |
| except error.CmdError as e: |
| logging.error('Failed to log tast version: %s', str(e)) |
| |
| def _tast_target(self): |
| """Returns the <target> portion of the tast command.""" |
| return '%s%s' % (self._host.hostname, ':%d' % |
| self._host.port if self._host.port else '') |
| |
| def _host_supports_clear_tpm_owner(self): |
| """True iff host supports TPM clear owner request.""" |
| return self._tast_target() != '-' |
| |
| def _run_tast(self, |
| subcommand, |
| extra_subcommand_args, |
| test_exprs, |
| timeout_sec, |
| log_stdout=False, |
| ignore_status=False): |
| """Runs the tast command locally to e.g. list available tests or perform |
| testing against the DUT. |
| |
| @param subcommand: Subcommand to pass to the tast executable, e.g. 'run' |
| or 'list'. |
| @param extra_subcommand_args: List of additional subcommand arguments. |
| @param test_exprs: Array of strings describing tests to run. |
| @param timeout_sec: Integer timeout for the command in seconds. |
| @param log_stdout: If true, write stdout to log. |
| @param ignore_status: If true, command execution errors are ignored. |
| |
| @returns client.common_lib.utils.CmdResult object describing the result. |
| |
| @raises error.TestFail if the tast command fails or times out. |
| """ |
| cmd = [ |
| self._tast_path, |
| '-verbose=true', |
| '-logtime=false', |
| subcommand, |
| '-sshretries=%d' % self._SSH_CONNECT_RETRIES, |
| '-downloaddata=%s' % ( |
| 'lazy' if self._download_data_lazily else 'batch'), |
| '-totalshards=%s' % self._totalshards, |
| '-shardindex=%s' % self._shardindex, |
| ] |
| if self._shardmethod: |
| cmd.append(f'-shardmethod={self._shardmethod}') |
| if self._f20_container: |
| cmd.extend(['-build=false']) |
| if self._run_private_tests: |
| cmd.append('-downloadprivatebundles=true') |
| elif self._build: |
| cmd.extend([ |
| '-build=true', |
| '-buildbundle=%s' % self._build_bundle, |
| '-checkbuilddeps=false', |
| ]) |
| else: |
| cmd.append('-build=false') |
| if self._ssp: |
| remote_test_runner_path = self._get_path( |
| self._SSP_REMOTE_TEST_RUNNER_PATH) |
| if not os.path.exists(remote_test_runner_path): |
| raise error.TestFail( |
| '%s does not exist (broken SSP?)' % |
| remote_test_runner_path) |
| cmd.extend([ |
| '-remotebundledir=%s' % self._get_path( |
| self._SSP_REMOTE_BUNDLE_DIR), |
| '-remotedatadir=%s' % self._get_path( |
| self._SSP_REMOTE_DATA_DIR), |
| '-remoterunner=%s' % remote_test_runner_path, |
| ]) |
| if subcommand == 'run': |
| cmd.append('-defaultvarsdir=%s' % |
| self._get_path(self._SSP_DEFAULT_VARS_DIR_PATH)) |
| if self._run_private_tests: |
| cmd.append('-downloadprivatebundles=true') |
| if not self._f20_container: |
| cmd.extend(self._devserver_args) |
| elif self._f20_container and self._cache_server: |
| cmd.extend(self._devserver_args) |
| cmd.extend(extra_subcommand_args) |
| cmd.append(self._tast_target()) |
| cmd.extend(test_exprs) |
| |
| logging.info('Running %s', |
| ' '.join([utils.sh_quote_word(a) for a in cmd])) |
| try: |
| return utils.run( |
| cmd, |
| ignore_status=ignore_status, |
| timeout=timeout_sec, |
| stdout_tee=(utils.TEE_TO_LOGS if log_stdout else None), |
| stderr_tee=utils.TEE_TO_LOGS, |
| stderr_is_expected=True, |
| stdout_level=logging.INFO, |
| stderr_level=logging.ERROR) |
| except error.CmdError as e: |
| # Run several commands to debug possible network issues. |
| # TODO(b/189332919): Remove this logic once we finish debugging. |
| logging.info('Tast exited abnormally. Running several commands to ' |
| 'diagnose possible network issues...') |
| utils.run('time getent ahosts %s' % self._host.hostname, |
| timeout=60, |
| ignore_status=True, |
| stdout_tee=utils.TEE_TO_LOGS, |
| stderr_tee=utils.TEE_TO_LOGS, |
| stderr_is_expected=True, |
| stdout_level=logging.INFO, |
| stderr_level=logging.ERROR) |
| utils.run( |
| 'ssh ' |
| # Enable maximum debug logging. |
| '-vvv ' |
| # Disable connection sharing to debug connection issues. |
| '-o ControlPath=none ' |
| # Following arguments were copied from Autotest logs. |
| '-a -x ' |
| '-o StrictHostKeyChecking=no ' |
| '-o UserKnownHostsFile=/dev/null ' |
| '-o BatchMode=yes ' |
| '-o ConnectTimeout=10 ' |
| '-o ConnectionAttempts=3 ' |
| '-l root %s%s true' % |
| ('-p %d ' % self._host.port if self._host.port else '', |
| self._host.hostname), |
| timeout=60, |
| ignore_status=True, |
| stdout_tee=utils.TEE_TO_LOGS, |
| stderr_tee=utils.TEE_TO_LOGS, |
| stderr_is_expected=True, |
| stdout_level=logging.INFO, |
| stderr_level=logging.ERROR) |
| # The tast command's output generally ends with a line describing |
| # the error that was encountered; include it in the first line of |
| # the TestFail exception. Fall back to stderr if stdout is empty (as |
| # is the case with the "list" subcommand, which uses stdout to print |
| # test data). |
| get_last_line = lambda s: s.strip().split('\n')[-1].strip() |
| last_line = (get_last_line(e.result_obj.stdout) or |
| get_last_line(e.result_obj.stderr)) |
| msg = (' (last line: %s)' % last_line) if last_line else '' |
| raise error.TestFail('Failed to run tast%s: %s' % (msg, str(e))) |
| except error.CmdTimeoutError as e: |
| raise error.TestFail('Got timeout while running tast: %s' % str(e)) |
| |
| def _get_tests_to_run(self): |
| """Runs the tast command to update the list of tests that will be run. |
| |
| @returns False if no tests matched by test_exprs; True otherwise |
| |
| @raises error.TestFail if the tast command fails or times out. |
| """ |
| logging.info('Getting list of tests that will be run') |
| args = ['-json=true'] + self._get_cloud_storage_info() |
| for var in self._varslist: |
| args.append('-var=%s' % var) |
| |
| result = self._run_tast('list', args, self._test_exprs, |
| self._LIST_TIMEOUT_SEC) |
| try: |
| self._tests_to_run = _encode_json(json.loads( |
| result.stdout.strip())) |
| except ValueError as e: |
| raise error.TestFail('Failed to parse tests: %s' % str(e)) |
| if len(self._tests_to_run) == 0: |
| expr = ' '.join([utils.sh_quote_word(a) for a in self._test_exprs]) |
| logging.warning('No tests matched by %s', expr) |
| return False |
| |
| logging.info('Expect to run %d test(s)', len(self._tests_to_run)) |
| |
| logging.info('Tests in scope:') |
| for test in self._tests_to_run: |
| logging.info('Test: %s', test['name']) |
| |
| return True |
| |
| def _run_tests(self): |
| """Runs the tast command to perform testing. |
| |
| @raises error.TestFail if the tast command fails or times out (but not |
| if individual tests fail). |
| """ |
| args = [ |
| '-resultsdir=' + self.resultsdir, |
| '-waituntilready=true', |
| '-timeout=' + str(self._max_run_sec), |
| '-continueafterfailure=true', |
| '-maxsysmsglogsize=' + str(self._max_sys_msg_log_size), |
| ] |
| args.extend(self._get_servo_args()) |
| args.extend(self._get_rpm_args()) |
| args.extend(self._get_wificell_args()) |
| args.extend(self._get_cloud_storage_info()) |
| args.extend(self._get_firmware_args()) |
| if self._retries: |
| args.append('-retries=%d' % self._retries) |
| |
| for varsfile in self._varsfiles: |
| args.append('-varsfile=%s' % varsfile) |
| |
| for var in self._varslist: |
| args.append('-var=%s' % var) |
| |
| if self._maybemissingvars: |
| args.append('-maybemissingvars=%s' % self._maybemissingvars) |
| |
| for role, dut in sorted(self._companion_duts.items()): |
| args.append( |
| '-companiondut=%s:%s%s' % |
| (role, dut.hostname, ':%d' % dut.port if dut.port else '')) |
| |
| for file in self._test_filter_files: |
| args.append('-testfilterfile=%s' % file) |
| |
| if self._extrauseflags: |
| args.append('-extrauseflags=%s' % self._extrauseflags) |
| |
| logging.info('Running tests with timeout of %d sec', self._max_run_sec) |
| # This option will exclude tests that are requested, but not found in |
| # `tast list` command |
| if self._exclude_missing: |
| tests_to_run_list = [test["name"] for test in self._tests_to_run] |
| self._run_tast('run', |
| args, |
| tests_to_run_list, |
| self._max_run_sec + tast._RUN_EXIT_SEC, |
| log_stdout=True) |
| else: |
| self._run_tast('run', |
| args, |
| self._test_exprs, |
| self._max_run_sec + tast._RUN_EXIT_SEC, |
| log_stdout=True) |
| |
| def _read_run_error(self): |
| """Reads a global run error message written by the tast command.""" |
| # The file is only written if a run error occurred. |
| path = os.path.join(self.resultsdir, self._RUN_ERROR_FILENAME) |
| if os.path.exists(path): |
| with open(path, 'r') as f: |
| self._run_error = f.read().strip() |
| |
| def maybe_replace(self, test, failed): |
| """ Removes a test from the list of failed results |
| |
| @param test: Name of test to remove from failed list |
| @param failed: List of failed tests |
| """ |
| # Remove the result, will take & only count the second result. |
| if test[_KEY_NAME] in failed: |
| failed.remove(test[_KEY_NAME]) |
| |
| def _parse_results(self, ignore_missing_file, run_error_msg): |
| """Parses results written by the tast command. |
| |
| @param ignore_missing_file: If True, return without raising an exception |
| if the Tast results file is missing. This is used to avoid raising a |
| new error if there was already an earlier error while running the |
| tast process. |
| @param run_error_msg: The error message from Tast when there is an |
| error. It will be None if Tast encounters no errors. |
| |
| @raises error.TestFail if results file is missing and |
| ignore_missing_file is False, or one or more tests failed and |
| _ignore_test_failures is false. |
| """ |
| # The file may not exist if "tast run" failed to run. Tests that were |
| # seen from the earlier "tast list" command will be reported as having |
| # missing results. |
| path = os.path.join(self.resultsdir, self._STREAMED_RESULTS_FILENAME) |
| if not os.path.exists(path): |
| if ignore_missing_file: |
| return |
| raise error.TestFail('Results file %s not found' % path) |
| |
| failed = set() |
| seen_test_names = set() |
| with open(path, 'r') as f: |
| for line in f: |
| line = line.strip() |
| if not line: |
| continue |
| try: |
| test = _encode_json(json.loads(line)) |
| except ValueError as e: |
| raise error.TestFail('Failed to parse %s: %s' % (path, e)) |
| self._test_results.append(test) |
| if test[_KEY_NAME] in seen_test_names: |
| self.maybe_replace(test, failed) |
| |
| name = test[_KEY_NAME] |
| seen_test_names.add(name) |
| |
| if test.get(_KEY_ERRORS): |
| for err in test[_KEY_ERRORS]: |
| logging.warning('%s: %s', name, err[_KEY_REASON]) |
| failed.add(name) |
| else: |
| # The test will have a zero (i.e. 0001-01-01 00:00:00 UTC) |
| # end time (preceding the Unix epoch) if it didn't report |
| # completion. |
| if _rfc3339_time_to_timestamp(test[_KEY_END]) <= 0: |
| failed.add(name) |
| |
| missing = [ |
| t[_KEY_NAME] for t in self._tests_to_run |
| if t[_KEY_NAME] not in seen_test_names |
| ] |
| |
| if missing: |
| self._record_missing_tests(missing) |
| time_str = '%sZ' % datetime.datetime.utcnow().isoformat() |
| for name in missing: |
| t = {} |
| t[_KEY_NAME] = name |
| t[_KEY_START] = time_str |
| t[_KEY_END] = time_str |
| if self._run_error: |
| t[_KEY_MISSING_REASON] = '%s due to global error: %s' % ( |
| self._TEST_DID_NOT_RUN_MSG, self._run_error) |
| elif run_error_msg: |
| t[_KEY_MISSING_REASON] = run_error_msg |
| else: |
| t[_KEY_MISSING_REASON] = self._TEST_DID_NOT_RUN_MSG |
| self._test_results.append(t) |
| |
| failure_msg = self._get_failure_message(failed, missing) |
| if failure_msg: |
| raise error.TestFail(failure_msg) |
| |
| def _get_failure_message(self, failed, missing): |
| """Returns an error message describing failed and/or missing tests. |
| |
| @param failed: List of string names of Tast tests that failed. |
| @param missing: List of string names of Tast tests with missing results. |
| |
| @returns String to be used as error.TestFail message. |
| """ |
| def list_tests(names): |
| """Returns a string listing tests. |
| |
| @param names: List of string test names. |
| |
| @returns String listing tests. |
| """ |
| s = ' '.join(sorted(names)[:self._MAX_TEST_NAMES_IN_ERROR]) |
| if len(names) > self._MAX_TEST_NAMES_IN_ERROR: |
| s += ' ...' |
| return s |
| |
| msg = '' |
| if failed and not self._ignore_test_failures: |
| msg = '%d failed: %s' % (len(failed), list_tests(failed)) |
| if missing: |
| if msg: |
| msg += '; ' |
| msg += '%d missing: %s' % (len(missing), list_tests(missing)) |
| return msg |
| |
| def _log_all_unique_tests(self): |
| """Writes entries to the TKO status.log file describing the results of |
| all tests. |
| |
| If there are 2 tests with the same name, AND it has an error (failure) |
| replace the result. |
| Because: if it has an err AND a second result, its either: |
| The first attempt is logged and failed and we want to use the |
| retry result |
| Or the attempts are out of order, and the already logged attempt is |
| the second attempt which failed, meaning the first ALSO failed. |
| So in this case, its still safe to override because we just |
| need to mark the failure. |
| The benefit of this is, if the first result is logged and failed, the |
| retry might have passed, so we want to log that. |
| |
| """ |
| seen_test_names = set() |
| tests_to_log = OrderedDict() |
| for test_res in self._test_results: |
| test_name = test_res[_KEY_NAME] |
| |
| dup_res = tests_to_log.get(test_name) |
| if not dup_res or dup_res.get(_KEY_ERRORS): |
| tests_to_log[test_name] = test_res |
| for test in tests_to_log.values(): |
| self._log_test(test) |
| seen_test_names.add(test[_KEY_NAME]) |
| |
| def _log_test(self, test): |
| """Writes events to the TKO status.log file describing the results from |
| a Tast test. |
| |
| @param test: A JSON object corresponding to a single test from a Tast |
| results.json file. See TestResult in |
| src/platform/tast/src/chromiumos/cmd/tast/run/results.go for |
| details. |
| """ |
| name = test[_KEY_NAME] |
| start_time = _rfc3339_time_to_timestamp(test[_KEY_START]) |
| end_time = _rfc3339_time_to_timestamp(test[_KEY_END]) |
| |
| test_reported_errors = bool(test.get(_KEY_ERRORS)) |
| test_skipped = bool(test.get(_KEY_SKIP_REASON)) |
| test_not_run = bool(test.get(_KEY_MISSING_REASON)) |
| # The test will have a zero (i.e. 0001-01-01 00:00:00 UTC) end time |
| # (preceding the Unix epoch) if it didn't report completion. |
| test_finished = end_time > 0 |
| |
| # Avoid reporting tests that were skipped. |
| if test_skipped and not test_reported_errors and not self._report_skipped: |
| return |
| |
| # Look for magic error _TEST_DID_NOT_RUN_MSG and mark test as not run. |
| for err in test.get(_KEY_ERRORS) or []: |
| if err[_KEY_REASON] == self._TEST_DID_NOT_RUN_MSG: |
| test_not_run = True |
| test[_KEY_MISSING_REASON] = self._TEST_DID_NOT_RUN_MSG |
| |
| self._log_test_event(self._JOB_STATUS_START, name, start_time) |
| |
| if test_not_run: |
| self._log_test_event(self._JOB_STATUS_NOSTATUS, name, end_time, |
| test[_KEY_MISSING_REASON]) |
| end_status = self._JOB_STATUS_END_NOSTATUS |
| elif test_skipped and not test_reported_errors and self._report_skipped: |
| self._log_test_event(self._JOB_STATUS_SKIP, name, end_time, |
| test.get(_KEY_SKIP_REASON)) |
| end_status = self._JOB_STATUS_END_SKIP |
| elif test_finished and not test_reported_errors: |
| self._log_test_event(self._JOB_STATUS_GOOD, name, end_time) |
| end_status = self._JOB_STATUS_END_GOOD |
| else: |
| # The previous START event automatically increases the log |
| # indentation level until the following END event. |
| if test_reported_errors: |
| for err in test[_KEY_ERRORS]: |
| error_time = _rfc3339_time_to_timestamp(err[_KEY_TIME]) |
| self._log_test_event(self._JOB_STATUS_FAIL, name, |
| error_time, err[_KEY_REASON]) |
| if not test_finished: |
| # If a run-level error was encountered (e.g. the SSH connection |
| # to the DUT was lost), report it here to make it easier to see |
| # the reason why the test didn't finish. |
| if self._run_error: |
| self._log_test_event(self._JOB_STATUS_FAIL, name, |
| start_time, self._run_error) |
| self._log_test_event(self._JOB_STATUS_FAIL, name, start_time, |
| self._TEST_DID_NOT_FINISH_MSG) |
| end_time = start_time |
| |
| end_status = self._JOB_STATUS_END_FAIL |
| |
| self._log_test_event(end_status, name, end_time) |
| |
| def _log_test_event(self, status_code, test_name, timestamp, message=''): |
| """Logs a single event to the TKO status.log file. |
| |
| @param status_code: Event status code, e.g. 'END GOOD'. See |
| client/common_lib/log.py for accepted values. |
| @param test_name: Tast test name, e.g. 'ui.ChromeLogin'. |
| @param timestamp: Event timestamp (as seconds since Unix epoch). |
| @param message: Optional human-readable message. |
| """ |
| full_name = self._TEST_NAME_PREFIX + test_name |
| # The TKO parser code chokes on floating-point timestamps. |
| entry = base_job.status_log_entry(status_code, |
| None, |
| full_name, |
| message, |
| None, |
| timestamp=int(timestamp)) |
| self.job.record_entry(entry, False) |
| |
| def _record_missing_tests(self, missing): |
| """Records tests with missing results in job keyval file. |
| |
| @param missing: List of string names of Tast tests with missing results. |
| """ |
| keyvals = {} |
| for i, name in enumerate(sorted(missing)): |
| keyvals['%s%d' % (self._MISSING_TEST_KEYVAL_PREFIX, i)] = name |
| utils.write_keyval(self.job.resultdir, keyvals) |
| |
| |
| class _LessBrokenParserInfo(dateutil.parser.parserinfo): |
| """dateutil.parser.parserinfo that interprets years before 100 correctly. |
| |
| Our version of dateutil.parser.parse misinteprets an unambiguous string like |
| '0001-01-01T00:00:00Z' as having a two-digit year, which it then converts to |
| 2001. This appears to have been fixed by |
| https://github.com/dateutil/dateutil/commit/fc696254. This parserinfo |
| implementation always honors the provided year to prevent this from |
| happening. |
| """ |
| def convertyear(self, year, century_specified=False): |
| """Overrides convertyear in dateutil.parser.parserinfo.""" |
| return int(year) |
| |
| |
| def _rfc3339_time_to_timestamp(time_str): |
| """Converts an RFC3339 time into a Unix timestamp. |
| |
| @param time_str: RFC3339-compatible time, e.g. |
| '2018-02-25T07:45:35.916929332-07:00'. |
| |
| @returns Float number of seconds since the Unix epoch. Negative if the time |
| precedes the epoch. |
| """ |
| dt = dateutil.parser.parse(time_str, parserinfo=_LessBrokenParserInfo()) |
| return (dt - _UNIX_EPOCH).total_seconds() |
