xref: /external/autotest/site_utils/acts_lib.py

# 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.client.common_lib.cros import dev_server
from autotest_lib.server import adb_utils
from autotest_lib.server import constants
from autotest_lib.server.cros import dnsname_mangler
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,
                                      job_repo_url, 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 job_repo_url: The job repo url for where to pull build from.
    @param target_zip_file: The zip file to create on the teststation.

    @returns An ActsPackage containing all the information about the zipped
             artifact.
    """
    devserver_url = dev_server.AndroidBuildServer.get_server_url(job_repo_url)
    devserver = dev_server.AndroidBuildServer(devserver_url)
    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_enviroment(self,
                          container_directory,
                          testbed,
                          testbed_name=None,
                          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 testbed: The testbed that the test enviroment
                        will be testing on.
        @param testbed_name: An overriding name for the testbed.
        @param internal_acts_directory: The directory inside of the package
                                        that holds acts.

        @returns: An ActsTestingEnviroment with info on the unpacked
                  acts testing enviroment.
        """
        if testbed.teststation != self.test_station:
            raise error.TestError('Creating a contianer for a testbed on a '
                                  'different teststation is not allowed.')

        self.test_station.run('unzip "%s" -x -d "%s"' %
                              (self.zip_file, container_directory))

        return ActsTestingEnviroment(
            testbed=testbed,
            container_directory=container_directory,
            testbed_name=testbed_name,
            acts_directory=internal_acts_directory)


class AndroidTestingEnviroment(object):
    """A container for testing android devices on a test station."""

    def __init__(self, testbed, testbed_name=None):
        """Creates a new android testing enviroment.

        @param testbed: The testbed to test on.
        @param testbed_name: An overriding name for the testbed.
        """
        self.testbed = testbed

        if not testbed_name:
            # If no override is given get the name from the hostname.
            hostname = testbed.hostname
            if dnsname_mangler.is_ip_address(hostname):
                testbed_name = hostname
            else:
                testbed_name = hostname.split('.')[0]

        self.testbed_name = testbed_name

    def install_sl4a_apk(self, force_reinstall=True):
        """Install sl4a to a test bed.

        @param force_reinstall: If true the apk will be force to reinstall.
        """
        for serial, adb_host in self.testbed.get_adb_devices().iteritems():
            adb_utils.install_apk_from_build(
                adb_host,
                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 contianing 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 serial, adb_host in self.testbed.get_adb_devices().iteritems():
            adb_utils.install_apk_from_build(
                adb_host,
                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)

    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)
        result = 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

    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.setup_file)
        install_deps_command = '%s %s install_deps' % (python_bin,
                                                       self.setup_file)

        self.test_station.run(uninstall_command)
        self.test_station.run(install_deps_command)


class ActsTestingEnviroment(ActsContainer, AndroidTestingEnviroment):
    """A container for running acts tests with a contained version of acts."""

    def __init__(self,
                 container_directory,
                 testbed,
                 testbed_name=None,
                 acts_directory=None):
        """
        @param testbed: The testbed to test on.
        @param container_directory: The directory on the teststation this
                                    container operates out of.
        @param testbed_name: An overriding name for the testbed.
        @param acts_directory: The directory within the container that holds
                               acts. If none then it defaults to
                               DEFAULT_ACTS_INTERNAL_DIRECTORY.
        """
        AndroidTestingEnviroment.__init__(
            self, testbed, testbed_name=testbed_name)

        ActsContainer.__init__(
            self,
            testbed.teststation,
            container_directory=container_directory,
            acts_directory=acts_directory)

        self.config_location = os.path.join(self.container_directory,
                                            CONFIG_DIR_NAME)

        self.log_directory = os.path.join(self.container_directory,
                                          LOG_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)
        self.test_station.run('mkdir %s' % self.working_directory,
                              ignore_status=True)

        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.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.upload_file(full_name, full_name)
        self.campaigns[campaign_file] = full_path

        return full_path

    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.get_enviroment()

        # Setup enviroment 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.working_directory

        if additional_cmd_line_params:
            act_base_cmd = '%s %s -c %s -tb %s %s ' % (
                python_bin, self.acts_file, full_config, self.testbed_name,
                additional_cmd_line_params)
        else:
            act_base_cmd = '%s %s -c %s -tb %s ' % (
                python_bin, self.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.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,
            self.testbed,
            testbed_name=self.testbed_name,
            run_result=act_result,
            log_directory=self.log_directory,
            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,
                 testbed,
                 testbed_name=None,
                 run_result=None,
                 log_directory=None,
                 exception=None):
        """
        @param name: A name to identify the test run.
        @param testbed: The testbed that ran the test.
        @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 = log_directory
        self.test_station = testbed.teststation

        self.testbed = testbed
        if not testbed_name:
            # If no override is given get the name from the hostname.
            hostname = testbed.hostname
            if dnsname_mangler.is_ip_address(hostname):
                self.testbed_name = hostname
            else:
                self.testbed_name = hostname.split('.')[0]
        else:
            self.testbed_name = testbed_name

        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.
        """
        if self.testbed and self:
            self.testbed.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)