1# SPDX-License-Identifier: GPL-2.0 2 3import builtins 4import functools 5import inspect 6import sys 7import time 8import traceback 9from .consts import KSFT_MAIN_NAME 10from .utils import global_defer_queue 11 12KSFT_RESULT = None 13KSFT_RESULT_ALL = True 14KSFT_DISRUPTIVE = True 15 16 17class KsftFailEx(Exception): 18 pass 19 20 21class KsftSkipEx(Exception): 22 pass 23 24 25class KsftXfailEx(Exception): 26 pass 27 28 29def ksft_pr(*objs, **kwargs): 30 print("#", *objs, **kwargs) 31 32 33def _fail(*args): 34 global KSFT_RESULT 35 KSFT_RESULT = False 36 37 stack = inspect.stack() 38 started = False 39 for frame in reversed(stack[2:]): 40 # Start printing from the test case function 41 if not started: 42 if frame.function == 'ksft_run': 43 started = True 44 continue 45 46 ksft_pr("Check| At " + frame.filename + ", line " + str(frame.lineno) + 47 ", in " + frame.function + ":") 48 ksft_pr("Check| " + frame.code_context[0].strip()) 49 ksft_pr(*args) 50 51 52def ksft_eq(a, b, comment=""): 53 global KSFT_RESULT 54 if a != b: 55 _fail("Check failed", a, "!=", b, comment) 56 57 58def ksft_ne(a, b, comment=""): 59 global KSFT_RESULT 60 if a == b: 61 _fail("Check failed", a, "==", b, comment) 62 63 64def ksft_true(a, comment=""): 65 if not a: 66 _fail("Check failed", a, "does not eval to True", comment) 67 68 69def ksft_in(a, b, comment=""): 70 if a not in b: 71 _fail("Check failed", a, "not in", b, comment) 72 73 74def ksft_ge(a, b, comment=""): 75 if a < b: 76 _fail("Check failed", a, "<", b, comment) 77 78 79def ksft_lt(a, b, comment=""): 80 if a >= b: 81 _fail("Check failed", a, ">=", b, comment) 82 83 84class ksft_raises: 85 def __init__(self, expected_type): 86 self.exception = None 87 self.expected_type = expected_type 88 89 def __enter__(self): 90 return self 91 92 def __exit__(self, exc_type, exc_val, exc_tb): 93 if exc_type is None: 94 _fail(f"Expected exception {str(self.expected_type.__name__)}, none raised") 95 elif self.expected_type != exc_type: 96 _fail(f"Expected exception {str(self.expected_type.__name__)}, raised {str(exc_type.__name__)}") 97 self.exception = exc_val 98 # Suppress the exception if its the expected one 99 return self.expected_type == exc_type 100 101 102def ksft_busy_wait(cond, sleep=0.005, deadline=1, comment=""): 103 end = time.monotonic() + deadline 104 while True: 105 if cond(): 106 return 107 if time.monotonic() > end: 108 _fail("Waiting for condition timed out", comment) 109 return 110 time.sleep(sleep) 111 112 113def ktap_result(ok, cnt=1, case="", comment=""): 114 global KSFT_RESULT_ALL 115 KSFT_RESULT_ALL = KSFT_RESULT_ALL and ok 116 117 res = "" 118 if not ok: 119 res += "not " 120 res += "ok " 121 res += str(cnt) + " " 122 res += KSFT_MAIN_NAME 123 if case: 124 res += "." + str(case.__name__) 125 if comment: 126 res += " # " + comment 127 print(res) 128 129 130def ksft_flush_defer(): 131 global KSFT_RESULT 132 133 i = 0 134 qlen_start = len(global_defer_queue) 135 while global_defer_queue: 136 i += 1 137 entry = global_defer_queue.pop() 138 try: 139 entry.exec_only() 140 except: 141 ksft_pr(f"Exception while handling defer / cleanup (callback {i} of {qlen_start})!") 142 tb = traceback.format_exc() 143 for line in tb.strip().split('\n'): 144 ksft_pr("Defer Exception|", line) 145 KSFT_RESULT = False 146 147 148def ksft_disruptive(func): 149 """ 150 Decorator that marks the test as disruptive (e.g. the test 151 that can down the interface). Disruptive tests can be skipped 152 by passing DISRUPTIVE=False environment variable. 153 """ 154 155 @functools.wraps(func) 156 def wrapper(*args, **kwargs): 157 if not KSFT_DISRUPTIVE: 158 raise KsftSkipEx(f"marked as disruptive") 159 return func(*args, **kwargs) 160 return wrapper 161 162 163def ksft_setup(env): 164 """ 165 Setup test framework global state from the environment. 166 """ 167 168 def get_bool(env, name): 169 value = env.get(name, "").lower() 170 if value in ["yes", "true"]: 171 return True 172 if value in ["no", "false"]: 173 return False 174 try: 175 return bool(int(value)) 176 except: 177 raise Exception(f"failed to parse {name}") 178 179 if "DISRUPTIVE" in env: 180 global KSFT_DISRUPTIVE 181 KSFT_DISRUPTIVE = get_bool(env, "DISRUPTIVE") 182 183 return env 184 185 186def ksft_run(cases=None, globs=None, case_pfx=None, args=()): 187 cases = cases or [] 188 189 if globs and case_pfx: 190 for key, value in globs.items(): 191 if not callable(value): 192 continue 193 for prefix in case_pfx: 194 if key.startswith(prefix): 195 cases.append(value) 196 break 197 198 totals = {"pass": 0, "fail": 0, "skip": 0, "xfail": 0} 199 200 print("KTAP version 1") 201 print("1.." + str(len(cases))) 202 203 global KSFT_RESULT 204 cnt = 0 205 stop = False 206 for case in cases: 207 KSFT_RESULT = True 208 cnt += 1 209 comment = "" 210 cnt_key = "" 211 212 try: 213 case(*args) 214 except KsftSkipEx as e: 215 comment = "SKIP " + str(e) 216 cnt_key = 'skip' 217 except KsftXfailEx as e: 218 comment = "XFAIL " + str(e) 219 cnt_key = 'xfail' 220 except BaseException as e: 221 stop |= isinstance(e, KeyboardInterrupt) 222 tb = traceback.format_exc() 223 for line in tb.strip().split('\n'): 224 ksft_pr("Exception|", line) 225 if stop: 226 ksft_pr("Stopping tests due to KeyboardInterrupt.") 227 KSFT_RESULT = False 228 cnt_key = 'fail' 229 230 ksft_flush_defer() 231 232 if not cnt_key: 233 cnt_key = 'pass' if KSFT_RESULT else 'fail' 234 235 ktap_result(KSFT_RESULT, cnt, case, comment=comment) 236 totals[cnt_key] += 1 237 238 if stop: 239 break 240 241 print( 242 f"# Totals: pass:{totals['pass']} fail:{totals['fail']} xfail:{totals['xfail']} xpass:0 skip:{totals['skip']} error:0" 243 ) 244 245 246def ksft_exit(): 247 global KSFT_RESULT_ALL 248 sys.exit(0 if KSFT_RESULT_ALL else 1) 249