runners/host/config_parser.py

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

from builtins import str

import copy
import signal
import sys
import traceback

from vts.runners.host import keys
from vts.runners.host import errors
from vts.runners.host import signals
from vts.runners.host import utils

_DEFAULT_CONFIG_TEMPLATE = {
    "test_bed": {
        "AndroidDevice": "*",
    },
    "log_path": "/tmp/logs",
    "test_paths": ["./"],
    "enable_web": False,
}


def GetDefaultConfig(test_name):
    """Returns a default config data structure (when no config file is given)."""
    result = copy.deepcopy(_DEFAULT_CONFIG_TEMPLATE)
    result[keys.ConfigKeys.KEY_TESTBED][
        keys.ConfigKeys.KEY_TESTBED_NAME] = test_name
    return result


def gen_term_signal_handler(test_runners):
    """Generates a termination signal handler function.

    Args:
        test_runners: A list of TestRunner objects.

    Returns:
        A function to be called when termination signals are received from
        command line. This function stops all TestRunner objects.
    """

    def termination_sig_handler(signal_num, frame):
        for t in test_runners:
            t.stop()
        sys.exit(1)

    return termination_sig_handler


def load_test_config_file(test_config_path,
                          tb_filters=None,
                          baseline_config=None):
    """Processes the test configuration file provided by user.

    Loads the configuration file into a json object, unpacks each testbed
    config into its own json object, and validate the configuration in the
    process.

    Args:
        test_config_path: Path to the test configuration file.
        tb_filters: A list of strings, each is a test bed name. If None, all
                    test beds are picked up. Otherwise only test bed names
                    specified will be picked up.
        baseline_config: dict, the baseline config to use (used iff
                         test_config_path does not have device info).

    Returns:
        A list of test configuration json objects to be passed to TestRunner.
    """
    try:
        configs = utils.load_config(test_config_path)
        if keys.ConfigKeys.KEY_TESTBED not in configs and baseline_config:
            configs.update(baseline_config)

        if tb_filters:
            tbs = []
            for tb in configs[keys.ConfigKeys.KEY_TESTBED]:
                if tb[keys.ConfigKeys.KEY_TESTBED_NAME] in tb_filters:
                    tbs.append(tb)
            if len(tbs) != len(tb_filters):
                print("Expect to find %d test bed configs, found %d." %
                      (len(tb_filters), len(tbs)))
                print("Check if you have the correct test bed names.")
                return None
            configs[keys.ConfigKeys.KEY_TESTBED] = tbs
        _validate_test_config(configs)
        _validate_testbed_configs(configs[keys.ConfigKeys.KEY_TESTBED])
        k_log_path = keys.ConfigKeys.KEY_LOG_PATH
        configs[k_log_path] = utils.abs_path(configs[k_log_path])
        tps = configs[keys.ConfigKeys.KEY_TEST_PATHS]
    except errors.USERError as e:
        print("Something is wrong in the test configurations.")
        print(str(e))
        return None
    except Exception as e:
        print("Error loading test config {}".format(test_config_path))
        print(traceback.format_exc())
        return None
    # Unpack testbeds into separate json objects.
    beds = configs.pop(keys.ConfigKeys.KEY_TESTBED)
    config_jsons = []
    for original_bed_config in beds:
        new_test_config = dict(configs)
        new_test_config[keys.ConfigKeys.KEY_TESTBED] = original_bed_config
        # Keys in each test bed config will be copied to a level up to be
        # picked up for user_params. If the key already exists in the upper
        # level, the local one defined in test bed config overwrites the
        # general one.
        new_test_config.update(original_bed_config)
        config_jsons.append(new_test_config)
    return config_jsons


def parse_test_list(test_list):
    """Parse user provided test list into internal format for test_runner.

    Args:
        test_list: A list of test classes/cases.

    Returns:
        A list of tuples, each has a test class name and a list of test case
        names.
    """
    result = []
    for elem in test_list:
        result.append(_parse_one_test_specifier(elem))
    return result


