Commit 23710925 authored by Jakub Kicinski's avatar Jakub Kicinski
Browse files

selftests: drv-net: test dumping qstats per device 

Add a test for dumping qstats device by device.

ksft framework grows a ksft_raises() helper, to be used
under with, which should be familiar to unittest users.

Link: https://lore.kernel.org/r/20240420023543.3300306-5-kuba@kernel.org


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

tools/testing/selftests/drivers/net/stats.py

+59 −3
Original line number Diff line number Diff line 
#!/usr/bin/env python3

# SPDX-License-Identifier: GPL-2.0



from lib.py import ksft_run, ksft_exit

from lib.py import ksft_in, ksft_true, KsftSkipEx, KsftXfailEx

from lib.py import ksft_run, ksft_exit, ksft_pr

from lib.py import ksft_ge, ksft_eq, ksft_in, ksft_true, ksft_raises, KsftSkipEx, KsftXfailEx

from lib.py import EthtoolFamily, NetdevFamily, RtnlFamily, NlError

from lib.py import NetDrvEnv
@@ -77,9 +77,65 @@ def pkt_byte_sum(cfg) -> None: 
            raise Exception("Qstats are lower, fetched later")





def qstat_by_ifindex(cfg) -> None:

    global netfam

    global rtnl



    # Construct a map ifindex -> [dump, by-index, dump]

    ifindexes = {}

    stats = netfam.qstats_get({}, dump=True)

    for entry in stats:

        ifindexes[entry['ifindex']] = [entry, None, None]



    for ifindex in ifindexes.keys():

        entry = netfam.qstats_get({"ifindex": ifindex}, dump=True)

        ksft_eq(len(entry), 1)

        ifindexes[entry[0]['ifindex']][1] = entry[0]



    stats = netfam.qstats_get({}, dump=True)

    for entry in stats:

        ifindexes[entry['ifindex']][2] = entry



    if len(ifindexes) == 0:

        raise KsftSkipEx("No ifindex supports qstats")



    # Now make sure the stats match/make sense

    for ifindex, triple in ifindexes.items():

        all_keys = triple[0].keys() | triple[1].keys() | triple[2].keys()



        for key in all_keys:

            ksft_ge(triple[1][key], triple[0][key], comment="bad key: " + key)

            ksft_ge(triple[2][key], triple[1][key], comment="bad key: " + key)



    # Test invalid dumps

    # 0 is invalid

    with ksft_raises(NlError) as cm:

        netfam.qstats_get({"ifindex": 0}, dump=True)

    ksft_eq(cm.exception.nl_msg.error, -34)

    ksft_eq(cm.exception.nl_msg.extack['bad-attr'], '.ifindex')



    # loopback has no stats

    with ksft_raises(NlError) as cm:

        netfam.qstats_get({"ifindex": 1}, dump=True)

    ksft_eq(cm.exception.nl_msg.error, -95)

    ksft_eq(cm.exception.nl_msg.extack['bad-attr'], '.ifindex')



    # Try to get stats for lowest unused ifindex but not 0

    devs = rtnl.getlink({}, dump=True)

    all_ifindexes = set([dev["ifi-index"] for dev in devs])

    lowest = 2

    while lowest in all_ifindexes:

        lowest += 1



    with ksft_raises(NlError) as cm:

        netfam.qstats_get({"ifindex": lowest}, dump=True)

    ksft_eq(cm.exception.nl_msg.error, -19)

    ksft_eq(cm.exception.nl_msg.extack['bad-attr'], '.ifindex')





def main() -> None:

    with NetDrvEnv(__file__) as cfg:

        ksft_run([check_pause, check_fec, pkt_byte_sum],

        ksft_run([check_pause, check_fec, pkt_byte_sum, qstat_by_ifindex],

                 args=(cfg, ))

    ksft_exit()

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

+18 −0
Original line number Diff line number Diff line
@@ -53,6 +53,24 @@ def ksft_ge(a, b, comment=""): 
        _fail("Check failed", a, "<", b, comment)





class ksft_raises:

    def __init__(self, expected_type):

        self.exception = None

        self.expected_type = expected_type



    def __enter__(self):

        return self



    def __exit__(self, exc_type, exc_val, exc_tb):

        if exc_type is None:

            _fail(f"Expected exception {str(self.expected_type.__name__)}, none raised")

        elif self.expected_type != exc_type:

            _fail(f"Expected exception {str(self.expected_type.__name__)}, raised {str(exc_type.__name__)}")

        self.exception = exc_val

        # Suppress the exception if its the expected one

        return self.expected_type == exc_type





def ksft_busy_wait(cond, sleep=0.005, deadline=1, comment=""):

    end = time.monotonic() + deadline

    while True: