Commit 8510801a authored by Jakub Kicinski's avatar Jakub Kicinski
Browse files

selftests: drv-net: add ability to schedule cleanup with defer() 

This implements what I was describing in [1]. When writing a test
author can schedule cleanup / undo actions right after the creation
completes, eg:

  cmd("touch /tmp/file")
  defer(cmd, "rm /tmp/file")

defer() takes the function name as first argument, and the rest are
arguments for that function. defer()red functions are called in
inverse order after test exits. It's also possible to capture them
and execute earlier (in which case they get automatically de-queued).

  undo = defer(cmd, "rm /tmp/file")
  # ... some unsafe code ...
  undo.exec()

As a nice safety all exceptions from defer()ed calls are captured,
printed, and ignored (they do make the test fail, however).
This addresses the common problem of exceptions in cleanup paths
often being unhandled, leading to potential leaks.

There is a global action queue, flushed by ksft_run(). We could support
function level defers too, I guess, but there's no immediate need..

Link: https://lore.kernel.org/all/877cedb2ki.fsf@nvidia.com/

 # [1]
Reviewed-by: default avatarPrzemek Kitszel <przemyslaw.kitszel@intel.com>
Reviewed-by: default avatarPetr Machata <petrm@nvidia.com>
Link: https://patch.msgid.link/20240627185502.3069139-3-kuba@kernel.org


Signed-off-by: default avatarJakub Kicinski <kuba@kernel.org>
parent 147997af

tools/testing/selftests/net/lib/py/ksft.py

+21 −0
Original line number Diff line number Diff line
@@ -6,6 +6,7 @@ import sys 
import time

import traceback

from .consts import KSFT_MAIN_NAME

from .utils import global_defer_queue



KSFT_RESULT = None

KSFT_RESULT_ALL = True
@@ -108,6 +109,24 @@ def ktap_result(ok, cnt=1, case="", comment=""): 
    print(res)





def ksft_flush_defer():

    global KSFT_RESULT



    i = 0

    qlen_start = len(global_defer_queue)

    while global_defer_queue:

        i += 1

        entry = global_defer_queue.pop()

        try:

            entry.exec_only()

        except:

            ksft_pr(f"Exception while handling defer / cleanup (callback {i} of {qlen_start})!")

            tb = traceback.format_exc()

            for line in tb.strip().split('\n'):

                ksft_pr("Defer Exception|", line)

            KSFT_RESULT = False





def ksft_run(cases=None, globs=None, case_pfx=None, args=()):

    cases = cases or []
@@ -148,6 +167,8 @@ def ksft_run(cases=None, globs=None, case_pfx=None, args=()): 
            KSFT_RESULT = False

            cnt_key = 'fail'



        ksft_flush_defer()



        if not cnt_key:

            cnt_key = 'pass' if KSFT_RESULT else 'fail'

tools/testing/selftests/net/lib/py/utils.py

+34 −0
Original line number Diff line number Diff line
@@ -66,6 +66,40 @@ class bkg(cmd): 
        return self.process(terminate=self.terminate, fail=self.check_fail)





global_defer_queue = []





class defer:

    def __init__(self, func, *args, **kwargs):

        global global_defer_queue



        if not callable(func):

            raise Exception("defer created with un-callable object, did you call the function instead of passing its name?")



        self.func = func

        self.args = args

        self.kwargs = kwargs



        self._queue =  global_defer_queue

        self._queue.append(self)



    def __enter__(self):

        return self



    def __exit__(self, ex_type, ex_value, ex_tb):

        return self.exec()



    def exec_only(self):

        self.func(*self.args, **self.kwargs)



    def cancel(self):

        self._queue.remove(self)



    def exec(self):

        self.cancel()

        self.exec_only()





def tool(name, args, json=None, ns=None, host=None):

    cmd_str = name + ' '

    if json: