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