tools/testing/selftests/net/nk_qlease.py

Source file repositories/reference/linux-study-clean/tools/testing/selftests/net/nk_qlease.py

File Facts

System
Linux kernel
Corpus path
tools/testing/selftests/net/nk_qlease.py
Extension
.py
Size
66049 bytes
Lines
2110
Domain
Support Tooling And Documentation
Bucket
tools
Inferred role
Support Tooling And Documentation: tools
Status
atlas-only

Why This File Exists

Repository support layer: documentation, build tooling, samples, user-space helper tools, generated initramfs support, licenses, and validation utilities.

Dependency Surface

Detected Declarations

Annotated Snippet

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0

import errno
import time
from lib.py import (
    ksft_run,
    ksft_exit,
    ksft_eq,
    ksft_ne,
    ksft_in,
    ksft_not_in,
    ksft_raises,
)
from lib.py import (
    NetNS,
    NetNSEnter,
    EthtoolFamily,
    NetdevFamily,
    RtnlFamily,
    NetdevSimDev,
)
from lib.py import (
    NlError,
    Netlink,
    cmd,
    defer,
    ip,
)


def wait_until(cond, timeout=2.0, interval=0.05):
    deadline = time.monotonic() + timeout
    while not cond():
        if time.monotonic() >= deadline:
            return
        time.sleep(interval)


def create_netkit(rxqueues, mode="l2"):
    all_links = ip("-d link show", json=True)
    old_idxs = {
        link["ifindex"]
        for link in all_links
        if link.get("linkinfo", {}).get("info_kind") == "netkit"
    }

    rtnl = RtnlFamily()
    rtnl.newlink(
        {
            "linkinfo": {
                "kind": "netkit",
                "data": {
                    "mode": mode,
                    "policy": "forward",
                    "peer-policy": "forward",
                },
            },
            "num-rx-queues": rxqueues,
        },
        flags=[Netlink.NLM_F_CREATE, Netlink.NLM_F_EXCL],
    )

    all_links = ip("-d link show", json=True)
    nk_links = [
        link
        for link in all_links
        if link.get("linkinfo", {}).get("info_kind") == "netkit"
        and link["ifindex"] not in old_idxs
    ]

Annotation

Implementation Notes