1#!/usr/bin/env python3.4
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.
16
17from builtins import str
18
19import os
20import random
21import sys
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        if k not in test_config:
46            raise ActsConfigError(
47                "Required key %s missing in test config." % k)
48
49
50def _validate_testbed_name(name):
51    """Validates the name of a test bed.
52
53    Since test bed names are used as part of the test run id, it needs to meet
54    certain requirements.
55
56    Args:
57        name: The test bed's name specified in config file.
58
59    Raises:
60        If the name does not meet any criteria, ActsConfigError is raised.
61    """
62    if not name:
63        raise ActsConfigError("Test bed names can't be empty.")
64    if not isinstance(name, str):
65        raise ActsConfigError("Test bed names have to be string.")
66    for l in name:
67        if l not in utils.valid_filename_chars:
68            raise ActsConfigError(
69                "Char '%s' is not allowed in test bed names." % l)
70
71
72def _update_file_paths(config, config_path):
73    """ Checks if the path entries are valid.
74
75    If the file path is invalid, assume it is a relative path and append
76    that to the config file path.
77
78    Args:
79        config : the config object to verify.
80        config_path : The path to the config file, which can be used to
81                      generate absolute paths from relative paths in configs.
82
83    Raises:
84        If the file path is invalid, ActsConfigError is raised.
85    """
86    # Check the file_path_keys and update if it is a relative path.
87    for file_path_key in keys.Config.file_path_keys.value:
88        if file_path_key in config:
89            config_file = config[file_path_key]
90            if type(config_file) is str:
91                if not os.path.isfile(config_file):
92                    config_file = os.path.join(config_path, config_file)
93                if not os.path.isfile(config_file):
94                    raise ActsConfigError("Unable to load config %s from test "
95                                          "config file.", config_file)
96                config[file_path_key] = config_file
97
98
99def _validate_testbed_configs(testbed_configs, config_path):
100    """Validates the testbed configurations.
101
102    Args:
103        testbed_configs: A list of testbed configuration json objects.
104        config_path : The path to the config file, which can be used to
105                      generate absolute paths from relative paths in configs.
106
107    Raises:
108        If any part of the configuration is invalid, ActsConfigError is raised.
109    """
110    # Cross checks testbed configs for resource conflicts.
111    for name, config in testbed_configs.items():
112        _update_file_paths(config, config_path)
113        _validate_testbed_name(name)
114
115
116def gen_term_signal_handler(test_runners):
117    def termination_sig_handler(signal_num, frame):
118        print('Received sigterm %s.' % signal_num)
119        for t in test_runners:
120            t.stop()
121        sys.exit(1)
122
123    return termination_sig_handler
124
125
126def _parse_one_test_specifier(item):
127    """Parse one test specifier from command line input.
128
129    This also verifies that the test class name and test case names follow
130    ACTS's naming conventions. A test class name has to end with "Test"; a test
131    case name has to start with "test".
132
133    Args:
134        item: A string that specifies a test class or test cases in one test
135            class to run.
136
137    Returns:
138        A tuple of a string and a list of strings. The string is the test class
139        name, the list of strings is a list of test case names. The list can be
140        None.
141    """
142    tokens = item.split(':')
143    if len(tokens) > 2:
144        raise ActsConfigError("Syntax error in test specifier %s" % item)
145    if len(tokens) == 1:
146        # This should be considered a test class name
147        test_cls_name = tokens[0]
148        return test_cls_name, None
149    elif len(tokens) == 2:
150        # This should be considered a test class name followed by
151        # a list of test case names.
152        test_cls_name, test_case_names = tokens
153        clean_names = []
154        for elem in test_case_names.split(','):
155            test_case_name = elem.strip()
156            if not test_case_name.startswith("test_"):
157                raise ActsConfigError(
158                    ("Requested test case '%s' in test class "
159                     "'%s' does not follow the test case "
160                     "naming convention test_*.") % (test_case_name,
161                                                     test_cls_name))
162            clean_names.append(test_case_name)
163        return test_cls_name, clean_names
164
165
166def parse_test_list(test_list):
167    """Parse user provided test list into internal format for test_runner.
168
169    Args:
170        test_list: A list of test classes/cases.
171    """
172    result = []
173    for elem in test_list:
174        result.append(_parse_one_test_specifier(elem))
175    return result
176
177
178def test_randomizer(test_identifiers, test_case_iterations=10):
179    """Generate test lists by randomizing user provided test list.
180
181    Args:
182        test_identifiers: A list of test classes/cases.
183        test_case_iterations: The range of random iterations for each case.
184    Returns:
185        A list of randomized test cases.
186    """
187    random_tests = []
188    preflight_tests = []
189    postflight_tests = []
190    for test_class, test_cases in test_identifiers:
191        if "Preflight" in test_class:
192            preflight_tests.append((test_class, test_cases))
193        elif "Postflight" in test_class:
194            postflight_tests.append((test_class, test_cases))
195        else:
196            for test_case in test_cases:
197                random_tests.append((test_class,
198                                     [test_case] * random.randrange(
199                                         1, test_case_iterations + 1)))
200    random.shuffle(random_tests)
201    new_tests = []
202    previous_class = None
203    for test_class, test_cases in random_tests:
204        if test_class == previous_class:
205            previous_cases = new_tests[-1][1]
206            previous_cases.extend(test_cases)
207        else:
208            new_tests.append((test_class, test_cases))
209        previous_class = test_class
210    return preflight_tests + new_tests + postflight_tests
211
212
213def load_test_config_file(test_config_path,
214                          tb_filters=None,
215                          override_test_path=None,
216                          override_log_path=None,
217                          override_test_args=None,
218                          override_random=None,
219                          override_test_case_iterations=None):
220    """Processes the test configuration file provided by the user.
221
222    Loads the configuration file into a json object, unpacks each testbed
223    config into its own json object, and validate the configuration in the
224    process.
225
226    Args:
227        test_config_path: Path to the test configuration file.
228        tb_filters: A subset of test bed names to be pulled from the config
229                    file. If None, then all test beds will be selected.
230        override_test_path: If not none the test path to use instead.
231        override_log_path: If not none the log path to use instead.
232        override_test_args: If not none the test args to use instead.
233        override_random: If not None, override the config file value.
234        override_test_case_iterations: If not None, override the config file
235                                       value.
236
237    Returns:
238        A list of test configuration json objects to be passed to
239        test_runner.TestRunner.
240    """
241    configs = utils.load_config(test_config_path)
242    if override_test_path:
243        configs[keys.Config.key_test_paths.value] = override_test_path
244    if override_log_path:
245        configs[keys.Config.key_log_path.value] = override_log_path
246    if override_test_args:
247        configs[keys.Config.ikey_cli_args.value] = override_test_args
248    if override_random:
249        configs[keys.Config.key_random.value] = override_random
250    if override_test_case_iterations:
251        configs[keys.Config.key_test_case_iterations.value] = \
252            override_test_case_iterations
253
254    testbeds = configs[keys.Config.key_testbed.value]
255    if type(testbeds) is list:
256        tb_dict = dict()
257        for testbed in testbeds:
258            tb_dict[testbed[keys.Config.key_testbed_name.value]] = testbed
259        testbeds = tb_dict
260    elif type(testbeds) is dict:
261        # For compatibility, make sure the entry name is the same as
262        # the testbed's "name" entry
263        for name, testbed in testbeds.items():
264            testbed[keys.Config.key_testbed_name.value] = name
265
266    if tb_filters:
267        tbs = {}
268        for name in tb_filters:
269            if name in testbeds:
270                tbs[name] = testbeds[name]
271            else:
272                raise ActsConfigError(
273                    'Expected testbed named "%s", but none was found. Check'
274                    'if you have the correct testbed names.' % name)
275        testbeds = tbs
276
277    if (keys.Config.key_log_path.value not in configs
278            and _ENV_ACTS_LOGPATH in os.environ):
279        print('Using environment log path: %s' %
280              (os.environ[_ENV_ACTS_LOGPATH]))
281        configs[keys.Config.key_log_path.value] = os.environ[_ENV_ACTS_LOGPATH]
282    if (keys.Config.key_test_paths.value not in configs
283            and _ENV_ACTS_TESTPATHS in os.environ):
284        print('Using environment test paths: %s' %
285              (os.environ[_ENV_ACTS_TESTPATHS]))
286        configs[keys.Config.key_test_paths.value] = os.environ[
287            _ENV_ACTS_TESTPATHS].split(_PATH_SEPARATOR)
288    if (keys.Config.key_test_failure_tracebacks not in configs
289            and _ENV_TEST_FAILURE_TRACEBACKS in os.environ):
290        configs[keys.Config.key_test_failure_tracebacks.value] = os.environ[
291            _ENV_TEST_FAILURE_TRACEBACKS]
292
293    # Add the global paths to the global config.
294    k_log_path = keys.Config.key_log_path.value
295    configs[k_log_path] = utils.abs_path(configs[k_log_path])
296
297    # TODO: See if there is a better way to do this: b/29836695
298    config_path, _ = os.path.split(utils.abs_path(test_config_path))
299    configs[keys.Config.key_config_path] = config_path
300    _validate_test_config(configs)
301    _validate_testbed_configs(testbeds, config_path)
302    # Unpack testbeds into separate json objects.
303    configs.pop(keys.Config.key_testbed.value)
304    config_jsons = []
305
306    for _, original_bed_config in testbeds.items():
307        new_test_config = dict(configs)
308        new_test_config[keys.Config.key_testbed.value] = original_bed_config
309        # Keys in each test bed config will be copied to a level up to be
310        # picked up for user_params. If the key already exists in the upper
311        # level, the local one defined in test bed config overwrites the
312        # general one.
313        new_test_config.update(original_bed_config)
314        config_jsons.append(new_test_config)
315    return config_jsons
316
317
318def parse_test_file(fpath):
319    """Parses a test file that contains test specifiers.
320
321    Args:
322        fpath: A string that is the path to the test file to parse.
323
324    Returns:
325        A list of strings, each is a test specifier.
326    """
327    with open(fpath, 'r') as f:
328        tf = []
329        for line in f:
330            line = line.strip()
331            if not line:
332                continue
333            if len(tf) and (tf[-1].endswith(':') or tf[-1].endswith(',')):
334                tf[-1] += line
335            else:
336                tf.append(line)
337        return tf
338