# Copyright 2016 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import json
import logging
import os
from autotest_lib.client.common_lib import error
from autotest_lib.client.common_lib import global_config
from autotest_lib.server import adb_utils
from autotest_lib.server import constants
from autotest_lib.server.hosts import adb_host
DEFAULT_ACTS_INTERNAL_DIRECTORY = 'tools/test/connectivity/acts'
CONFIG_FOLDER_LOCATION = global_config.global_config.get_config_value(
'ACTS', 'acts_config_folder', default='')
TEST_DIR_NAME = 'tests'
FRAMEWORK_DIR_NAME = 'framework'
SETUP_FILE_NAME = 'setup.py'
CONFIG_DIR_NAME = 'autotest_config'
CAMPAIGN_DIR_NAME = 'autotest_campaign'
LOG_DIR_NAME = 'logs'
ACTS_EXECUTABLE_IN_FRAMEWORK = 'acts/bin/act.py'
ACTS_TESTPATHS_ENV_KEY = 'ACTS_TESTPATHS'
ACTS_LOGPATH_ENV_KEY = 'ACTS_LOGPATH'
ACTS_PYTHONPATH_ENV_KEY = 'PYTHONPATH'
def create_acts_package_from_current_artifact(test_station, job_repo_url,
target_zip_file):
"""Creates an acts package from the build branch being used.
Creates an acts artifact from the build branch being used. This is
determined by the job_repo_url passed in.
@param test_station: The teststation that should be creating the package.
@param job_repo_url: The job_repo_url to get the build info from.
@param target_zip_file: The zip file to create form the artifact on the
test_station.
@returns An ActsPackage containing all the information about the zipped
artifact.
"""
build_info = adb_host.ADBHost.get_build_info_from_build_url(job_repo_url)
return create_acts_package_from_artifact(
test_station, build_info['branch'], build_info['target'],
build_info['build_id'], job_repo_url, target_zip_file)
def create_acts_package_from_artifact(test_station, branch, target, build_id,
devserver, target_zip_file):
"""Creates an acts package from a specified branch.
Grabs the packaged acts artifact from the branch and places it on the
test_station.
@param test_station: The teststation that should be creating the package.
@param branch: The name of the branch where the artifact is to be pulled.
@param target: The name of the target where the artifact is to be pulled.
@param build_id: The build id to pull the artifact from.
@param devserver: The devserver to use.
@param target_zip_file: The zip file to create on the teststation.
@returns An ActsPackage containing all the information about the zipped
artifact.
"""
devserver.trigger_download(
target, build_id, branch, files='acts.zip', synchronous=True)
pull_base_url = devserver.get_pull_url(target, build_id, branch)
download_ulr = os.path.join(pull_base_url, 'acts.zip')
test_station.download_file(download_ulr, target_zip_file)
return ActsPackage(test_station, target_zip_file)
def create_acts_package_from_zip(test_station, zip_location, target_zip_file):
"""Creates an acts package from an existing zip.
Creates an acts package from a zip file that already sits on the drone.
@param test_station: The teststation to create the package on.
@param zip_location: The location of the zip on the drone.
@param target_zip_file: The zip file to create on the teststaiton.
@returns An ActsPackage containing all the information about the zipped
artifact.
"""
if not os.path.isabs(zip_location):
zip_location = os.path.join(CONFIG_FOLDER_LOCATION, 'acts_artifacts',
zip_location)
test_station.send_file(zip_location, target_zip_file)
return ActsPackage(test_station, target_zip_file)
class ActsPackage(object):
"""A packaged version of acts on a teststation."""
def __init__(self, test_station, zip_file_path):
"""
@param test_station: The teststation this package is on.
@param zip_file_path: The path to the zip file on the test station that
holds the package on the teststation.
"""
self.test_station = test_station
self.zip_file = zip_file_path
def create_container(self,
container_directory,
internal_acts_directory=None):
"""Unpacks this package into a container.
Unpacks this acts package into a container to interact with acts.
@param container_directory: The directory on the teststation to hold
the container.
@param internal_acts_directory: The directory inside of the package
that holds acts.
@returns: An ActsContainer with info on the unpacked acts container.
"""
self.test_station.run('unzip "%s" -x -d "%s"' %
(self.zip_file, container_directory))
return ActsContainer(
self.test_station,
container_directory,
acts_directory=internal_acts_directory)
def create_environment(self,
container_directory,
devices,
testbed_name,
internal_acts_directory=None):
"""Unpacks this package into an acts testing enviroment.
Unpacks this acts package into a test enviroment to test with acts.
@param container_directory: The directory on the teststation to hold
the test enviroment.
@param devices: The list of devices in the environment.
@param testbed_name: The name of the testbed.
@param internal_acts_directory: The directory inside of the package
that holds acts.
@returns: An ActsTestingEnvironment with info on the unpacked
acts testing environment.
"""
container = self.create_container(container_directory,
internal_acts_directory)
return ActsTestingEnviroment(
devices=devices,
container=container,
testbed_name=testbed_name)
class AndroidTestingEnvironment(object):
"""A container for testing android devices on a test station."""
def __init__(self, devices, testbed_name):
"""Creates a new android testing environment.
@param devices: The devices on the testbed to use.
@param testbed_name: The name for the testbed.
"""
self.devices = devices
self.testbed_name = testbed_name
def install_sl4a_apk(self, force_reinstall=True):
"""Install sl4a to all provided devices..
@param force_reinstall: If true the apk will be force to reinstall.
"""
for device in self.devices:
adb_utils.install_apk_from_build(
device,
constants.SL4A_APK,
constants.SL4A_ARTIFACT,
package_name=constants.SL4A_PACKAGE,
force_reinstall=force_reinstall)
def install_apk(self, apk_info, force_reinstall=True):
"""Installs an additional apk on all adb devices.
@param apk_info: A dictionary containing the apk info. This dictionary
should contain the keys:
apk="Name of the apk",
package="Name of the package".
artifact="Name of the artifact", if missing
the package name is used."
@param force_reinstall: If true the apk will be forced to reinstall.
"""
for device in self.devices:
adb_utils.install_apk_from_build(
device,
apk_info['apk'],
apk_info.get('artifact') or constants.SL4A_ARTIFACT,
package_name=apk_info['package'],
force_reinstall=force_reinstall)
class ActsContainer(object):
"""A container for working with acts."""
def __init__(self, test_station, container_directory, acts_directory=None):
"""
@param test_station: The test station that the container is on.
@param container_directory: The directory on the teststation this
container operates out of.
@param acts_directory: The directory within the container that holds
acts. If none then it defaults to
DEFAULT_ACTS_INTERNAL_DIRECTORY.
"""
self.test_station = test_station
self.container_directory = container_directory
if not acts_directory:
acts_directory = DEFAULT_ACTS_INTERNAL_DIRECTORY
if not os.path.isabs(acts_directory):
self.acts_directory = os.path.join(container_directory,
acts_directory)
else:
self.acts_directory = acts_directory
self.tests_directory = os.path.join(self.acts_directory, TEST_DIR_NAME)
self.framework_directory = os.path.join(self.acts_directory,
FRAMEWORK_DIR_NAME)
self.acts_file = os.path.join(self.framework_directory,
ACTS_EXECUTABLE_IN_FRAMEWORK)
self.setup_file = os.path.join(self.framework_directory,
SETUP_FILE_NAME)
self.log_directory = os.path.join(container_directory,
LOG_DIR_NAME)
self.config_location = os.path.join(container_directory,
CONFIG_DIR_NAME)
self.acts_file = os.path.join(self.framework_directory,
ACTS_EXECUTABLE_IN_FRAMEWORK)
self.working_directory = os.path.join(container_directory,
CONFIG_DIR_NAME)
test_station.run('mkdir %s' % self.working_directory,
ignore_status=True)
def get_test_paths(self):
"""Get all test paths within this container.
Gets all paths that hold tests within the container.
@returns: A list of paths on the teststation that hold tests.
"""
get_test_paths_result = self.test_station.run('find %s -type d' %
self.tests_directory)
test_search_dirs = get_test_paths_result.stdout.splitlines()
return test_search_dirs
def get_python_path(self):
"""Get the python path being used.
Gets the python path that will be set in the enviroment for this
container.
@returns: A string of the PYTHONPATH enviroment variable to be used.
"""
return '%s:$PYTHONPATH' % self.framework_directory
def get_enviroment(self):
"""Gets the enviroment variables to be used for this container.
@returns: A dictionary of enviroment variables to be used by this
container.
"""
env = {
ACTS_TESTPATHS_ENV_KEY: ':'.join(self.get_test_paths()),
ACTS_LOGPATH_ENV_KEY: self.log_directory,
ACTS_PYTHONPATH_ENV_KEY: self.get_python_path()
}
return env
def upload_file(self, src, dst):
"""Uploads a file to be used by the container.
Uploads a file from the drone to the test staiton to be used by the
test container.
@param src: The source file on the drone. If a relative path is given
it is assumed to exist in CONFIG_FOLDER_LOCATION.
@param dst: The destination on the teststation. If a relative path is
given it is assumed that it is within the container.
@returns: The full path on the teststation.
"""
if not os.path.isabs(src):
src = os.path.join(CONFIG_FOLDER_LOCATION, src)
if not os.path.isabs(dst):
dst = os.path.join(self.container_directory, dst)
path = os.path.dirname(dst)
self.test_station.run('mkdir "%s"' % path, ignore_status=True)
original_dst = dst
if os.path.basename(src) == os.path.basename(dst):
dst = os.path.dirname(dst)
self.test_station.send_file(src, dst)
return original_dst
class ActsTestingEnviroment(AndroidTestingEnvironment):
"""A container for running acts tests with a contained version of acts."""
def __init__(self, container, devices, testbed_name):
"""
@param container: The acts container to use.
@param devices: The list of devices to use.
@testbed_name: The name of the testbed being used.
"""
super(ActsTestingEnviroment, self).__init__(devices=devices,
testbed_name=testbed_name)
self.container = container
self.configs = {}
self.campaigns = {}
def upload_config(self, config_file):
"""Uploads a config file to the container.
Uploads a config file to the config folder in the container.
@param config_file: The config file to upload. This must be a file
within the autotest_config directory under the
CONFIG_FOLDER_LOCATION.
@returns: The full path of the config on the test staiton.
"""
full_name = os.path.join(CONFIG_DIR_NAME, config_file)
full_path = self.container.upload_file(full_name, full_name)
self.configs[config_file] = full_path
return full_path
def upload_campaign(self, campaign_file):
"""Uploads a campaign file to the container.
Uploads a campaign file to the campaign folder in the container.
@param campaign_file: The campaign file to upload. This must be a file
within the autotest_campaign directory under the
CONFIG_FOLDER_LOCATION.
@returns: The full path of the campaign on the test staiton.
"""
full_name = os.path.join(CAMPAIGN_DIR_NAME, campaign_file)
full_path = self.container.upload_file(full_name, full_name)
self.campaigns[campaign_file] = full_path
return full_path
def setup_enviroment(self, python_bin='python'):
"""Sets up the teststation system enviroment so the container can run.
Prepares the remote system so that the container can run. This involves
uninstalling all versions of acts for the version of python being
used and installing all needed dependencies.
@param python_bin: The python binary to use.
"""
uninstall_command = '%s %s uninstall' % (
python_bin, self.container.setup_file)
install_deps_command = '%s %s install_deps' % (
python_bin, self.container.setup_file)
self.container.test_station.run(uninstall_command)
self.container.test_station.run(install_deps_command)
def run_test(self,
config,
campaign=None,
test_case=None,
extra_env={},
python_bin='python',
timeout=7200,
additional_cmd_line_params=None):
"""Runs a test within the container.
Runs a test within a container using the given settings.
@param config: The name of the config file to use as the main config.
This should have already been uploaded with
upload_config. The string passed into upload_config
should be used here.
@param campaign: The campaign file to use for this test. If none then
test_case is assumed. This file should have already
been uploaded with upload_campaign. The string passed
into upload_campaign should be used here.
@param test_case: The test case to run the test with. If none then the
campaign will be used. If multiple are given,
multiple will be run.
@param extra_env: Extra enviroment variables to run the test with.
@param python_bin: The python binary to execute the test with.
@param timeout: How many seconds to wait before timing out.
@param additional_cmd_line_params: Adds the ability to add any string
to the end of the acts.py command
line string. This is intended to
add acts command line flags however
this is unbounded so it could cause
errors if incorrectly set.
@returns: The results of the test run.
"""
if not config in self.configs:
# Check if the config has been uploaded and upload if it hasn't
self.upload_config(config)
full_config = self.configs[config]
if campaign:
# When given a campaign check if it's upload.
if not campaign in self.campaigns:
self.upload_campaign(campaign)
full_campaign = self.campaigns[campaign]
else:
full_campaign = None
full_env = self.container.get_enviroment()
# Setup environment variables.
if extra_env:
for k, v in extra_env.items():
full_env[k] = extra_env
logging.info('Using env: %s', full_env)
exports = ('export %s=%s' % (k, v) for k, v in full_env.items())
env_command = ';'.join(exports)
# Make sure to execute in the working directory.
command_setup = 'cd %s' % self.container.working_directory
if additional_cmd_line_params:
act_base_cmd = '%s %s -c %s -tb %s %s ' % (
python_bin, self.container.acts_file, full_config,
self.testbed_name, additional_cmd_line_params)
else:
act_base_cmd = '%s %s -c %s -tb %s ' % (
python_bin, self.container.acts_file, full_config,
self.testbed_name)
# Format the acts command based on what type of test is being run.
if test_case and campaign:
raise error.TestError(
'campaign and test_file cannot both have a value.')
elif test_case:
if isinstance(test_case, str):
test_case = [test_case]
if len(test_case) < 1:
raise error.TestError('At least one test case must be given.')
tc_str = ''
for tc in test_case:
tc_str = '%s %s' % (tc_str, tc)
tc_str = tc_str.strip()
act_cmd = '%s -tc %s' % (act_base_cmd, tc_str)
elif campaign:
act_cmd = '%s -tf %s' % (act_base_cmd, full_campaign)
else:
raise error.TestFail('No tests was specified!')
# Format all commands into a single command.
command_list = [command_setup, env_command, act_cmd]
full_command = '; '.join(command_list)
try:
# Run acts on the remote machine.
act_result = self.container.test_station.run(full_command,
timeout=timeout)
excep = None
except Exception as e:
# Catch any error to store in the results.
act_result = None
excep = e
return ActsTestResults(str(test_case) or campaign,
container=self.container,
devices=self.devices,
testbed_name=self.testbed_name,
run_result=act_result,
exception=excep)
class ActsTestResults(object):
"""The packaged results of a test run."""
acts_result_to_autotest = {
'PASS': 'GOOD',
'FAIL': 'FAIL',
'UNKNOWN': 'WARN',
'SKIP': 'ABORT'
}
def __init__(self,
name,
container,
devices,
testbed_name,
run_result=None,
exception=None):
"""
@param name: A name to identify the test run.
@param testbed_name: The name the testbed was run with, if none the
default name of the testbed is used.
@param run_result: The raw i/o result of the test run.
@param log_directory: The directory that acts logged to.
@param exception: An exception that was thrown while running the test.
"""
self.name = name
self.run_result = run_result
self.exception = exception
self.log_directory = container.log_directory
self.test_station = container.test_station
self.testbed_name = testbed_name
self.devices = devices
self.reported_to = set()
self.json_results = {}
self.results_dir = None
if self.log_directory:
self.results_dir = os.path.join(self.log_directory,
self.testbed_name, 'latest')
results_file = os.path.join(self.results_dir,
'test_run_summary.json')
cat_log_result = self.test_station.run('cat %s' % results_file,
ignore_status=True)
if not cat_log_result.exit_status:
self.json_results = json.loads(cat_log_result.stdout)
def log_output(self):
"""Logs the output of the test."""
if self.run_result:
logging.debug('ACTS Output:\n%s', self.run_result.stdout)
def save_test_info(self, test):
"""Save info about the test.
@param test: The test to save.
"""
for device in self.devices:
device.save_info(test.resultsdir)
def rethrow_exception(self):
"""Re-throws the exception thrown during the test."""
if self.exception:
raise self.exception
def upload_to_local(self, local_dir):
"""Saves all acts results to a local directory.
@param local_dir: The directory on the local machine to save all results
to.
"""
if self.results_dir:
self.test_station.get_file(self.results_dir, local_dir)
def report_to_autotest(self, test):
"""Reports the results to an autotest test object.
Reports the results to the test and saves all acts results under the
tests results directory.
@param test: The autotest test object to report to. If this test object
has already recived our report then this call will be
ignored.
"""
if test in self.reported_to:
return
if self.results_dir:
self.upload_to_local(test.resultsdir)
if not 'Results' in self.json_results:
return
results = self.json_results['Results']
for result in results:
verdict = self.acts_result_to_autotest[result['Result']]
details = result['Details']
test.job.record(verdict, None, self.name, status=(details or ''))
self.reported_to.add(test)