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