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