249 lines · python
1# SPDX-License-Identifier: GPL-2.02 3import builtins4import functools5import inspect6import sys7import time8import traceback9from .consts import KSFT_MAIN_NAME10from .utils import global_defer_queue11 12KSFT_RESULT = None13KSFT_RESULT_ALL = True14KSFT_DISRUPTIVE = True15 16 17class KsftFailEx(Exception):18 pass19 20 21class KsftSkipEx(Exception):22 pass23 24 25class KsftXfailEx(Exception):26 pass27 28 29def ksft_pr(*objs, **kwargs):30 print("#", *objs, **kwargs)31 32 33def _fail(*args):34 global KSFT_RESULT35 KSFT_RESULT = False36 37 stack = inspect.stack()38 started = False39 for frame in reversed(stack[2:]):40 # Start printing from the test case function41 if not started:42 if frame.function == 'ksft_run':43 started = True44 continue45 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_RESULT54 if a != b:55 _fail("Check failed", a, "!=", b, comment)56 57 58def ksft_ne(a, b, comment=""):59 global KSFT_RESULT60 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 = None87 self.expected_type = expected_type88 89 def __enter__(self):90 return self91 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_val98 # Suppress the exception if its the expected one99 return self.expected_type == exc_type100 101 102def ksft_busy_wait(cond, sleep=0.005, deadline=1, comment=""):103 end = time.monotonic() + deadline104 while True:105 if cond():106 return107 if time.monotonic() > end:108 _fail("Waiting for condition timed out", comment)109 return110 time.sleep(sleep)111 112 113def ktap_result(ok, cnt=1, case="", comment=""):114 global KSFT_RESULT_ALL115 KSFT_RESULT_ALL = KSFT_RESULT_ALL and ok116 117 res = ""118 if not ok:119 res += "not "120 res += "ok "121 res += str(cnt) + " "122 res += KSFT_MAIN_NAME123 if case:124 res += "." + str(case.__name__)125 if comment:126 res += " # " + comment127 print(res)128 129 130def ksft_flush_defer():131 global KSFT_RESULT132 133 i = 0134 qlen_start = len(global_defer_queue)135 while global_defer_queue:136 i += 1137 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 = False146 147 148def ksft_disruptive(func):149 """150 Decorator that marks the test as disruptive (e.g. the test151 that can down the interface). Disruptive tests can be skipped152 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 wrapper161 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 True172 if value in ["no", "false"]:173 return False174 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_DISRUPTIVE181 KSFT_DISRUPTIVE = get_bool(env, "DISRUPTIVE")182 183 return env184 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 continue193 for prefix in case_pfx:194 if key.startswith(prefix):195 cases.append(value)196 break197 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_RESULT204 cnt = 0205 stop = False206 for case in cases:207 KSFT_RESULT = True208 cnt += 1209 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 = False228 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] += 1237 238 if stop:239 break240 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_ALL248 sys.exit(0 if KSFT_RESULT_ALL else 1)249