1 # 2 # Copyright (C) 2017 The Android Open Source Project 3 # 4 # Licensed under the Apache License, Version 2.0 (the "License"); 5 # you may not use this file except in compliance with the License. 6 # You may obtain a copy of the License at 7 # 8 # http://www.apache.org/licenses/LICENSE-2.0 9 # 10 # Unless required by applicable law or agreed to in writing, software 11 # distributed under the License is distributed on an "AS IS" BASIS, 12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 13 # See the License for the specific language governing permissions and 14 # limitations under the License. 15 # 16 17 import copy 18 import logging 19 20 from vts.runners.host import const 21 from vts.runners.host import keys 22 from vts.runners.host import test_runner 23 from vts.testcases.template.gtest_binary_test import gtest_binary_test 24 from vts.testcases.template.gtest_binary_test import gtest_test_case 25 from vts.utils.python.cpu import cpu_frequency_scaling 26 27 28 class HidlHalGTest(gtest_binary_test.GtestBinaryTest): 29 '''Base class to run a VTS target-side HIDL HAL test. 30 31 Attributes: 32 DEVICE_TEST_DIR: string, temp location for storing binary 33 TAG_PATH_SEPARATOR: string, separator used to separate tag and path 34 tags: all the tags that appeared in binary list 35 testcases: list of GtestTestCase objects, list of test cases to run 36 _cpu_freq: CpuFrequencyScalingController instance of a target device. 37 _dut: AndroidDevice, the device under test as config 38 ''' 39 40 def setUpClass(self): 41 """Checks precondition.""" 42 super(HidlHalGTest, self).setUpClass() 43 44 opt_params = [keys.ConfigKeys.IKEY_SKIP_IF_THERMAL_THROTTLING] 45 self.getUserParams(opt_param_names=opt_params) 46 47 self._skip_if_thermal_throttling = self.getUserParam( 48 keys.ConfigKeys.IKEY_SKIP_IF_THERMAL_THROTTLING, 49 default_value=False) 50 51 if not self._skip_all_testcases: 52 logging.info("Disable CPU frequency scaling") 53 self._cpu_freq = cpu_frequency_scaling.CpuFrequencyScalingController( 54 self._dut) 55 self._cpu_freq.DisableCpuScaling() 56 else: 57 self._cpu_freq = None 58 59 def CreateTestCases(self): 60 """Create testcases and conditionally enable passthrough mode. 61 62 Create testcases as defined in HidlHalGtest. If the passthrough option 63 is provided in the configuration or if coverage is enabled, enable 64 passthrough mode on the test environment. 65 """ 66 super(HidlHalGTest, self).CreateTestCases() 67 68 passthrough_opt = self.getUserParam( 69 keys.ConfigKeys.IKEY_PASSTHROUGH_MODE, default_value=False) 70 71 # Enable coverage if specified in the configuration or coverage enabled. 72 # TODO(ryanjcampbell@) support binderized mode 73 if passthrough_opt or self.coverage.enabled: 74 self._EnablePassthroughMode() 75 76 # @Override 77 def CreateTestCase(self, path, tag=''): 78 '''Create a list of GtestTestCase objects from a binary path. 79 80 Support testing against different service names by first executing a 81 dummpy test case which lists all the registered hal services. Then 82 query the service name(s) for each registered service with lshal. 83 For each service name, create a new test case each with the service 84 name as an additional argument. 85 86 Args: 87 path: string, absolute path of a gtest binary on device 88 tag: string, a tag that will be appended to the end of test name 89 90 Returns: 91 A list of GtestTestCase objects. 92 ''' 93 initial_test_cases = super(HidlHalGTest, self).CreateTestCase(path, 94 tag) 95 if not initial_test_cases: 96 return initial_test_cases 97 # first, run one test with --list_registered_services. 98 list_service_test_case = copy.copy(initial_test_cases[0]) 99 list_service_test_case.args += " --list_registered_services" 100 results = self.shell.Execute(list_service_test_case.GetRunCommand()) 101 if (results[const.EXIT_CODE][0]): 102 logging.error('Failed to list test cases from binary %s', 103 list_service_test_case.path) 104 # parse the results to get the registered service list. 105 registered_services = [] 106 for line in results[const.STDOUT][0].split('\n'): 107 line = str(line) 108 if line.startswith('hal_service: '): 109 service = line[len('hal_service: '):] 110 registered_services.append(service) 111 112 # If no service registered, return the initial test cases directly. 113 if not registered_services: 114 return initial_test_cases 115 116 # find the correponding service name(s) for each registered service and 117 # store the mapping in dict service_instances. 118 service_instances = {} 119 for service in registered_services: 120 cmd = '"lshal -i | grep -o %s/.* | sort -u"' % service 121 out = str(self._dut.adb.shell(cmd)).split() 122 service_names = map(lambda x: x[x.find('/') + 1:], out) 123 logging.info("registered service: %s with name: %s" % 124 (service, ' '.join(service_names))) 125 service_instances[service] = service_names 126 127 # get all the combination of service instances. 128 service_instance_combinations = self._GetServiceInstancesCombinations( 129 registered_services, service_instances) 130 131 new_test_cases = [] 132 for test_case in initial_test_cases: 133 for instance_combination in service_instance_combinations: 134 new_test_case = copy.copy(test_case) 135 for instance in instance_combination: 136 new_test_case.args += " --hal_service_instance=" + instance 137 new_test_case.tag = instance[instance.find( 138 '/'):] + new_test_case.tag 139 new_test_cases.append(new_test_case) 140 return new_test_cases 141 142 @classmethod 143 def _GetServiceInstancesCombinations(self, services, service_instances): 144 '''Create all combinations of instances for all services. 145 146 Args: 147 services: list, all services used in the test. e.g. [s1, s2] 148 service_instances: dictionary, mapping of each service and the 149 corresponding service name(s). 150 e.g. {"s1": ["n1"], "s2": ["n2", "n3"]} 151 152 Returns: 153 A list of all service instance combinations. 154 e.g. [[s1/n1, s2/n2], [s1/n1, s2/n3]] 155 ''' 156 157 service_instance_combinations = [] 158 if not services: 159 return service_instance_combinations 160 service = services.pop() 161 pre_instance_combs = self._GetServiceInstancesCombinations( 162 services, service_instances) 163 if service not in service_instances: 164 return pre_instance_combs 165 for name in service_instances[service]: 166 if not pre_instance_combs: 167 new_instance_comb = [service + '/' + name] 168 service_instance_combinations.append(new_instance_comb) 169 else: 170 for instance_comb in pre_instance_combs: 171 new_instance_comb = [service + '/' + name] 172 new_instance_comb.extend(instance_comb) 173 service_instance_combinations.append(new_instance_comb) 174 175 return service_instance_combinations 176 177 def _EnablePassthroughMode(self): 178 """Enable passthrough mode by setting getStub to true. 179 180 This funciton should be called after super class' setupClass method. 181 If called before setupClass, user_params will be changed in order to 182 trigger setupClass method to invoke this method again. 183 """ 184 if self.testcases: 185 for test_case in self.testcases: 186 envp = ' %s=true' % const.VTS_HAL_HIDL_GET_STUB 187 test_case.envp += envp 188 else: 189 logging.warn('No test cases are defined yet. Maybe setupClass ' 190 'has not been called. Changing user_params to ' 191 'enable passthrough mode option.') 192 self.user_params[keys.ConfigKeys.IKEY_PASSTHROUGH_MODE] = True 193 194 def setUp(self): 195 """Skips the test case if thermal throttling lasts for 30 seconds.""" 196 super(HidlHalGTest, self).setUp() 197 198 if (self._skip_if_thermal_throttling and 199 getattr(self, "_cpu_freq", None)): 200 self._cpu_freq.SkipIfThermalThrottling(retry_delay_secs=30) 201 202 def tearDown(self): 203 """Skips the test case if there is thermal throttling.""" 204 if (self._skip_if_thermal_throttling and 205 getattr(self, "_cpu_freq", None)): 206 self._cpu_freq.SkipIfThermalThrottling() 207 208 super(HidlHalGTest, self).tearDown() 209 210 def tearDownClass(self): 211 """Turns off CPU frequency scaling.""" 212 if (not self._skip_all_testcases and getattr(self, "_cpu_freq", None)): 213 logging.info("Enable CPU frequency scaling") 214 self._cpu_freq.EnableCpuScaling() 215 216 super(HidlHalGTest, self).tearDownClass() 217 218 219 if __name__ == "__main__": 220 test_runner.main() 221
