1 # 2 # Copyright (C) 2016 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 import itertools 20 import operator 21 import os 22 23 from vts.runners.host import const 24 from vts.utils.python.common import cmd_utils 25 from vts.utils.python.os import path_utils 26 27 from vts.testcases.kernel.ltp.shell_environment import shell_environment 28 from vts.testcases.kernel.ltp import ltp_enums 29 from vts.testcases.kernel.ltp import ltp_configs 30 from vts.testcases.kernel.ltp import requirements 31 32 33 class EnvironmentRequirementChecker(object): 34 """LTP testcase environment checker. 35 36 This class contains a dictionary for some known environment 37 requirements for a set of test cases and several environment 38 check functions to be mapped with. All check functions' results 39 are cached in a dictionary for multiple use. 40 41 Attributes: 42 _REQUIREMENT_DEFINITIONS: dictionary {string, obj}, a map between 43 requirement name and the actual definition class object 44 _result_cache: dictionary {requirement_check_method_name: 45 (bool, string)}, a map between check method name and cached result 46 tuples (boolean, note) 47 _executable_available: dict {string, bool}, a map between executable 48 path and its existance on target 49 _shell_env: ShellEnvironment object, which checks and sets 50 shell environments given a shell mirror 51 shell: shell mirror object, can be used to execute shell 52 commands on target side through runner 53 ltp_bin_host_path: string, host path of ltp binary 54 """ 55 56 def __init__(self, shell): 57 self.shell = shell 58 self._result_cache = {} 59 self._executable_available = {} 60 self._shell_env = shell_environment.ShellEnvironment(self.shell) 61 self._REQUIREMENT_DEFINITIONS = requirements.GetRequrementDefinitions() 62 63 @property 64 def shell(self): 65 """Get the runner's shell mirror object to execute commands""" 66 return self._shell 67 68 @shell.setter 69 def shell(self, shell): 70 """Set the runner's shell mirror object to execute commands""" 71 self._shell = shell 72 73 def Cleanup(self): 74 """Run all cleanup jobs at the end of tests""" 75 self._shell_env.Cleanup() 76 77 def GetRequirements(self, test_case): 78 """Get a list of requirements for a fiven test case 79 80 Args: 81 test_case: TestCase object, the test case to query 82 """ 83 result = copy.copy(ltp_configs.REQUIREMENT_FOR_ALL) 84 85 result.extend(rule 86 for rule, tests in 87 ltp_configs.REQUIREMENTS_TO_TESTCASE.iteritems() 88 if test_case.fullname in tests) 89 90 result.extend(rule 91 for rule, tests in 92 ltp_configs.REQUIREMENT_TO_TESTSUITE.iteritems() 93 if test_case.testsuite in tests) 94 95 return list(set(result)) 96 97 def Check(self, test_case): 98 """Check whether a given test case's requirement has been satisfied. 99 Skip the test if not. 100 101 If check failed, this method returns False and the reason is set 102 to test_case.note. 103 104 Args: 105 test_case: TestCase object, a given test case to check 106 107 Returns: 108 True if check pass; False otherwise 109 """ 110 if (test_case.requirement_state == 111 ltp_enums.RequirementState.UNSATISFIED or 112 not self.TestBinaryExists(test_case)): 113 return False 114 115 for requirement in self.GetRequirements(test_case): 116 if requirement not in self._result_cache: 117 definitions = self._REQUIREMENT_DEFINITIONS[requirement] 118 self._result_cache[ 119 requirement] = self._shell_env.ExecuteDefinitions( 120 definitions) 121 122 result, note = self._result_cache[requirement] 123 logging.info("Result for %s's requirement %s is %s", test_case, 124 requirement, result) 125 if result is False: 126 test_case.requirement_state = ltp_enums.RequirementState.UNSATISFIED 127 test_case.note = note 128 return False 129 130 test_case.requirement_state = ltp_enums.RequirementState.SATISFIED 131 return True 132 133 def CheckAllTestCaseExecutables(self, test_cases): 134 """Run a batch job to check executable exists and set permissions. 135 136 The result will be stored in self._executable_available for use in 137 TestBinaryExists method. 138 139 Args: 140 test_case: list of TestCase objects. 141 """ 142 executables_generators = ( 143 test_case.GetRequiredExecutablePaths(self.ltp_bin_host_path) 144 for test_case in test_cases) 145 executables = list( 146 set(itertools.chain.from_iterable(executables_generators))) 147 148 # Set all executables executable permission using chmod. 149 logging.info("Setting permissions on device") 150 permission_command = "chmod 775 %s" % path_utils.JoinTargetPath( 151 ltp_configs.LTPBINPATH, '*') 152 permission_result = self.shell.Execute(permission_command) 153 if permission_result[const.EXIT_CODE][0]: 154 logging.error("Permission command '%s' failed.", 155 permission_command) 156 157 # Check existence of all executables used in test definition. 158 # Some executables needed by test cases but not listed in test 159 # definition will not be checked here 160 logging.info("Checking binary existence on host") 161 162 executable_exists_results = map(os.path.exists, executables) 163 164 self._executable_available = dict( 165 zip(executables, executable_exists_results)) 166 167 not_exists = [ 168 exe for exe, exists in self._executable_available.iteritems() 169 if not exists 170 ] 171 if not_exists: 172 logging.info("The following binaries does not exist: %s", 173 not_exists) 174 175 logging.info("Finished checking binary existence on host.") 176 177 # Check whether all the internal binaries in path needed exist 178 bin_path_exist_commands = [ 179 "which %s" % bin for bin in ltp_configs.INTERNAL_BINS 180 ] 181 bin_path_results = map( 182 operator.not_, 183 self.shell.Execute(bin_path_exist_commands)[const.EXIT_CODE]) 184 185 bin_path_results = map( 186 operator.not_, 187 self.shell.Execute(bin_path_exist_commands)[const.EXIT_CODE]) 188 189 self._executable_available.update( 190 dict(zip(ltp_configs.INTERNAL_BINS, bin_path_results))) 191 192 def TestBinaryExists(self, test_case): 193 """Check whether the given test case's binary exists. 194 195 Args: 196 test_case: TestCase, the object representing the test case 197 198 Return: 199 True if exists, False otherwise 200 """ 201 if test_case.requirement_state == ltp_enums.RequirementState.UNSATISFIED: 202 logging.warn("[Checker] Attempting to run test case that has " 203 "already been checked and requirement not satisfied." 204 "%s" % test_case) 205 return False 206 207 executables = test_case.GetRequiredExecutablePaths( 208 self.ltp_bin_host_path) 209 results = [ 210 self._executable_available[executable] 211 for executable in executables 212 ] 213 214 if not all(results): 215 test_case.requirement_state = ltp_enums.RequirementState.UNSATISFIED 216 test_case.note = "Some executables not exist: {}".format( 217 zip(executables, results)) 218 logging.error("[Checker] Binary existance check failed for {}. " 219 "Reason: {}".format(test_case, test_case.note)) 220 return False 221 else: 222 return True 223
