Home | History | Annotate | Download | only in brillo_Gtests
      1 # Copyright 2015 The Chromium OS Authors. All rights reserved.
      2 # Use of this source code is governed by a BSD-style license that can be
      3 # found in the LICENSE file.
      4 
      5 from collections import namedtuple
      6 import fnmatch
      7 import logging
      8 import os
      9 
     10 import common
     11 from autotest_lib.client.common_lib import error
     12 from autotest_lib.client.common_lib.cros import dev_server
     13 from autotest_lib.server import afe_utils
     14 from autotest_lib.server import site_gtest_runner
     15 from autotest_lib.server import test
     16 
     17 
     18 NATIVE_TESTS_PATH = '/data/nativetest'
     19 WHITELIST_FILE = '/data/nativetest/tests.txt'
     20 ANDROID_NATIVE_TESTS_FILE_FMT = (
     21         '%(build_target)s-continuous_native_tests-%(build_id)s.zip')
     22 BRILLO_NATIVE_TESTS_FILE_FMT = '%(build_target)s-brillo-tests-%(build_id)s.zip'
     23 LIST_TEST_BINARIES_TEMPLATE = (
     24         'find %(path)s -type f -mindepth 2 -maxdepth 2 '
     25         '\( -perm -100 -o -perm -010 -o -perm -001 \)')
     26 NATIVE_ONLY_BOARDS = ['dragonboard']
     27 
     28 GtestSuite = namedtuple('GtestSuite', ['name', 'path', 'run_as_root', 'args'])
     29 
     30 class brillo_Gtests(test.test):
     31     """Run one or more native gTest Suites."""
     32     version = 1
     33 
     34 
     35     def initialize(self, host=None, gtest_suites=None, use_whitelist=False,
     36                    filter_tests=None, native_tests=None):
     37         if not afe_utils.host_in_lab(host):
     38             return
     39         # TODO(ralphnathan): Remove this once we can determine this in another
     40         # way (b/29185385).
     41         if host.get_board_name() in NATIVE_ONLY_BOARDS:
     42             self._install_nativetests(
     43                     host, BRILLO_NATIVE_TESTS_FILE_FMT, 'nativetests')
     44         else:
     45             self._install_nativetests(host,
     46                                       ANDROID_NATIVE_TESTS_FILE_FMT,
     47                                       'continuous_native_tests')
     48 
     49 
     50     def _install_nativetests(self, host, test_file_format, artifact):
     51         """Install the nativetests zip onto the DUT.
     52 
     53         Device images built by the Android Build System do not have the
     54         native gTests installed. This method requests the devserver to
     55         download the nativetests package into the lab, the test station
     56         will download/unzip the package, and finally install it onto the DUT.
     57 
     58         @param host: host object to install the nativetests onto.
     59         @param test_file_format: Format of the zip file containing the tests.
     60         @param artifact: Devserver artifact to stage.
     61         """
     62         info = host.host_info_store.get()
     63         ds = dev_server.AndroidBuildServer.resolve(info.build, host.hostname)
     64         ds.stage_artifacts(image=info.build, artifacts=[artifact])
     65         build_url = os.path.join(ds.url(), 'static', info.build)
     66         nativetests_file = (test_file_format %
     67                             host.get_build_info_from_build_url(build_url))
     68         tmp_dir = host.teststation.get_tmp_dir()
     69         host.download_file(build_url, nativetests_file, tmp_dir, unzip=True)
     70         host.adb_run('push %s %s' % (os.path.join(tmp_dir, 'DATA',
     71                                                   'nativetest'),
     72                                      NATIVE_TESTS_PATH))
     73 
     74 
     75     def _get_whitelisted_tests(self, whitelist_path):
     76         """Return the list of whitelisted tests.
     77 
     78         The whitelist is expected to be a three column CSV file containing:
     79         * the test name
     80         * "yes" or "no" whether the test should be run as root or not.
     81         * optional command line arguments to be passed to the test.
     82 
     83         Anything after a # on a line is considered to be a comment and  ignored.
     84 
     85         @param whitelist_path: Path to the whitelist.
     86 
     87         @return a map of test name to GtestSuite tuple.
     88         """
     89         suite_map = dict()
     90         for line in self.host.run_output(
     91                 'cat %s' % whitelist_path).splitlines():
     92             # Remove anything after the first # (comments).
     93             line = line.split('#')[0]
     94             if line.strip() == '':
     95                 continue
     96 
     97             parts = line.split(',')
     98             if len(parts) < 2:
     99                 logging.error('badly formatted line in %s: %s', whitelist_path,
    100                               line)
    101                 continue
    102 
    103             name = parts[0].strip()
    104             extra_args = parts[2].strip() if len(parts) > 2 else ''
    105             path = '' # Path will be updated if the test is present on the DUT.
    106             suite_map[name] = GtestSuite(name, path, parts[1].strip() == 'yes',
    107                                          extra_args)
    108         return suite_map
    109 
    110 
    111     def _find_all_gtestsuites(self, use_whitelist=False, filter_tests=None):
    112         """Find all the gTest Suites installed on the DUT.
    113 
    114         @param use_whitelist: Only whitelisted tests found on the system will
    115                               be used.
    116         @param filter_tests: Only tests that match these globs will be used.
    117         """
    118         list_cmd = LIST_TEST_BINARIES_TEMPLATE % {'path': NATIVE_TESTS_PATH}
    119         gtest_suites_path = self.host.run_output(list_cmd).splitlines()
    120         gtest_suites = [GtestSuite(os.path.basename(path), path, True, '')
    121                         for path in gtest_suites_path]
    122 
    123         if use_whitelist:
    124             try:
    125                 whitelisted = self._get_whitelisted_tests(WHITELIST_FILE)
    126                 suites_to_run = []
    127                 for suite in gtest_suites:
    128                     if whitelisted.get(suite.name):
    129                         whitelisted_suite = whitelisted.get(suite.name)
    130                         # Get the name and path from the suites on the DUT and
    131                         # get the other args from the whitelist map.
    132                         suites_to_run.append(GtestSuite(
    133                                 suite.name, suite.path,
    134                                 whitelisted_suite.run_as_root,
    135                                 whitelisted_suite.args))
    136                 gtest_suites = suites_to_run
    137                 if (len(suites_to_run) != len(whitelisted)):
    138                     whitelist_test_names = set(whitelisted.keys())
    139                     found_test_names = set([t.name for t in suites_to_run])
    140                     diff_tests = list(whitelist_test_names - found_test_names)
    141                     for t in diff_tests:
    142                         logging.warning('Could not find %s', t);
    143                     raise error.TestWarn(
    144                             'Not all whitelisted tests found on the DUT. '
    145                             'Expected %i tests but only found %i' %
    146                             (len(whitelisted), len(suites_to_run)))
    147             except error.GenericHostRunError:
    148                 logging.error('Failed to read whitelist %s', WHITELIST_FILE)
    149 
    150         if filter_tests:
    151             gtest_suites = [t for t in gtest_suites
    152                             if any(fnmatch.fnmatch(t.path, n)
    153                                    for n in filter_tests)]
    154             logging.info('Running tests:\n  %s',
    155                          '\n  '.join(t.path for t in gtest_suites))
    156 
    157         if not gtest_suites:
    158             raise error.TestWarn('No test executables found on the DUT')
    159         logging.debug('Test executables found:\n%s',
    160                       '\n'.join([str(t) for t in gtest_suites]))
    161         return gtest_suites
    162 
    163 
    164     def run_gtestsuite(self, gtestSuite):
    165         """Run a gTest Suite.
    166 
    167         @param gtestSuite: GtestSuite tuple.
    168 
    169         @return True if the all the tests in the gTest Suite pass. False
    170                 otherwise.
    171         """
    172         # Make sure the gTest Suite exists.
    173         result = self.host.run('test -e %s' % gtestSuite.path,
    174                                ignore_status=True)
    175         if not result.exit_status == 0:
    176             logging.error('Unable to find %s', gtestSuite.path)
    177             return False
    178 
    179         result = self.host.run('test -x %s' % gtestSuite.path,
    180                                ignore_status=True)
    181         if not result.exit_status == 0:
    182             self.host.run('chmod +x %s' % gtestSuite.path)
    183 
    184         logging.debug('Running: %s', gtestSuite)
    185         command = '%s %s' % (gtestSuite.path, gtestSuite.args)
    186         if not gtestSuite.run_as_root:
    187           command = 'su shell %s' % command
    188 
    189         # host.run() will print the stdout/stderr output in the debug logs
    190         # properly interleaved.
    191         result = self.host.run(command, ignore_status=True)
    192 
    193         parser = site_gtest_runner.gtest_parser()
    194         for line in result.stdout.splitlines():
    195             parser.ProcessLogLine(line)
    196         passed_tests = parser.PassedTests()
    197         if passed_tests:
    198             logging.debug('Passed Tests: %s', passed_tests)
    199         failed_tests = parser.FailedTests(include_fails=True,
    200                                           include_flaky=True)
    201         if failed_tests:
    202             logging.error('Failed Tests: %s', failed_tests)
    203             for test in failed_tests:
    204                 logging.error('Test %s failed:\n%s', test,
    205                               parser.FailureDescription(test))
    206             return False
    207         if result.exit_status != 0:
    208             logging.error('%s exited with exit code: %s',
    209                           gtestSuite, result.exit_status)
    210             return False
    211         return True
    212 
    213 
    214     def run_once(self, host=None, gtest_suites=None, use_whitelist=False,
    215                  filter_tests=None, native_tests=None):
    216         """Run gTest Suites on the DUT.
    217 
    218         @param host: host object representing the device under test.
    219         @param gtest_suites: List of gTest suites to run. Default is to run
    220                              every gTest suite on the host.
    221         @param use_whitelist: If gTestSuites is not passed in and use_whitelist
    222                               is true, only whitelisted tests found on the
    223                               system will be used.
    224         @param filter_tests: If gTestSuites is not passed in, search for tests
    225                              that match these globs to run instead.
    226         @param native_tests: Execute these specific tests.
    227 
    228         @raise TestFail: The test failed.
    229         """
    230         self.host = host
    231         if not gtest_suites and native_tests:
    232             gtest_suites = [GtestSuite('', t, True, '') for t in native_tests]
    233         if not gtest_suites:
    234             gtest_suites = self._find_all_gtestsuites(
    235                     use_whitelist=use_whitelist, filter_tests=filter_tests)
    236 
    237         failed_gtest_suites = []
    238         for gtestSuite in gtest_suites:
    239             if not self.run_gtestsuite(gtestSuite):
    240                 failed_gtest_suites.append(gtestSuite)
    241 
    242         if failed_gtest_suites:
    243             logging.error(
    244                     'The following gTest Suites failed: \n %s',
    245                     '\n'.join([str(t) for t in failed_gtest_suites]))
    246             raise error.TestFail(
    247                     'Not all gTest Suites completed successfully. '
    248                     '%s out of %s suites failed. '
    249                     'Failed Suites: %s'
    250                     % (len(failed_gtest_suites),
    251                        len(gtest_suites),
    252                        failed_gtest_suites))
    253