#
# Copyright (C) 2017 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 copy
import logging
from vts.runners.host import const
from vts.runners.host import keys
from vts.runners.host import test_runner
from vts.testcases.template.gtest_binary_test import gtest_binary_test
from vts.testcases.template.gtest_binary_test import gtest_test_case
from vts.utils.python.cpu import cpu_frequency_scaling
from vts.utils.python.hal import hal_service_name_utils
from vts.utils.python.precondition import precondition_utils
# The pattern indicating a full hal test name including the service name info.
# e.g. TM.TC(default)_32bit
_HAL_TEST_NAME_PATTERN = ".*\(.*\).*"
class HidlHalGTest(gtest_binary_test.GtestBinaryTest):
"""Base class to run a VTS target-side HIDL HAL test.
Attributes:
DEVICE_TEST_DIR: string, temp location for storing binary
TAG_PATH_SEPARATOR: string, separator used to separate tag and path
tags: all the tags that appeared in binary list
testcases: list of GtestTestCase objects, list of test cases to run
_cpu_freq: CpuFrequencyScalingController instance of a target device.
_dut: AndroidDevice, the device under test as config
_initial_test_case_cnt: Number of initial test cases.
_target_hals: List of String, the targeting hal service of the test.
e.g (["android.hardware.foo@1.0::IFoo"])
"""
def setUpClass(self):
"""Checks precondition."""
self._initial_test_case_cnt = 0
super(HidlHalGTest, self).setUpClass()
if not hasattr(self, "_target_hals"):
self._target_hals = []
opt_params = [keys.ConfigKeys.IKEY_SKIP_IF_THERMAL_THROTTLING,
keys.ConfigKeys.IKEY_DISABLE_CPU_FREQUENCY_SCALING]
self.getUserParams(opt_param_names=opt_params)
self._skip_if_thermal_throttling = self.getUserParam(
keys.ConfigKeys.IKEY_SKIP_IF_THERMAL_THROTTLING,
default_value=False)
self._disable_cpu_frequency_scaling = self.getUserParam(
keys.ConfigKeys.IKEY_DISABLE_CPU_FREQUENCY_SCALING,
default_value=True)
if not self.isSkipAllTests():
self._cpu_freq = cpu_frequency_scaling.CpuFrequencyScalingController(
self._dut)
if self._disable_cpu_frequency_scaling:
logging.debug("Disabling CPU frequency scaling")
self._cpu_freq.DisableCpuScaling()
else:
self._cpu_freq = None
if not self.isSkipAllTests():
ret = precondition_utils.CanRunHidlHalTest(
self, self._dut, self.shell, self.run_as_compliance_test)
if not ret:
self.skipAllTests("HIDL HAL precondition check failed.")
# Extend timeout if there are multiple service instance combinations.
if (not self.isSkipAllTests() and self._initial_test_case_cnt and
len(self.testcases) > self._initial_test_case_cnt):
self.resetTimeout(self.timeout * len(self.testcases) /
float(self._initial_test_case_cnt))
if self.sancov.enabled and self._target_hals:
self.sancov.InitializeDeviceCoverage(self._dut,
self._target_hals)
if self.coverage.enabled and self._target_hals:
self.coverage.SetHalNames(self._target_hals)
self.coverage.SetCoverageReportFilePrefix(self.test_module_name + self.abi_bitness)
self.coverage.InitializeDeviceCoverage(self._dut)
def CreateTestCases(self):
"""Create testcases and conditionally enable passthrough mode.
Create testcases as defined in HidlHalGtest. If the passthrough option
is provided in the configuration or if coverage is enabled, enable
passthrough mode on the test environment.
"""
super(HidlHalGTest, self).CreateTestCases()
passthrough_opt = self.getUserParam(
keys.ConfigKeys.IKEY_PASSTHROUGH_MODE, default_value=False)
# Enable coverage if specified in the configuration.
if passthrough_opt:
self._EnablePassthroughMode()
# @Override
def CreateTestCase(self, path, tag=''):
"""Create a list of GtestTestCase objects from a binary path.
Support testing against different service names by first executing a
dummpy test case which lists all the registered hal services. Then
query the service name(s) for each registered service with lshal.
For each service name, create a new test case each with the service
name as an additional argument.
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.
"""
initial_test_cases = super(HidlHalGTest, self).CreateTestCase(path,
tag)
self._initial_test_case_cnt += len(initial_test_cases)
if not initial_test_cases:
return initial_test_cases
# first, run one test with --list_registered_services.
list_service_test_case = copy.copy(initial_test_cases[0])
list_service_test_case.args = " --list_registered_services"
results = self.shell.Execute(
list_service_test_case.GetRunCommand(raw_command=True))
if (results[const.EXIT_CODE][0]):
logging.error("Failed to list test cases from binary %s",
list_service_test_case.path)
# parse the results to get the registered service list.
registered_services = []
comb_mode = hal_service_name_utils.CombMode.FULL_PERMUTATION
# TODO: consider to use a standard data format (e.g. json) instead of
# parsing the print output.
for line in results[const.STDOUT][0].split("\n"):
line = str(line)
if line.startswith("hal_service: "):
service = line[len("hal_service: "):]
registered_services.append(service)
if line.startswith("service_comb_mode: "):
comb_mode = int(line[len("service_comb_mode: "):])
# If no service registered, return the initial test cases directly.
if not registered_services:
logging.error("No hal service registered.")
return initial_test_cases
self._target_hals = copy.copy(registered_services)
# find the correponding service name(s) for each registered service and
# store the mapping in dict service_instances.
service_instances = {}
for service in registered_services:
testable, service_names = hal_service_name_utils.GetHalServiceName(
self.shell, service, self.abi_bitness,
self.run_as_compliance_test)
if not testable:
self.skipAllTests("Hal: %s is not testable, "
"skip all tests." % service)
return initial_test_cases
if not service_names:
self.skipAllTests("No service name found for: %s, skip all tests." % service)
# If any of the test services are not available, return the
# initial test cases directly.
return initial_test_cases
else:
service_instances[service] = service_names
logging.debug("registered service instances: %s", service_instances)
logging.debug("service comb mode: %d", comb_mode)
# If request NO_COMBINATION mode, return the initial test cases directly.
if comb_mode == hal_service_name_utils.CombMode.NO_COMBINATION:
return initial_test_cases
# get all the combination of service instances.
service_instance_combinations = hal_service_name_utils.GetServiceInstancesCombinations(
registered_services, service_instances, comb_mode);
new_test_cases = []
appendix_list = []
for instance_combination in service_instance_combinations:
for test_case in initial_test_cases:
new_test_case = copy.copy(test_case)
service_name_list = []
for instance in instance_combination:
new_test_case.args += " --hal_service_instance=" + instance
service_name_list.append(instance[instance.find('/')+1:])
name_appendix = "({0})".format(",".join(service_name_list))
new_test_case.name_appendix = name_appendix
new_test_cases.append(new_test_case)
appendix_list.append(name_appendix)
self.test_filter.ExpandAppendix(appendix_list, _HAL_TEST_NAME_PATTERN)
return new_test_cases
def _EnablePassthroughMode(self):
"""Enable passthrough mode by setting getStub to true.
This funciton should be called after super class' setupClass method.
If called before setupClass, user_params will be changed in order to
trigger setupClass method to invoke this method again.
"""
if self.testcases:
for test_case in self.testcases:
envp = ' %s=true' % const.VTS_HAL_HIDL_GET_STUB
test_case.envp += envp
else:
logging.warn('No test cases are defined yet. Maybe setupClass '
'has not been called. Changing user_params to '
'enable passthrough mode option.')
self.user_params[keys.ConfigKeys.IKEY_PASSTHROUGH_MODE] = True
def setUp(self):
"""Skips the test case if thermal throttling lasts for 30 seconds."""
super(HidlHalGTest, self).setUp()
if (self._skip_if_thermal_throttling and
getattr(self, "_cpu_freq", None)):
self._cpu_freq.SkipIfThermalThrottling(retry_delay_secs=30)
def tearDown(self):
"""Skips the test case if there is thermal throttling."""
if (self._skip_if_thermal_throttling and
getattr(self, "_cpu_freq", None)):
self._cpu_freq.SkipIfThermalThrottling()
super(HidlHalGTest, self).tearDown()
def tearDownClass(self):
"""Turns off CPU frequency scaling."""
if (not self.isSkipAllTests() and getattr(self, "_cpu_freq", None)
and self._disable_cpu_frequency_scaling):
logging.debug("Enabling CPU frequency scaling")
self._cpu_freq.EnableCpuScaling()
if self.sancov.enabled and self._target_hals:
self.sancov.FlushDeviceCoverage(self._dut, self._target_hals)
self.sancov.ProcessDeviceCoverage(self._dut,
self._target_hals)
self.sancov.Upload()
super(HidlHalGTest, self).tearDownClass()
if __name__ == "__main__":
test_runner.main()