tools/testing/selftests/tc-testing/plugin-lib/nsPlugin.py

Source file repositories/reference/linux-study-clean/tools/testing/selftests/tc-testing/plugin-lib/nsPlugin.py

File Facts

System
Linux kernel
Corpus path
tools/testing/selftests/tc-testing/plugin-lib/nsPlugin.py
Extension
.py
Size
8321 bytes
Lines
259
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

import os
import signal
from string import Template
import subprocess
import time
from multiprocessing import Pool
from functools import cached_property
from TdcPlugin import TdcPlugin

from tdc_config import *

try:
    from pyroute2 import netns
    from pyroute2 import IPRoute
    netlink = True
except ImportError:
    netlink = False
    print("!!! Consider installing pyroute2 !!!")

class SubPlugin(TdcPlugin):
    def __init__(self):
        self.sub_class = 'ns/SubPlugin'
        super().__init__()

    def pre_suite(self, testcount, testlist):
        super().pre_suite(testcount, testlist)

    def prepare_test(self, test):
        if 'skip' in test and test['skip'] == 'yes':
            return

        if 'nsPlugin' not in test['plugins']:
            return

        if netlink == True:
            self._nl_ns_create()
        else:
            self._ipr2_ns_create()

        # Make sure the netns is visible in the fs
        ticks = 20
        while True:
            if ticks == 0:
                raise TimeoutError
            self._proc_check()
            try:
                ns = self.args.NAMES['NS']
                f = open('/run/netns/{}'.format(ns))
                f.close()
                break
            except:
                time.sleep(0.1)
                ticks -= 1
                continue

    def pre_case(self, test, test_skip):
        if self.args.verbose:
            print('{}.pre_case'.format(self.sub_class))

        if test_skip:
            return

        self.prepare_test(test)

    def post_case(self):
        if self.args.verbose:
            print('{}.post_case'.format(self.sub_class))

        if netlink == True:
            self._nl_ns_destroy()

Annotation

Implementation Notes