KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use _get_pps_sriov_rescind method in lisa

Best Python code snippet using lisa_python
dpdktestpmd.py
Source:dpdktestpmd.py
...375    def get_min_tx_pps(self) -> int:376        self._check_pps_data("TX")377        return min(self.tx_pps_data)378    def get_mean_tx_pps_sriov_rescind(self) -> Tuple[int, int, int]:379        return self._get_pps_sriov_rescind(self._tx_pps_key)380    def get_mean_rx_pps_sriov_rescind(self) -> Tuple[int, int, int]:381        return self._get_pps_sriov_rescind(self._rx_pps_key)382    def add_sample_apps_to_build_list(self, apps: Union[List[str], None]) -> None:383        if apps:384            self._sample_apps_to_build = apps385        else:386            self._sample_apps_to_build = []387    def __init__(self, *args: Any, **kwargs: Any) -> None:388        super().__init__(*args, **kwargs)389        self._dpdk_source = kwargs.pop("dpdk_source", PACKAGE_MANAGER_SOURCE)390        self._dpdk_branch = kwargs.pop("dpdk_branch", "main")391        self._sample_apps_to_build = kwargs.pop("sample_apps", [])392        self._dpdk_version_info = VersionInfo(0, 0)393        self._testpmd_install_path: str = ""394        self.find_testpmd_binary(assert_on_fail=False)395    def _determine_network_hardware(self) -> None:396        lspci = self.node.tools[Lspci]397        device_list = lspci.get_devices()398        self.is_connect_x3 = any(399            ["ConnectX-3" in dev.device_info for dev in device_list]400        )401    def _check_pps_data_exists(self, rx_or_tx: str) -> None:402        data_attr_name = f"{rx_or_tx.lower()}_pps_data"403        assert_that(hasattr(self, data_attr_name)).described_as(404            (405                f"PPS data ({rx_or_tx}) did not exist for testpmd object. "406                "This indicates either testpmd did not run or the suite is "407                "missing an assert. Contact the test maintainer."408            )409        ).is_true()410    def _check_pps_data(self, rx_or_tx: str) -> None:411        self._check_pps_data_exists(rx_or_tx)412        data_set: List[int] = []413        if rx_or_tx == "RX":414            data_set = self.rx_pps_data415        elif rx_or_tx == "TX":416            data_set = self.tx_pps_data417        else:418            fail(419                "Identifier passed to _check_pps_data was not recognized, "420                f"must be RX or TX. Found {rx_or_tx}"421            )422        assert_that(any(data_set)).described_as(423            f"any({str(data_set)}) resolved to false. Test data was "424            f"empty or all zeroes for dpdktestpmd.{rx_or_tx.lower()}_pps_data."425        ).is_true()426    def _install(self) -> bool:427        self._testpmd_output_after_reenable = ""428        self._testpmd_output_before_rescind = ""429        self._testpmd_output_during_rescind = ""430        self._last_run_output = ""431        self._determine_network_hardware()432        node = self.node433        if isinstance(node.os, Debian):434            repos = node.os.get_repositories()435            backport_repo = f"{node.os.information.codename}-backports"436            if any([backport_repo in repo.name for repo in repos]):437                self._debian_backports_args = [f"-t {backport_repo}"]438            else:439                self._debian_backports_args = []440        self._install_dependencies()441        # installing from distro package manager442        if self.use_package_manager_install():443            self.node.log.info(444                "Installing dpdk and dev package from package manager..."445            )446            if isinstance(node.os, Debian):447                node.os.install_packages(448                    ["dpdk", "dpdk-dev"],449                    extra_args=self._debian_backports_args,450                )451            elif isinstance(node.os, Fedora):452                node.os.install_packages(["dpdk", "dpdk-devel"])453            else:454                raise NotImplementedError(455                    "Dpdk package names are missing in dpdktestpmd.install"456                    f" for os {node.os.name}"457                )458            self._dpdk_version_info = node.os.get_package_information("dpdk")459            self.node.log.info(460                f"Installed DPDK version {str(self._dpdk_version_info)} "461                "from package manager"462            )463            self.find_testpmd_binary()464            self._load_drivers_for_dpdk()465            return True466        # otherwise install from source tarball or git467        self.node.log.info(f"Installing dpdk from source: {self._dpdk_source}")468        self._dpdk_repo_path_name = "dpdk"469        self.dpdk_path = self.node.working_path.joinpath(self._dpdk_repo_path_name)470        if self.find_testpmd_binary(471            assert_on_fail=False, check_path="/usr/local/bin"472        ):  # tools are already installed473            return True474        git_tool = node.tools[Git]475        echo_tool = node.tools[Echo]476        if self._dpdk_source and self._dpdk_source.endswith(".tar.gz"):477            wget_tool = node.tools[Wget]478            tar_tool = node.tools[Tar]479            if self._dpdk_branch:480                node.log.warn(481                    (482                        "DPDK tarball source does not need dpdk_branch defined. "483                        "User-defined variable dpdk_branch will be ignored."484                    )485                )486            working_path = str(node.working_path)487            wget_tool.get(488                self._dpdk_source,489                working_path,490            )491            dpdk_filename = self._dpdk_source.split("/")[-1]492            # extract tar into dpdk/ folder and discard old root folder name493            tar_tool.extract(494                str(node.working_path.joinpath(dpdk_filename)),495                str(self.dpdk_path),496                strip_components=1,497            )498            self.set_version_info_from_source_install(499                self._dpdk_source, self._version_info_from_tarball_regex500            )501        else:502            git_tool.clone(503                self._dpdk_source,504                cwd=node.working_path,505                dir_name=self._dpdk_repo_path_name,506            )507            if not self._dpdk_branch:508                # dpdk stopped using a default branch509                # if a branch is not specified, get latest version tag.510                self._dpdk_branch = git_tool.get_tag(511                    self.dpdk_path, filter=r"^v.*"  # starts w 'v'512                )513            git_tool.checkout(self._dpdk_branch, cwd=self.dpdk_path)514            self.set_version_info_from_source_install(515                self._dpdk_branch, self._version_info_from_git_tag_regex516            )517        self._load_drivers_for_dpdk()518        # add sample apps to compilation if they are present519        if self._sample_apps_to_build:520            sample_apps = f"-Dexamples={','.join(self._sample_apps_to_build)}"521        else:522            sample_apps = ""523        node.execute(524            f"meson {sample_apps} build",525            shell=True,526            cwd=self.dpdk_path,527            expected_exit_code=0,528            expected_exit_code_failure_message=(529                "meson build for dpdk failed, check that"530                "dpdk build has not changed to eliminate the use of meson or "531                "meson version is compatible with this dpdk version and OS."532            ),533        )534        self.dpdk_build_path = self.dpdk_path.joinpath("build")535        node.execute(536            "ninja",537            cwd=self.dpdk_build_path,538            timeout=1800,539            expected_exit_code=0,540            expected_exit_code_failure_message=(541                "ninja build for dpdk failed. check build spew for missing headers "542                "or dependencies. Also check that this ninja version requirement "543                "has not changed for dpdk."544            ),545        )546        node.execute(547            "ninja install",548            cwd=self.dpdk_build_path,549            sudo=True,550            expected_exit_code=0,551            expected_exit_code_failure_message=(552                "ninja install failed for dpdk binaries."553            ),554        )555        node.execute(556            "ldconfig",557            cwd=self.dpdk_build_path,558            sudo=True,559            expected_exit_code=0,560            expected_exit_code_failure_message="ldconfig failed, check for error spew.",561        )562        library_bashrc_lines = [563            "export PKG_CONFIG_PATH=${PKG_CONFIG_PATH}:/usr/local/lib64/pkgconfig/",564            "export LD_LIBRARY_PATH=${LD_LIBRARY_PATH}:/usr/local/lib64/",565        ]566        echo_tool.write_to_file(567            ";".join(library_bashrc_lines),568            node.get_pure_path("~/.bashrc"),569            append=True,570        )571        self.find_testpmd_binary(check_path="/usr/local/bin")572        return True573    def _load_drivers_for_dpdk(self) -> None:574        self.node.log.info("Loading drivers for infiniband, rdma, and mellanox hw...")575        if self.is_connect_x3:576            mellanox_drivers = ["mlx4_core", "mlx4_ib"]577        else:578            mellanox_drivers = ["mlx5_core", "mlx5_ib"]579        modprobe = self.node.tools[Modprobe]580        if isinstance(self.node.os, Ubuntu):581            # Ubuntu shouldn't need any special casing, skip to loading rdma/ib582            pass583        elif isinstance(self.node.os, Debian):584            # NOTE: debian buster doesn't include rdma and ib drivers585            # on 5.4 specifically for linux-image-cloud:586            # https://bugs.debian.org/cgi-bin/bugreport.cgi?bug=1012639587            # for backports on this release we should update the kernel to latest588            kernel_info = self.node.os.get_kernel_information(force_run=True)589            # update to at least 5.10 (known good for buster linux-image-cloud-(arch))590            if (591                self.node.os.information.codename == "buster"592                and kernel_info.version <= "5.10.0"593            ):594                self.node.log.debug(595                    f"Debian (buster) kernel version found: {str(kernel_info.version)} "596                    "Updating linux-image-cloud to most recent kernel."597                )598                # grab the linux-image package name from kernel version metadata599                linux_image_package = "linux-image-cloud-[a-zA-Z0-9]*"600                self.node.os.install_packages([linux_image_package])601                self.node.reboot()602        elif isinstance(self.node.os, Fedora):603            if not self.is_connect_x3:604                self.node.execute(605                    f"dracut --add-drivers '{' '.join(mellanox_drivers)} ib_uverbs' -f",606                    cwd=self.node.working_path,607                    expected_exit_code=0,608                    expected_exit_code_failure_message=(609                        "Issue loading mlx and ib_uverb drivers into ramdisk."610                    ),611                    sudo=True,612                )613        else:614            raise UnsupportedDistroException(self.node.os)615        rmda_drivers = ["ib_core", "ib_uverbs", "rdma_ucm"]616        # some versions of dpdk require these two, some don't.617        # some systems have them, some don't. Load if they're there.618        for module in ["ib_ipoib", "ib_umad"]:619            if modprobe.module_exists(module):620                rmda_drivers.append(module)621        modprobe.load(rmda_drivers)622        modprobe.load(mellanox_drivers)623    def _install_dependencies(self) -> None:624        node = self.node625        if isinstance(node.os, Ubuntu):626            self._install_ubuntu_dependencies()627        elif isinstance(node.os, Debian):628            node.os.install_packages(629                self._debian_packages, extra_args=self._debian_backports_args630            )631        elif isinstance(node.os, Fedora):632            self._install_fedora_dependencies()633        else:634            raise UnsupportedDistroException(635                node.os, "This OS does not have dpdk installation implemented yet."636            )637    def _install_ubuntu_dependencies(self) -> None:638        node = self.node639        ubuntu = node.os640        if not isinstance(ubuntu, Ubuntu):641            fail(642                "_install_ubuntu_dependencies was called on node "643                f"which was not Ubuntu: {node.os.information.full_version}"644            )645            return  # appease the type checker646        if ubuntu.information.version < "18.4.0":647            raise SkippedException(648                f"Ubuntu {str(ubuntu.information.version)} is not supported. "649                "Minimum documented version for DPDK support is >=18.04"650            )651        elif ubuntu.information.version < "20.4.0":652            ubuntu.install_packages(653                self._ubuntu_packages_1804,654                extra_args=self._debian_backports_args,655            )656            if not self.use_package_manager_install():657                self._install_ninja_and_meson()658        else:659            ubuntu.install_packages(660                self._ubuntu_packages_2004,661                extra_args=self._debian_backports_args,662            )663    def _install_fedora_dependencies(self) -> None:664        node = self.node665        rhel = node.os666        if not isinstance(rhel, Fedora):667            fail(668                "_install_fedora_dependencies was called on node "669                f"which was not Fedora: {node.os.information.full_version}"670            )671            return  # appease the type checker672        # DPDK is very sensitive to rdma-core/kernel mismatches673        # update to latest kernel before instaling dependencies674        rhel.install_packages("kernel")675        node.reboot()676        if rhel.information.version.major == 7:677            # Add packages for rhel7678            rhel.install_packages(["libmnl-devel", "libbpf-devel"])679        try:680            rhel.install_packages("kernel-devel-$(uname -r)")681        except MissingPackagesException:682            node.log.debug("kernel-devel-$(uname -r) not found. Trying kernel-devel")683            rhel.install_packages("kernel-devel")684        # RHEL 8 doesn't require special cases for installed packages.685        # TODO: RHEL9 may require updates upon release686        rhel.group_install_packages("Development Tools")687        rhel.group_install_packages("Infiniband Support")688        rhel.install_packages(self._fedora_packages)689        # ensure RDMA service is started if present.690        service_name = "rdma"691        service = node.tools[Service]692        if service.check_service_exists(service_name):693            if not service.check_service_status(service_name):694                service.enable_service(service_name)695            # some versions of RHEL and CentOS have service.rdma696            # that will refuse manual start/stop and will return697            # NOPERMISSION. This is not fatal and can be continued.698            # If the service is present it should start when needed.699            service.restart_service(700                service_name, ignore_exit_code=service.SYSTEMD_EXIT_NOPERMISSION701            )702        if not self.use_package_manager_install():703            self._install_ninja_and_meson()704    def _install_ninja_and_meson(self) -> None:705        node = self.node706        cwd = node.working_path707        node.execute(708            "pip3 install --upgrade meson",709            cwd=cwd,710            sudo=True,711            expected_exit_code=0,712            expected_exit_code_failure_message=(713                "Failed to update Meson to latest version with pip3"714            ),715        )716        if node.shell.exists(node.get_pure_path("/usr/bin/meson")):717            node.execute(718                "mv /usr/bin/meson /usr/bin/meson.bak",719                cwd=cwd,720                sudo=True,721                expected_exit_code=0,722                expected_exit_code_failure_message=(723                    "renaming previous meson binary or link in /usr/bin/meson"724                    " failed."725                ),726            )727        node.execute(728            "ln -s /usr/local/bin/meson /usr/bin/meson",729            cwd=cwd,730            sudo=True,731            expected_exit_code=0,732            expected_exit_code_failure_message=(733                "Failed to link new meson version as the " "default version in /usr/bin"734            ),735        )736        # NOTE: finding latest ninja is a pain,737        # so just fetch latest from github here738        wget_tool = self.node.tools[Wget]739        wget_tool.get(740            self._ninja_url,741            file_path=cwd.as_posix(),742            filename="ninja-linux.zip",743        )744        node.tools[Unzip].extract(745            file=str(cwd.joinpath("ninja-linux.zip")),746            dest_dir=str(cwd),747            sudo=True,748        )749        node.execute(750            "mv ninja /usr/bin/ninja",751            cwd=cwd,752            sudo=True,753            expected_exit_code=0,754            expected_exit_code_failure_message=(755                "Could not move latest ninja script after unzip into /usr/bin."756            ),757        )758        node.execute(759            "pip3 install --upgrade pyelftools",760            sudo=True,761            cwd=cwd,762            expected_exit_code=0,763            expected_exit_code_failure_message=(764                "Could not upgrade pyelftools with pip3."765            ),766        )767    def find_testpmd_binary(768        self, check_path: str = "", assert_on_fail: bool = True769    ) -> bool:770        node = self.node771        if self._testpmd_install_path:772            return True773        for bin_name in ["dpdk-testpmd", "testpmd"]:774            if check_path:775                bin_path = PurePosixPath(check_path).joinpath(bin_name)776                bin_name = str(bin_path)777            result = node.execute(f"which {bin_name}")778            if result.exit_code == 0:779                self._testpmd_install_path = result.stdout.strip()780                break781        found_path = PurePosixPath(self._testpmd_install_path)782        path_check = bool(self._testpmd_install_path) and node.shell.exists(found_path)783        if assert_on_fail and not path_check:784            fail("Could not locate testpmd binary after installation!")785        elif not path_check:786            self._testpmd_install_path = ""787        return path_check788    def _split_testpmd_output(self) -> None:789        search_str = "Port 0: device removal event"790        device_removal_index = self._last_run_output.find(search_str)791        assert_that(device_removal_index).described_as(792            "Could not locate SRIOV rescind event in testpmd output"793        ).is_not_equal_to(-1)794        self._testpmd_output_before_rescind = self._last_run_output[795            :device_removal_index796        ]797        after_rescind = self._last_run_output[device_removal_index:]798        # Identify the device add event799        hotplug_match = self._search_hotplug_regex.finditer(after_rescind)800        matches_list = list(hotplug_match)801        if not list(matches_list):802            hotplug_alt_match = self._search_hotplug_regex_alt.finditer(after_rescind)803            if hotplug_alt_match:804                matches_list = list(hotplug_alt_match)805            else:806                command_dumped = "timeout: the monitored command dumped core"807                if command_dumped in self._last_run_output:808                    raise LisaException("Testpmd crashed after device removal.")809        # pick the last match810        if len(matches_list) > 0:811            last_match = matches_list[-1]812        else:813            raise LisaException(814                "Found no vf hotplug events in testpmd output. "815                "Check output to verify if PPS drop occurred and port removal "816                "event message matches the expected forms."817            )818        self.node.log.info(f"Identified hotplug event: {last_match.group(0)}")819        before_reenable = after_rescind[: last_match.start()]820        after_reenable = after_rescind[last_match.end() :]821        self._testpmd_output_during_rescind = before_reenable822        self._testpmd_output_after_reenable = after_reenable823    def _get_pps_sriov_rescind(824        self,825        key_constant: str,826    ) -> Tuple[int, int, int]:827        if not all(828            [829                self._testpmd_output_during_rescind,830                self._testpmd_output_after_reenable,831                self._testpmd_output_before_rescind,832            ]833        ):834            self._split_testpmd_output()835        before_rescind = self.get_data_from_testpmd_output(836            key_constant,837            self._testpmd_output_before_rescind,...
Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.