def _validate_test_config(test_config):
    """Validates the raw configuration loaded from the config file.

    Making sure all the required keys exist.

    Args:
        test_config: A dict that is the config to validate.

    Raises:
        errors.USERError is raised if any required key is missing from the
        config.
    """
    for k in keys.ConfigKeys.RESERVED_KEYS:
        if k not in test_config:
            raise errors.USERError(("Required key {} missing in test "
                                    "config.").format(k))


def _parse_one_test_specifier(item):
    """Parse one test specifier from command line input.

    This also verifies that the test class name and test case names follow
    ACTS's naming conventions. A test class name has to end with "Test"; a test
    case name has to start with "test".

    Args:
        item: A string that specifies a test class or test cases in one test
            class to run.

    Returns:
        A tuple of a string and a list of strings. The string is the test class
        name, the list of strings is a list of test case names. The list can be
        None.
    """
    tokens = item.split(':')
    if len(tokens) > 2:
        raise errors.USERError("Syntax error in test specifier %s" % item)
    if len(tokens) == 1:
        # This should be considered a test class name
        test_cls_name = tokens[0]
        _validate_test_class_name(test_cls_name)
        return (test_cls_name, None)
    elif len(tokens) == 2:
        # This should be considered a test class name followed by
        # a list of test case names.
        test_cls_name, test_case_names = tokens
        clean_names = []
        _validate_test_class_name(test_cls_name)
        for elem in test_case_names.split(','):
            test_case_name = elem.strip()
            if not test_case_name.startswith("test_"):
                raise errors.USERError(
                    ("Requested test case '%s' in test class "
                     "'%s' does not follow the test case "
                     "naming convention test_*.") % (test_case_name,
                                                     test_cls_name))
            clean_names.append(test_case_name)
        return (test_cls_name, clean_names)


def _parse_test_file(fpath):
    """Parses a test file that contains test specifiers.

    Args:
        fpath: A string that is the path to the test file to parse.

    Returns:
        A list of strings, each is a test specifier.
    """
    try:
        with open(fpath, 'r') as f:
            tf = []
            for line in f:
                line = line.strip()
                if not line:
                    continue
                if len(tf) and (tf[-1].endswith(':') or tf[-1].endswith(',')):
                    tf[-1] += line
                else:
                    tf.append(line)
            return tf
    except:
        print("Error loading test file.")
        raise


def _validate_test_class_name(test_cls_name):
    """Checks if a string follows the test class name convention.

    Args:
        test_cls_name: A string that should be a test class name.

    Raises:
        errors.USERError is raised if the input does not follow test class
        naming convention.
    """
    if not test_cls_name.endswith("Test"):
        raise errors.USERError(
            ("Requested test class '%s' does not follow the test "
             "class naming convention *Test.") % test_cls_name)


def _validate_testbed_configs(testbed_configs):
    """Validates the testbed configurations.

    Args:
        testbed_configs: A list of testbed configuration json objects.

    Raises:
        If any part of the configuration is invalid, errors.USERError is raised.
    """
    seen_names = set()
    # Cross checks testbed configs for resource conflicts.
    for config in testbed_configs:
        # Check for conflicts between multiple concurrent testbed configs.
        # No need to call it if there's only one testbed config.
        name = config[keys.ConfigKeys.KEY_TESTBED_NAME]
        _validate_testbed_name(name)
        # Test bed names should be unique.
        if name in seen_names:
            raise errors.USERError("Duplicate testbed name {} found.".format(
                name))
        seen_names.add(name)


def _validate_testbed_name(name):
    """Validates the name of a test bed.

    Since test bed names are used as part of the test run id, it needs to meet
    certain requirements.

    Args:
        name: The test bed's name specified in config file.

    Raises:
        If the name does not meet any criteria, errors.USERError is raised.
    """
    if not name:
        raise errors.USERError("Test bed names can't be empty.")
    if not isinstance(name, str) and not isinstance(name, basestring):
        raise errors.USERError("Test bed names have to be string. Found: %s" %
                               type(name))
    for l in name:
        if l not in utils.valid_filename_chars:
            raise errors.USERError(
                "Char '%s' is not allowed in test bed names." % l)