testcases/template/gtest_binary_test/gtest_binary_test.py

#
# 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 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()
                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:
        # Avoid batch mode as it creates overly large filters
        return test_cases

        # Gtest batch mode
        # test_names = map(lambda test: test.full_name, test_cases)
        #test_names = {}

        #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 self.batch_mode:
            for test_case in self.testcases:
                logging.info('Running %s test cases in batch.',
                             len(test_case.full_name.split(':')))
                self.RunTestCase(test_case)

                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()