blob: dfbe100a2bdac4339c17a4bf75f3cdf3d9ac0a26 [file] [log] [blame]
#
# 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_WHITE_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
<testsuites tests="1" failures="1" disabled="0" errors="0"
timestamp="2017-05-24T18:32:10" time="0.012" name="AllTests">
<testsuite name="ConsumerIrHidlTest"
tests="1" failures="1" disabled="0" errors="0" time="0.01">
<testcase name="TransmitTest" status="run" time="0.01"
classname="ConsumerIrHidlTest">
<failure message="hardware/interfaces..." type="">
<![CDATA[hardware/interfaces...]]>
</failure>
</testcase>
</testsuite>
</testsuites>
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_WHITE_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()