1#!/usr/bin/env python3
2#
3#   Copyright 2016 - The Android Open Source Project
4#
5#   Licensed under the Apache License, Version 2.0 (the "License");
6#   you may not use this file except in compliance with the License.
7#   You may obtain a copy of the License at
8#
9#       http://www.apache.org/licenses/LICENSE-2.0
10#
11#   Unless required by applicable law or agreed to in writing, software
12#   distributed under the License is distributed on an "AS IS" BASIS,
13#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14#   See the License for the specific language governing permissions and
15#   limitations under the License.
16import itertools
17import os
18import sys
19from builtins import str
20
21import mobly.config_parser as mobly_config_parser
22
23from acts import keys
24from acts import utils
25
26# An environment variable defining the base location for ACTS logs.
27_ENV_ACTS_LOGPATH = 'ACTS_LOGPATH'
28# An environment variable that enables test case failures to log stack traces.
29_ENV_TEST_FAILURE_TRACEBACKS = 'ACTS_TEST_FAILURE_TRACEBACKS'
30# An environment variable defining the test search paths for ACTS.
31_ENV_ACTS_TESTPATHS = 'ACTS_TESTPATHS'
32_PATH_SEPARATOR = ':'
33
34
35class ActsConfigError(Exception):
36    """Raised when there is a problem in test configuration file."""
37
38
39def _validate_test_config(test_config):
40    """Validates the raw configuration loaded from the config file.
41
42    Making sure all the required fields exist.
43    """
44    for k in keys.Config.reserved_keys.value:
45        # TODO(markdr): Remove this continue after merging this with the
46        # validation done in Mobly's load_test_config_file.
47        if (k == keys.Config.key_test_paths.value
48                or k == keys.Config.key_log_path.value):
49            continue
50
51        if k not in test_config:
52            raise ActsConfigError("Required key %s missing in test config." %
53                                  k)
54
55
56def _validate_testbed_name(name):
57    """Validates the name of a test bed.
58
59    Since test bed names are used as part of the test run id, it needs to meet
60    certain requirements.
61
62    Args:
63        name: The test bed's name specified in config file.
64
65    Raises:
66        If the name does not meet any criteria, ActsConfigError is raised.
67    """
68    if not name:
69        raise ActsConfigError("Test bed names can't be empty.")
70    if not isinstance(name, str):
71        raise ActsConfigError("Test bed names have to be string.")
72    for l in name:
73        if l not in utils.valid_filename_chars:
74            raise ActsConfigError(
75                "Char '%s' is not allowed in test bed names." % l)
76
77
78def _update_file_paths(config, config_path):
79    """ Checks if the path entries are valid.
80
81    If the file path is invalid, assume it is a relative path and append
82    that to the config file path.
83
84    Args:
85        config : the config object to verify.
86        config_path : The path to the config file, which can be used to
87                      generate absolute paths from relative paths in configs.
88
89    Raises:
90        If the file path is invalid, ActsConfigError is raised.
91    """
92    # Check the file_path_keys and update if it is a relative path.
93    for file_path_key in keys.Config.file_path_keys.value:
94        if file_path_key in config:
95            config_file = config[file_path_key]
96            if type(config_file) is str:
97                if not os.path.isfile(config_file):
98                    config_file = os.path.join(config_path, config_file)
99                if not os.path.isfile(config_file):
100                    raise ActsConfigError(
101                        "Unable to load config %s from test "
102                        "config file.", config_file)
103                config[file_path_key] = config_file
104
105
106def _validate_testbed_configs(testbed_configs, config_path):
107    """Validates the testbed configurations.
108
109    Args:
110        testbed_configs: A list of testbed configuration json objects.
111        config_path : The path to the config file, which can be used to
112                      generate absolute paths from relative paths in configs.
113
114    Raises:
115        If any part of the configuration is invalid, ActsConfigError is raised.
116    """
117    # Cross checks testbed configs for resource conflicts.
118    for name, config in testbed_configs.items():
119        _update_file_paths(config, config_path)
120        _validate_testbed_name(name)
121
122
123def gen_term_signal_handler(test_runners):
124    def termination_sig_handler(signal_num, frame):
125        print('Received sigterm %s.' % signal_num)
126        for t in test_runners:
127            t.stop()
128        sys.exit(1)
129
130    return termination_sig_handler
131
132
133def _parse_one_test_specifier(item):
134    """Parse one test specifier from command line input.
135
136    Args:
137        item: A string that specifies a test class or test cases in one test
138            class to run.
139
140    Returns:
141        A tuple of a string and a list of strings. The string is the test class
142        name, the list of strings is a list of test case names. The list can be
143        None.
144    """
145    tokens = item.split(':')
146    if len(tokens) > 2:
147        raise ActsConfigError("Syntax error in test specifier %s" % item)
148    if len(tokens) == 1:
149        # This should be considered a test class name
150        test_cls_name = tokens[0]
151        return test_cls_name, None
152    elif len(tokens) == 2:
153        # This should be considered a test class name followed by
154        # a list of test case names.
155        test_cls_name, test_case_names = tokens
156        clean_names = [elem.strip() for elem in test_case_names.split(',')]
157        return test_cls_name, clean_names
158
159
160def parse_test_list(test_list):
161    """Parse user provided test list into internal format for test_runner.
162
163    Args:
164        test_list: A list of test classes/cases.
165    """
166    result = []
167    for elem in test_list:
168        result.append(_parse_one_test_specifier(elem))
169    return result
170
171
172def load_test_config_file(test_config_path, tb_filters=None):
173    """Processes the test configuration file provided by the user.
174
175    Loads the configuration file into a json object, unpacks each testbed
176    config into its own TestRunConfig object, and validate the configuration in
177    the process.
178
179    Args:
180        test_config_path: Path to the test configuration file.
181        tb_filters: A subset of test bed names to be pulled from the config
182                    file. If None, then all test beds will be selected.
183
184    Returns:
185        A list of mobly.config_parser.TestRunConfig objects to be passed to
186        test_runner.TestRunner.
187    """
188    configs = utils.load_config(test_config_path)
189
190    testbeds = configs[keys.Config.key_testbed.value]
191    if type(testbeds) is list:
192        tb_dict = dict()
193        for testbed in testbeds:
194            tb_dict[testbed[keys.Config.key_testbed_name.value]] = testbed
195        testbeds = tb_dict
196    elif type(testbeds) is dict:
197        # For compatibility, make sure the entry name is the same as
198        # the testbed's "name" entry
199        for name, testbed in testbeds.items():
200            testbed[keys.Config.key_testbed_name.value] = name
201
202    if tb_filters:
203        tbs = {}
204        for name in tb_filters:
205            if name in testbeds:
206                tbs[name] = testbeds[name]
207            else:
208                raise ActsConfigError(
209                    'Expected testbed named "%s", but none was found. Check '
210                    'if you have the correct testbed names.' % name)
211        testbeds = tbs
212
213    if (keys.Config.key_log_path.value not in configs
214            and _ENV_ACTS_LOGPATH in os.environ):
215        print('Using environment log path: %s' %
216              (os.environ[_ENV_ACTS_LOGPATH]))
217        configs[keys.Config.key_log_path.value] = os.environ[_ENV_ACTS_LOGPATH]
218    if (keys.Config.key_test_paths.value not in configs
219            and _ENV_ACTS_TESTPATHS in os.environ):
220        print('Using environment test paths: %s' %
221              (os.environ[_ENV_ACTS_TESTPATHS]))
222        configs[keys.Config.key_test_paths.
223                value] = os.environ[_ENV_ACTS_TESTPATHS].split(_PATH_SEPARATOR)
224    if (keys.Config.key_test_failure_tracebacks not in configs
225            and _ENV_TEST_FAILURE_TRACEBACKS in os.environ):
226        configs[keys.Config.key_test_failure_tracebacks.
227                value] = os.environ[_ENV_TEST_FAILURE_TRACEBACKS]
228
229    # TODO: See if there is a better way to do this: b/29836695
230    config_path, _ = os.path.split(utils.abs_path(test_config_path))
231    configs[keys.Config.key_config_path.value] = config_path
232    _validate_test_config(configs)
233    _validate_testbed_configs(testbeds, config_path)
234    # Unpack testbeds into separate json objects.
235    configs.pop(keys.Config.key_testbed.value)
236    test_run_configs = []
237
238    for _, testbed in testbeds.items():
239        test_run_config = mobly_config_parser.TestRunConfig()
240        test_run_config.testbed_name = testbed[
241            keys.Config.key_testbed_name.value]
242        test_run_config.controller_configs = testbed
243        test_run_config.controller_configs[
244            keys.Config.key_test_paths.value] = configs.get(
245                keys.Config.key_test_paths.value, None)
246        test_run_config.log_path = configs.get(keys.Config.key_log_path.value,
247                                               None)
248        if test_run_config.log_path is not None:
249            test_run_config.log_path = utils.abs_path(test_run_config.log_path)
250
251        user_param_pairs = []
252        for item in itertools.chain(configs.items(), testbed.items()):
253            if item[0] not in keys.Config.reserved_keys.value:
254                user_param_pairs.append(item)
255        test_run_config.user_params = dict(user_param_pairs)
256
257        test_run_configs.append(test_run_config)
258    return test_run_configs
259
260
261def parse_test_file(fpath):
262    """Parses a test file that contains test specifiers.
263
264    Args:
265        fpath: A string that is the path to the test file to parse.
266
267    Returns:
268        A list of strings, each is a test specifier.
269    """
270    with open(fpath, 'r') as f:
271        tf = []
272        for line in f:
273            line = line.strip()
274            if not line:
275                continue
276            if len(tf) and (tf[-1].endswith(':') or tf[-1].endswith(',')):
277                tf[-1] += line
278            else:
279                tf.append(line)
280        return tf
281