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

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

File Facts

System
Linux kernel
Corpus path
tools/testing/selftests/net/lib/py/utils.py
Extension
.py
Size
12184 bytes
Lines
376
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

# SPDX-License-Identifier: GPL-2.0

import json as _json
import os
import re
import select
import socket
import subprocess
import time


class CmdInitFailure(Exception):
    """ Command failed to start. Only raised by bkg(). """
    def __init__(self, msg, cmd_obj):
        super().__init__(msg + "\n" + repr(cmd_obj))
        self.cmd = cmd_obj


class CmdExitFailure(Exception):
    """ Command failed (returned non-zero exit code). """
    def __init__(self, msg, cmd_obj):
        super().__init__(msg + "\n" + repr(cmd_obj))
        self.cmd = cmd_obj


class CmdExitZeroFailure(CmdExitFailure):
    """ Command succeeded (returned zero exit code), but expected failure. """


def fd_read_timeout(fd, timeout):
    rlist, _, _ = select.select([fd], [], [], timeout)
    if rlist:
        return os.read(fd, 1024)
    raise TimeoutError("Timeout waiting for fd read")


class cmd:
    """
    Execute a command on local or remote host.

    @shell defaults to false, and class will try to split @comm into a list
    if it's a string with spaces.

    Use bkg() instead to run a command in the background.
    """
    def __init__(self, comm, shell=None, fail=True, expect_fail=False, ns=None,
                 background=False, host=None, timeout=5, ksft_ready=None,
                 ksft_wait=None):
        if ns:
            if hasattr(ns, 'user_ns_path'):
                comm = (f'nsenter --user={ns.user_ns_path} '
                        f'--net={ns.net_ns_path} --setuid=0 --setgid=0 -- '
                        + comm)
            else:
                comm = f'ip netns exec {ns} ' + comm

        self.stdout = None
        self.stderr = None
        self.ret = None
        self.ksft_term_fd = None

        self.host = host
        self.comm = comm

        if host:
            self.proc = host.cmd(comm)
        else:
            # If user doesn't explicitly request shell try to avoid it.
            if shell is None and isinstance(comm, str) and ' ' in comm:
                comm = comm.split()

Annotation

Implementation Notes