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 logging 18 import os 19 20 from vts.runners.host import asserts 21 from vts.runners.host import const 22 from vts.runners.host import keys 23 from vts.runners.host import test_runner 24 from vts.testcases.template.binary_test import binary_test 25 from vts.utils.python.hal import hal_service_name_utils 26 from vts.utils.python.os import path_utils 27 28 29 class HalHidlReplayTest(binary_test.BinaryTest): 30 """Base class to run a HAL HIDL replay test on a target device. 31 32 Attributes: 33 _dut: AndroidDevice, the device under test as config 34 DEVICE_TMP_DIR: string, target device's tmp directory path. 35 DEVICE_VTS_SPEC_FILE_PATH: string, target device's directory for storing 36 HAL spec files. 37 """ 38 39 DEVICE_TMP_DIR = "/data/local/tmp" 40 DEVICE_VTS_SPEC_FILE_PATH = "/data/local/tmp/spec" 41 DEVICE_VTS_TRACE_FILE_PATH = "/data/local/tmp/vts_replay_trace" 42 43 def setUpClass(self): 44 """Prepares class and initializes a target device.""" 45 self._test_hal_services = set() 46 super(HalHidlReplayTest, self).setUpClass() 47 48 if self.isSkipAllTests(): 49 return 50 51 if self.coverage.enabled and self._test_hal_services is not None: 52 self.coverage.SetHalNames(self._test_hal_services) 53 54 # @Override 55 def CreateTestCases(self): 56 """Create a list of HalHidlReplayTestCase objects.""" 57 required_params = [ 58 keys.ConfigKeys.IKEY_HAL_HIDL_REPLAY_TEST_TRACE_PATHS, 59 keys.ConfigKeys.IKEY_ABI_BITNESS 60 ] 61 opt_params = [keys.ConfigKeys.IKEY_BINARY_TEST_DISABLE_FRAMEWORK] 62 self.getUserParams( 63 req_param_names=required_params, opt_param_names=opt_params) 64 self.abi_bitness = str(self.abi_bitness) 65 self.trace_paths = map(str, self.hal_hidl_replay_test_trace_paths) 66 67 self.replayer_binary_path = path_utils.JoinTargetPath( 68 self.DEVICE_TMP_DIR, self.abi_bitness, 69 "vts_hal_replayer%s" % self.abi_bitness) 70 self.custom_ld_library_path = path_utils.JoinTargetPath( 71 self.DEVICE_TMP_DIR, self.abi_bitness) 72 73 for trace_path in self.trace_paths: 74 trace_file_name = str(os.path.basename(trace_path)) 75 target_trace_path = path_utils.JoinTargetPath( 76 self.DEVICE_VTS_TRACE_FILE_PATH, trace_file_name) 77 self._dut.adb.push( 78 path_utils.JoinTargetPath(self.data_file_path, 79 "hal-hidl-trace", trace_path), 80 target_trace_path) 81 service_instance_combinations = self._GetServiceInstanceCombinations( 82 target_trace_path) 83 84 if service_instance_combinations: 85 for instance_combination in service_instance_combinations: 86 test_case = self.CreateReplayTestCase( 87 trace_file_name, target_trace_path) 88 service_name_list = [] 89 for instance in instance_combination: 90 test_case.args += " --hal_service_instance=" + instance 91 service_name_list.append( 92 instance[instance.find('/') + 1:]) 93 name_appendix = "({0})".format(",".join(service_name_list)) 94 test_case.name_appendix = name_appendix 95 self.testcases.append(test_case) 96 else: 97 test_case = self.CreateReplayTestCase(trace_file_name, 98 target_trace_path) 99 self.testcases.append(test_case) 100 101 def CreateReplayTestCase(self, trace_file_name, target_trace_path): 102 """Create a replay test case object. 103 104 Args: 105 trace_file_name: string, name of the trace file used in the test. 106 target_trace_path: string, full path of the trace file or the target device. 107 """ 108 test_case = super(HalHidlReplayTest, self).CreateTestCase( 109 self.replayer_binary_path, '') 110 test_case.envp += "LD_LIBRARY_PATH=%s:$LD_LIBRARY_PATH" % self.custom_ld_library_path 111 test_case.args += " --spec_dir_path=" + self.DEVICE_VTS_SPEC_FILE_PATH 112 test_case.args += " " + target_trace_path 113 test_case.test_name = "replay_test_" + trace_file_name 114 return test_case 115 116 def VerifyTestResult(self, test_case, command_results): 117 """Parse Gtest xml result output. 118 119 Args: 120 test_case: BinaryTestCase object, the test being run. This param 121 is not currently used in this method. 122 command_results: dict of lists, shell command result 123 """ 124 asserts.assertTrue(command_results, 'Empty command response.') 125 asserts.assertEqual( 126 len(command_results), 3, 'Abnormal command response.') 127 128 for stdout in command_results[const.STDOUT]: 129 if stdout and stdout.strip(): 130 for line in stdout.split('\n'): 131 logging.info(line) 132 133 if any(command_results[const.EXIT_CODE]): 134 # print stderr only when test fails. 135 for stderr in command_results[const.STDERR]: 136 if stderr and stderr.strip(): 137 for line in stderr.split('\n'): 138 logging.error(line) 139 asserts.fail( 140 'Test {} failed with the following results: {}'.format( 141 test_case, command_results)) 142 143 def tearDownClass(self): 144 """Performs clean-up tasks.""" 145 # Delete the pushed file. 146 if not self.isSkipAllTests(): 147 for trace_path in self.trace_paths: 148 trace_file_name = str(os.path.basename(trace_path)) 149 target_trace_path = path_utils.JoinTargetPath( 150 self.DEVICE_TMP_DIR, "vts_replay_trace", trace_file_name) 151 cmd_results = self.shell.Execute( 152 "rm -f %s" % target_trace_path) 153 if not cmd_results or any(cmd_results[const.EXIT_CODE]): 154 logging.warning("Failed to remove: %s", cmd_results) 155 156 super(HalHidlReplayTest, self).tearDownClass() 157 158 def setUp(self): 159 """Setup for code coverage for each test case.""" 160 super(HalHidlReplayTest, self).setUp() 161 if self.coverage.enabled and not self.coverage.global_coverage: 162 self.coverage.SetCoverageReportFilePrefix( 163 self._current_record.test_name.replace('/', '_')) 164 self.coverage.InitializeDeviceCoverage(self._dut) 165 166 def tearDown(self): 167 """Generate the coverage data for each test case.""" 168 if self.coverage.enabled and not self.coverage.global_coverage: 169 self.coverage.SetCoverageData(dut=self._dut, isGlobal=False) 170 171 super(HalHidlReplayTest, self).tearDown() 172 173 def _GetServiceInstanceCombinations(self, trace_path): 174 """Create all combinations of instances for all services recorded in 175 the trace file. 176 177 Args: 178 trace_path: string, full path of a given trace file 179 Returns: 180 A list of all service instance combinations. 181 """ 182 registered_services = [] 183 service_instances = {} 184 self.shell.Execute("chmod 755 %s" % self.replayer_binary_path) 185 results = self.shell.Execute( 186 "LD_LIBRARY_PATH=%s:$LD_LIBRARY_PATH %s --list_service %s" % 187 (self.custom_ld_library_path, self.replayer_binary_path, 188 trace_path)) 189 190 asserts.assertFalse( 191 results[const.EXIT_CODE][0], 192 'Failed to list test cases. EXIT_CODE: %s\n STDOUT: %s\n STDERR: %s\n' 193 % (results[const.EXIT_CODE][0], results[const.STDOUT][0], 194 results[const.STDERR][0])) 195 196 # parse the results to get the registered service list. 197 for line in results[const.STDOUT][0].split('\n'): 198 line = str(line) 199 if line.startswith('hal_service: '): 200 service = line[len('hal_service: '):] 201 registered_services.append(service) 202 203 for service in registered_services: 204 testable, service_names = hal_service_name_utils.GetHalServiceName( 205 self.shell, service, self.abi_bitness, 206 self.run_as_compliance_test) 207 if not testable: 208 self.skipAllTests("Hal: %s is not testable, " 209 "skip all tests." % service) 210 return [] 211 if service_names: 212 service_instances[service] = service_names 213 self._test_hal_services.add(service) 214 else: 215 self.skipAllTests("No service name found for: %s, " 216 "skip all tests." % service) 217 return [] 218 logging.info("registered service instances: %s", service_instances) 219 220 service_instance_combinations = \ 221 hal_service_name_utils.GetServiceInstancesCombinations( 222 registered_services, service_instances) 223 224 return service_instance_combinations 225 226 227 if __name__ == "__main__": 228 test_runner.main() 229
