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
17
18from future import standard_library
19
20standard_library.install_aliases()
21
22import importlib
23import inspect
24import fnmatch
25import json
26import logging
27import os
28import pkgutil
29import sys
30
31from acts import base_test
32from acts import keys
33from acts import logger
34from acts import records
35from acts import signals
36from acts import utils
37from acts import error
38
39from mobly.records import ExceptionRecord
40
41
42def _find_test_class():
43    """Finds the test class in a test script.
44
45    Walk through module members and find the subclass of BaseTestClass. Only
46    one subclass is allowed in a test script.
47
48    Returns:
49        The test class in the test module.
50    """
51    test_classes = []
52    main_module_members = sys.modules['__main__']
53    for _, module_member in main_module_members.__dict__.items():
54        if inspect.isclass(module_member):
55            if issubclass(module_member, base_test.BaseTestClass):
56                test_classes.append(module_member)
57    if len(test_classes) != 1:
58        logging.error('Expected 1 test class per file, found %s.',
59                      [t.__name__ for t in test_classes])
60        sys.exit(1)
61    return test_classes[0]
62
63
64def execute_one_test_class(test_class, test_config, test_identifier):
65    """Executes one specific test class.
66
67    You could call this function in your own cli test entry point if you choose
68    not to use act.py.
69
70    Args:
71        test_class: A subclass of acts.base_test.BaseTestClass that has the test
72                    logic to be executed.
73        test_config: A dict representing one set of configs for a test run.
74        test_identifier: A list of tuples specifying which test cases to run in
75                         the test class.
76
77    Returns:
78        True if all tests passed without any error, False otherwise.
79
80    Raises:
81        If signals.TestAbortAll is raised by a test run, pipe it through.
82    """
83    tr = TestRunner(test_config, test_identifier)
84    try:
85        tr.run(test_class)
86        return tr.results.is_all_pass
87    except signals.TestAbortAll:
88        raise
89    except:
90        logging.exception('Exception when executing %s.', tr.testbed_name)
91    finally:
92        tr.stop()
93
94
95class TestRunner(object):
96    """The class that instantiates test classes, executes test cases, and
97    report results.
98
99    Attributes:
100        test_run_config: The TestRunConfig object specifying what tests to run.
101        id: A string that is the unique identifier of this test run.
102        log: The logger object used throughout this test run.
103        test_classes: A dictionary where we can look up the test classes by name
104            to instantiate. Supports unix shell style wildcards.
105        run_list: A list of tuples specifying what tests to run.
106        results: The test result object used to record the results of this test
107            run.
108        running: A boolean signifies whether this test run is ongoing or not.
109    """
110    def __init__(self, test_configs, run_list):
111        self.test_run_config = test_configs
112        self.testbed_name = self.test_run_config.testbed_name
113        start_time = logger.get_log_file_timestamp()
114        self.id = '{}@{}'.format(self.testbed_name, start_time)
115        self.test_run_config.log_path = os.path.abspath(
116            os.path.join(self.test_run_config.log_path, self.testbed_name,
117                         start_time))
118        logger.setup_test_logger(self.log_path, self.testbed_name)
119        self.log = logging.getLogger()
120        self.test_run_config.summary_writer = records.TestSummaryWriter(
121            os.path.join(self.log_path, records.OUTPUT_FILE_SUMMARY))
122        self.run_list = run_list
123        self.dump_config()
124        self.results = records.TestResult()
125        self.running = False
126
127    @property
128    def log_path(self):
129        """The path to write logs of this test run to."""
130        return self.test_run_config.log_path
131
132    @property
133    def summary_writer(self):
134        """The object responsible for writing summary and results data."""
135        return self.test_run_config.summary_writer
136
137    def import_test_modules(self, test_paths):
138        """Imports test classes from test scripts.
139
140        1. Locate all .py files under test paths.
141        2. Import the .py files as modules.
142        3. Find the module members that are test classes.
143        4. Categorize the test classes by name.
144
145        Args:
146            test_paths: A list of directory paths where the test files reside.
147
148        Returns:
149            A dictionary where keys are test class name strings, values are
150            actual test classes that can be instantiated.
151        """
152        def is_testfile_name(name, ext):
153            if ext == '.py':
154                if name.endswith('Test') or name.endswith('_test'):
155                    return True
156            return False
157
158        file_list = utils.find_files(test_paths, is_testfile_name)
159        test_classes = {}
160        for path, name, _ in file_list:
161            sys.path.append(path)
162            try:
163                with utils.SuppressLogOutput(
164                        log_levels=[logging.INFO, logging.ERROR]):
165                    module = importlib.import_module(name)
166            except:
167                for test_cls_name, _ in self.run_list:
168                    alt_name = name.replace('_', '').lower()
169                    alt_cls_name = test_cls_name.lower()
170                    # Only block if a test class on the run list causes an
171                    # import error. We need to check against both naming
172                    # conventions: AaaBbb and aaa_bbb.
173                    if name == test_cls_name or alt_name == alt_cls_name:
174                        msg = ('Encountered error importing test class %s, '
175                               'abort.') % test_cls_name
176                        # This exception is logged here to help with debugging
177                        # under py2, because "raise X from Y" syntax is only
178                        # supported under py3.
179                        self.log.exception(msg)
180                        raise ValueError(msg)
181                continue
182            for member_name in dir(module):
183                if not member_name.startswith('__'):
184                    if member_name.endswith('Test'):
185                        test_class = getattr(module, member_name)
186                        if inspect.isclass(test_class):
187                            test_classes[member_name] = test_class
188        return test_classes
189
190    def set_test_util_logs(self, module=None):
191        """Sets the log object to each test util module.
192
193        This recursively include all modules under acts.test_utils and sets the
194        main test logger to each module.
195
196        Args:
197            module: A module under acts.test_utils.
198        """
199        # Initial condition of recursion.
200        if not module:
201            module = importlib.import_module('acts.test_utils')
202        # Somehow pkgutil.walk_packages is not working for me.
203        # Using iter_modules for now.
204        pkg_iter = pkgutil.iter_modules(module.__path__, module.__name__ + '.')
205        for _, module_name, ispkg in pkg_iter:
206            m = importlib.import_module(module_name)
207            if ispkg:
208                self.set_test_util_logs(module=m)
209            else:
210                self.log.debug('Setting logger to test util module %s',
211                               module_name)
212                setattr(m, 'log', self.log)
213
214    def run_test_class(self, test_cls_name, test_cases=None):
215        """Instantiates and executes a test class.
216
217        If test_cases is None, the test cases listed by self.tests will be
218        executed instead. If self.tests is empty as well, no test case in this
219        test class will be executed.
220
221        Args:
222            test_cls_name: Name of the test class to execute.
223            test_cases: List of test case names to execute within the class.
224
225        Raises:
226            ValueError is raised if the requested test class could not be found
227            in the test_paths directories.
228        """
229        matches = fnmatch.filter(self.test_classes.keys(), test_cls_name)
230        if not matches:
231            self.log.info(
232                'Cannot find test class %s or classes matching pattern, '
233                'skipping for now.' % test_cls_name)
234            record = records.TestResultRecord('*all*', test_cls_name)
235            record.test_skip(signals.TestSkip('Test class does not exist.'))
236            self.results.add_record(record)
237            return
238        if matches != [test_cls_name]:
239            self.log.info('Found classes matching pattern %s: %s',
240                          test_cls_name, matches)
241
242        for test_cls_name_match in matches:
243            test_cls = self.test_classes[test_cls_name_match]
244            test_cls_instance = test_cls(self.test_run_config)
245            try:
246                cls_result = test_cls_instance.run(test_cases)
247                self.results += cls_result
248            except signals.TestAbortAll as e:
249                self.results += e.results
250                raise e
251
252    def run(self, test_class=None):
253        """Executes test cases.
254
255        This will instantiate controller and test classes, and execute test
256        classes. This can be called multiple times to repeatedly execute the
257        requested test cases.
258
259        A call to TestRunner.stop should eventually happen to conclude the life
260        cycle of a TestRunner.
261
262        Args:
263            test_class: The python module of a test class. If provided, run this
264                        class; otherwise, import modules in under test_paths
265                        based on run_list.
266        """
267        if not self.running:
268            self.running = True
269
270        if test_class:
271            self.test_classes = {test_class.__name__: test_class}
272        else:
273            t_paths = self.test_run_config.controller_configs[
274                keys.Config.key_test_paths.value]
275            self.test_classes = self.import_test_modules(t_paths)
276        self.log.debug('Executing run list %s.', self.run_list)
277        for test_cls_name, test_case_names in self.run_list:
278            if not self.running:
279                break
280
281            if test_case_names:
282                self.log.debug('Executing test cases %s in test class %s.',
283                               test_case_names, test_cls_name)
284            else:
285                self.log.debug('Executing test class %s', test_cls_name)
286
287            try:
288                self.run_test_class(test_cls_name, test_case_names)
289            except error.ActsError as e:
290                self.results.error.append(ExceptionRecord(e))
291                self.log.error('Test Runner Error: %s' % e.message)
292            except signals.TestAbortAll as e:
293                self.log.warning(
294                    'Abort all subsequent test classes. Reason: %s', e)
295                raise
296
297    def stop(self):
298        """Releases resources from test run. Should always be called after
299        TestRunner.run finishes.
300
301        This function concludes a test run and writes out a test report.
302        """
303        if self.running:
304            msg = '\nSummary for test run %s: %s\n' % (
305                self.id, self.results.summary_str())
306            self._write_results_to_file()
307            self.log.info(msg.strip())
308            logger.kill_test_logger(self.log)
309            self.running = False
310
311    def _write_results_to_file(self):
312        """Writes test results to file(s) in a serializable format."""
313        # Old JSON format
314        path = os.path.join(self.log_path, 'test_run_summary.json')
315        with open(path, 'w') as f:
316            f.write(self.results.json_str())
317        # New YAML format
318        self.summary_writer.dump(self.results.summary_dict(),
319                                 records.TestSummaryEntryType.SUMMARY)
320
321    def dump_config(self):
322        """Writes the test config to a JSON file under self.log_path"""
323        config_path = os.path.join(self.log_path, 'test_configs.json')
324        with open(config_path, 'a') as f:
325            json.dump(dict(
326                itertools.chain(
327                    self.test_run_config.user_params.items(),
328                    self.test_run_config.controller_configs.items())),
329                      f,
330                      skipkeys=True,
331                      indent=4)
332
333    def write_test_campaign(self):
334        """Log test campaign file."""
335        path = os.path.join(self.log_path, 'test_campaign.log')
336        with open(path, 'w') as f:
337            for test_class, test_cases in self.run_list:
338                f.write('%s:\n%s' % (test_class, ',\n'.join(test_cases)))
339                f.write('\n\n')
340