# # Copyright (C) 2016 The Android Open Source Project # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # import logging import os import re import tempfile import xml.etree.ElementTree from vts.runners.host import asserts from vts.runners.host import const from vts.runners.host import keys from vts.runners.host import test_runner from vts.testcases.template.binary_test import binary_test from vts.testcases.template.binary_test import binary_test_case from vts.testcases.template.gtest_binary_test import gtest_test_case _GTEST_RESULT_ATTRIBUTE_ALLOW_LIST = ('properties',) class GtestBinaryTest(binary_test.BinaryTest): '''Base class to run gtests binary on target. Attributes: DEVICE_TEST_DIR: string, temp location for storing binary TAG_PATH_SEPARATOR: string, separator used to separate tag and path shell: ShellMirrorObject, shell mirror tags: all the tags that appeared in binary list testcases: list of GtestTestCase objects, list of test cases to run _dut: AndroidDevice, the device under test as config _gtest_results: list of GtestResult objects, used during batch mode for result storage and parsing ''' # @Override def setUpClass(self): '''Prepare class, push binaries, set permission, create test cases.''' self.collect_tests_only = self.getUserParam( keys.ConfigKeys.IKEY_COLLECT_TESTS_ONLY, default_value=False) self.batch_mode = self.getUserParam( keys.ConfigKeys.IKEY_GTEST_BATCH_MODE, default_value=False) if self.batch_mode: if self.collect_tests_only: self.batch_mode = False logging.debug("Disable batch mode when collecting tests.") else: self._gtest_results = [] super(GtestBinaryTest, self).setUpClass() # @Override def CreateTestCase(self, path, tag=''): '''Create a list of GtestTestCase objects from a binary path. Args: path: string, absolute path of a gtest binary on device tag: string, a tag that will be appended to the end of test name Returns: A list of GtestTestCase objects on success; an empty list otherwise. In non-batch mode, each object respresents a test case in the gtest binary located at the provided path. Usually there are more than one object returned. In batch mode, each object represents a gtest binary located at the provided path; the returned list will always be a one object list in batch mode. Test case names are stored in full_name property in the object, delimited by ':' according to gtest documentation, after being filtered and processed according to host configuration. ''' working_directory = self.working_directory[ tag] if tag in self.working_directory else None envp = self.envp[tag] if tag in self.envp else '' args = self.args[tag] if tag in self.args else '' ld_library_path = self.ld_library_path[ tag] if tag in self.ld_library_path else None profiling_library_path = self.profiling_library_path[ tag] if tag in self.profiling_library_path else None gtest_list_args = args + " --gtest_list_tests" list_test_case = binary_test_case.BinaryTestCase( 'gtest_list_tests', path, path, tag, self.PutTag, working_directory, ld_library_path, profiling_library_path, envp=envp, args=gtest_list_args) cmd = ['chmod 755 %s' % path, list_test_case.GetRunCommand()] cmd_results = self.shell.Execute(cmd) test_cases = [] asserts.assertFalse(any(cmd_results[const.EXIT_CODE]), 'Failed to list test cases from %s. Command: %s, Result: %s.' % (path, cmd, cmd_results)) test_suite = '' for line in cmd_results[const.STDOUT][1].split('\n'): line = str(line) if not len(line.strip()): continue elif line.startswith(' '): # Test case name test_name = line.split('#')[0].strip() # Skip any test that doesn't instantiate the parameterized gtest if re.match('UninstantiatedParamaterizedTestSuite<(.*)>', test_name): continue test_case = gtest_test_case.GtestTestCase( test_suite, test_name, path, tag, self.PutTag, working_directory, ld_library_path, profiling_library_path, envp=envp, args=args) logging.debug('Gtest test case: %s' % test_case) test_cases.append(test_case) else: # Test suite name test_suite = line.strip() if test_suite.endswith('.'): test_suite = test_suite[:-1] if not self.batch_mode: return test_cases # Gtest batch mode test_names = map(lambda test: test.full_name, test_cases) gtest_batch = gtest_test_case.GtestTestCase( path, '', path, tag, self.PutTag, working_directory, ld_library_path, profiling_library_path, envp=envp) gtest_batch.full_name = ':'.join(test_names) return [gtest_batch] # @Override def VerifyTestResult(self, test_case, command_results): '''Parse Gtest xml result output. Sample Args: test_case: GtestTestCase object, the test being run. This param is not currently used in this method. command_results: dict of lists, shell command result ''' asserts.assertTrue(command_results, 'Empty command response.') asserts.assertEqual( len(command_results), 3, 'Abnormal command response.') for item in command_results.values(): asserts.assertEqual( len(item), 2, 'Abnormal command result length: %s' % command_results) for stderr in command_results[const.STDERR]: if stderr and stderr.strip(): for line in stderr.split('\n'): logging.error(line) xml_str = command_results[const.STDOUT][1] if self.batch_mode: self._ParseBatchResults(test_case, xml_str) return asserts.assertFalse( command_results[const.EXIT_CODE][1], 'Failed to show Gtest XML output: %s' % command_results) root = self._ParseResultXmlString(xml_str) asserts.assertEqual(root.get('tests'), '1', 'No tests available') success = True if root.get('errors') != '0' or root.get('failures') != '0': messages = [x.get('message') for x in root.findall('.//failure')] success = False for stdout in command_results[const.STDOUT]: if stdout and stdout.strip(): for line in stdout.split('\n'): if success: logging.debug(line) else: logging.error(line) if not success: asserts.fail('\n'.join([x for x in messages if x])) asserts.skipIf(root.get('disabled') == '1', 'Gtest test case disabled') def _ParseResultXmlString(self, xml_str): """Parses the xml result string into elements. Args: xml_str: string, result xml text content. Returns: xml.etree.ElementTree, parsed xml content. Raises: assertion failure if xml format is not expected. """ asserts.assertTrue(xml_str is not None, 'Test command result not received.') xml_str = xml_str.strip() asserts.assertTrue(xml_str, 'Test command result is empty.') try: return xml.etree.ElementTree.fromstring(xml_str) except: asserts.fail('Result xml content is corrupted.') def _ParseBatchResults(self, test_case_original, xml_str): '''Parse batch mode gtest results Args: test_case_original: GtestTestCase object, original batch test case object xml_str: string, result xml output content ''' root = self._ParseResultXmlString(xml_str) for test_suite in root: logging.debug('Test tag: %s, attribute: %s', test_suite.tag, test_suite.attrib) for test_case in test_suite: result = gtest_test_case.GtestTestCase( test_suite.get('name'), test_case.get('name'), '', test_case_original.tag, self.PutTag, name_appendix=test_case_original.name_appendix) failure_message = None for sub in test_case: if sub.tag == 'failure': failure_message = sub.get('message') test_case_filtered = filter( lambda sub: sub.tag not in _GTEST_RESULT_ATTRIBUTE_ALLOW_LIST, test_case) if len(test_case_filtered) and not failure_message: failure_message = 'Error: %s\n' % test_case.attrib for sub in test_case_filtered: failure_message += '%s: %s\n' % (sub.tag, sub.attrib) result.failure_message = failure_message self._gtest_results.append(result) def _VerifyBatchResult(self, gtest_result): '''Check a gtest test case result in batch mode Args: gtest_result: GtestTestCase object, representing gtest result ''' asserts.assertFalse(gtest_result.failure_message, gtest_result.failure_message) # @Override def generateAllTests(self): '''Runs all binary tests. If the test cases should run in batch mode, this method executes the gtest commands without adding test records, and then parses the XML reports to records. If the test cases should run in batch mode but be skipped (e.g., HAL is not implemented), this method applies the filters in base_test, skips the batch test cases, and adds one record for each of them. ''' if self.batch_mode and not self.isSkipAllTests(): # TODO(b/126412742): Convert filters to --gtest_filter. for test_case in self.testcases: logging.info('Running %s test cases in batch.', len(test_case.full_name.split(':'))) gtest_filter_flag=('--gtest_filter={test}').format(test = test_case) dst = '/data/local/tmp/filter_file' temp = tempfile.NamedTemporaryFile() try: temp.write(gtest_filter_flag) self._dut.adb.push('{src} {dst}'.format(src=temp.name, dst=dst)) finally: temp.close() test_case.filter_file = dst self.RunTestCase(test_case) self.shell.Execute('rm %s' % dst) self.runGeneratedTests( test_func=self._VerifyBatchResult, settings=self._gtest_results, name_func=str) self._gtest_results = [] return self.runGeneratedTests( test_func=self.RunTestCase, settings=self.testcases, name_func=str) if __name__ == "__main__": test_runner.main()