1# Copyright 2022 The Chromium OS Authors. All rights reserved. 2# Use of this source code is governed by a BSD-style license that can be 3# found in the LICENSE file. 4 5import os 6import re 7import logging 8import json 9 10import common 11from autotest_lib.server import autotest, test 12from autotest_lib.client.common_lib import error 13 14from google.protobuf.text_format import Parse 15 16# run protoc --proto_path=./ pass_criteria.proto --python_out ./ 17# with caution for version/upgrade compatibility 18from . import pass_criteria_pb2 19 20 21class test_with_pass_criteria(test.test): 22 """ 23 test_with_pass_criteria extends the base test implementation to allow for 24 test result comparison between the performance keyvalues output from a 25 target test, and the input pass_criteria dictionary. 26 27 It can be used to create a domain specific test wrapper such as 28 power_QualTestWrapper. 29 """ 30 31 def initialize(self, test_to_wrap): 32 """ 33 initialize implements the initialize call in test.test, is called before 34 execution of the test 35 """ 36 self._test_prefix = [] 37 self._perf_dict = {} 38 self._attr_dict = {} 39 self._results_path = self.job._server_offload_dir_path() 40 self._wrapper_results = self._results_path + self.tagged_testname + '/' 41 logging.debug('...results going to %s', str(self._results_path)) 42 self._wrapped_test_results_keyval_path = (self._wrapper_results + 43 test_to_wrap + 44 '/results/keyval') 45 self._wrapped_test_keyval_path = self._wrapper_results + test_to_wrap + '/keyval' 46 self._wrapper_test_keyval_path = self._wrapper_results + 'keyval' 47 48 def _check_wrapped_test_passed(self, test_name): 49 results_path = self._wrapper_results + test_name + "" 50 51 def _load_proto_to_pass_criteria(self): 52 """ 53 _load_proto_to_pass_criteria optionally inputs a textproto file 54 or a ':' separated string which represents the pass criteria for 55 the test, and adds it to the pass criteria dictionary. 56 """ 57 for textproto in self._textproto_path.split(':'): 58 if not os.path.exists(textproto): 59 raise error.TestFail('provided textproto path ' + textproto + 60 ' does not exist') 61 62 logging.info('loading criteria from textproto %s', textproto) 63 with open(textproto) as textpb: 64 textproto_criteria = Parse(textpb.read(), 65 pass_criteria_pb2.PassCriteria()) 66 for criteria in textproto_criteria.criteria: 67 lower_bound = criteria.lower_bound.bound if ( 68 criteria.HasField('lower_bound')) else None 69 upper_bound = criteria.upper_bound.bound if ( 70 criteria.HasField('upper_bound')) else None 71 if criteria.test_name != self._test_to_wrap and criteria.test_name != '': 72 logging.info('criteria %s does not apply', 73 criteria.name_regex) 74 continue 75 try: 76 self._pass_criteria[criteria.name_regex] = (lower_bound, 77 upper_bound) 78 logging.info('adding criteria %s', criteria.name_regex) 79 except: 80 raise error.TestFail('invalid pass criteria provided') 81 82 def add_prefix_test(self, test='', prefix_args_dict=None): 83 """ 84 add_prefix_test takes a test_name and args_dict for that test. 85 This function allows a user creating a domain specific test wrapper 86 to add any prefix tests that must run prior to execution of the 87 target test. 88 89 @param test: the name of the test to add as a prefix test operation 90 @param prefix_args_dict: the dictionary of args to pass to the test 91 when it is run 92 """ 93 if prefix_args_dict is None: 94 prefix_args_dict = {} 95 self._test_prefix.append((test, prefix_args_dict)) 96 97 def _print_bounds_error(self, criteria, failed_criteria, value): 98 """ 99 _print_bounds_error will indicate missing pass criteria, printing the 100 error string with failing criteria and target range 101 102 @param criteria: the name of the pass criteria to log a failure on 103 @param failed_criteria: the name of the criteria that regex matched 104 @param value: the actual value of the failing pass criteria 105 """ 106 logging.info('criteria %s: %s out of range %s', failed_criteria, 107 str(value), str(self._pass_criteria[criteria])) 108 109 def _parse_wrapped_results_keyvals(self): 110 """ 111 _parse_wrapped_results_keyvals first loads all of the performance and 112 and attribute keyvals from the wrapped test, and then copies all of 113 the test_attribute keyvals from that wrapped test into the wrapper. 114 Without these keyvals being copied over, none of the metadata from 115 the client job are captured in the job summary. 116 117 @raises: error.TestFail: If any of the respective keyvals are missing 118 """ 119 if os.path.exists(self._wrapped_test_results_keyval_path): 120 with open(self._wrapped_test_results_keyval_path 121 ) as results_keyval_file: 122 keyval_result = results_keyval_file.readline() 123 while keyval_result: 124 regmatch = re.search(r'(.*){(.*)}=(.*)', keyval_result) 125 if regmatch is None: 126 break 127 key = regmatch.group(1) 128 which_dict = regmatch.group(2) 129 value = regmatch.group(3) 130 if which_dict != 'perf': 131 continue 132 133 self._perf_dict[key] = value 134 keyval_result = results_keyval_file.readline() 135 136 with open(self._wrapped_test_keyval_path, 137 'r') as wrapped_test_keyval_file, open( 138 self._wrapper_test_keyval_path, 139 'a') as test_keyval_file: 140 for keyval in wrapped_test_keyval_file: 141 test_keyval_file.write(keyval) 142 143 def _find_matching_keyvals(self): 144 for c in self._pass_criteria: 145 self._criteria_to_keyvals[c] = [] 146 for key in self._perf_dict.keys(): 147 if re.fullmatch(c, key): 148 logging.info('adding %s as matched key', key) 149 self._criteria_to_keyvals[c].append(key) 150 151 def _verify_criteria(self): 152 failing_criteria = 0 153 for criteria in self._pass_criteria: 154 logging.info('Checking %s now', criteria) 155 if type(criteria) is not str: 156 criteria = criteria.decode('utf-8') 157 range_spec = self._pass_criteria[criteria] 158 159 for perf_val in self._criteria_to_keyvals[criteria]: 160 logging.info('Checking: %s against %s', str(criteria), 161 perf_val) 162 actual_value = self._perf_dict[perf_val] 163 logging.info('%s value is %s, spec is %s', perf_val, 164 float(actual_value), range_spec) 165 166 # range_spec is passed into the dictionary as a tuple of upper and lower 167 lower_bound, upper_bound = range_spec 168 169 if lower_bound is not None and not (float(actual_value) >= 170 float(lower_bound)): 171 failing_criteria = failing_criteria + 1 172 self._print_bounds_error(criteria, perf_val, actual_value) 173 174 if upper_bound is not None and not (float(actual_value) < 175 float(upper_bound)): 176 failing_criteria = failing_criteria + 1 177 self._print_bounds_error(criteria, perf_val, actual_value) 178 179 if failing_criteria > 0: 180 raise error.TestFail( 181 str(failing_criteria) + 182 ' criteria failed, see log for detail') 183 184 def run_once(self, 185 host=None, 186 test_to_wrap=None, 187 pdash_note='', 188 wrap_args={}, 189 pass_criteria={}): 190 """ 191 run_once implements the run_once call in test.test, is called to begin 192 execution of the test 193 194 @param host: host from control file with which to run the test 195 @param test_to_wrap: test name to execute in the wrapper 196 @param pdash_note: note to annotate results on the dashboard 197 @param wrap_args: args to pass to the wrapped test execution 198 @param pass_criteria: dictionary of criteria to compare results against 199 200 @raises error.TestFail: on failure of the wrapped tests 201 """ 202 logging.debug('running test_with_pass_criteria run_once') 203 logging.debug('with test name %s', str(self.tagged_testname)) 204 self._wrap_args = wrap_args 205 self._test_to_wrap = test_to_wrap 206 if self._test_to_wrap == None: 207 raise error.TestFail('No test_to_wrap given') 208 209 if isinstance(pass_criteria, dict): 210 self._pass_criteria = pass_criteria 211 else: 212 logging.info('loading from string dict %s', pass_criteria) 213 self._pass_criteria = json.loads(pass_criteria) 214 215 self._textproto_path = self._pass_criteria.get('textproto_path', None) 216 if self._textproto_path is None: 217 logging.info('not using textproto criteria definitions') 218 else: 219 self._pass_criteria.pop('textproto_path') 220 self._load_proto_to_pass_criteria() 221 222 logging.debug('wrapping test %s', self._test_to_wrap) 223 logging.debug('with wrap args %s', str(self._wrap_args)) 224 logging.debug('and pass criteria %s', str(self._pass_criteria)) 225 client_at = autotest.Autotest(host) 226 227 for test, argv in self._test_prefix: 228 argv['pdash_note'] = pdash_note 229 try: 230 client_at.run_test(test, check_client_result=True, **argv) 231 except: 232 raise error.TestFail('Prefix test failed, see log for details') 233 234 try: 235 client_at.run_test(self._test_to_wrap, 236 check_client_result=True, 237 **self._wrap_args) 238 except: 239 self.postprocess() 240 raise error.TestFail('Wrapped test failed, see log for details') 241 242 def postprocess(self): 243 """ 244 postprocess is called after the completion of run_once by the test framework 245 246 @raises error.TestFail: on any pass criteria failure 247 """ 248 self._parse_wrapped_results_keyvals() 249 if self._pass_criteria == {}: 250 return 251 self._criteria_to_keyvals = {} 252 self._find_matching_keyvals() 253 self._verify_criteria() 254