Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _verify_stats_exists method in lisa

Best Python code snippet using lisa_python

networksettings.py

Source:networksettings.py

...506    def validate_device_statistics(self, environment: Environment, log: Logger) -> None:507        server_node = cast(RemoteNode, environment.nodes[0])508        client_node = cast(RemoteNode, environment.nodes[1])509        ethtool = client_node.tools[Ethtool]510        self._verify_stats_exists(server_node, client_node)511        starving_queues: Tuple[List[int], List[int]] = ([], [])512        prev_starved_queues = starving_queues513        an_enabled = False514        device = client_node.nics.default_nic515        nic = client_node.nics.get_nic(device)516        if nic.lower:517            # If AN is enabled on this interface then use VF nic stats.518            an_enabled = True519            device = nic.lower520        timeout = 300521        timer = create_timer()522        self._run_iperf3(server_node, client_node, run_time_seconds=timeout)523        while timer.elapsed(False) < timeout:524            per_tx_queue_packets: List[int] = []525            per_rx_queue_packets: List[int] = []526            device_stats = ethtool.get_device_statistics(device, True)527            if an_enabled:528                per_tx_queue_packets = [529                    v530                    for (k, v) in device_stats.counters.items()531                    if self._vf_tx_stats_regex.search(k)532                ]533                per_rx_queue_packets = [534                    v535                    for (k, v) in device_stats.counters.items()536                    if self._vf_rx_stats_regex.search(k)537                ]538            else:539                per_tx_queue_packets = [540                    v541                    for (k, v) in device_stats.counters.items()542                    if self._tx_queue_stats_regex.search(k)543                ]544                per_rx_queue_packets = [545                    v546                    for (k, v) in device_stats.counters.items()547                    if self._rx_queue_stats_regex.search(k)548                ]549            starving_queues = (550                [queue for queue, pkts in enumerate(per_tx_queue_packets) if pkts == 0],551                [queue for queue, pkts in enumerate(per_rx_queue_packets) if pkts == 0],552            )553            if (not prev_starved_queues) and (not starving_queues):554                # This means there is no queue that was starved in last check555                # and this check. It establishes no single queue is being starved.556                # Hence Test would PASS.557                return558            prev_starved_queues = (559                [560                    element561                    for element in prev_starved_queues[0]562                    if element in starving_queues[0]563                ],564                [565                    element566                    for element in prev_starved_queues[1]567                    if element in starving_queues[1]568                ],569            )570            starving_queues = ([], [])571            time.sleep(2)572        assert_that(573            prev_starved_queues[0],574            f"The tx stats for queue/queues {prev_starved_queues[0]} is/are 0."575            " This can have perf impact, please ensure all tx queues are used for"576            " traffic distribution.",577        ).is_empty()578        assert_that(579            prev_starved_queues[1],580            f"The rx stats for queue/queues {prev_starved_queues[1]} is/are 0"581            " This can have perf impact, please ensure all rx queues are used for"582            " traffic distribution.",583        ).is_empty()584    def after_case(self, log: Logger, **kwargs: Any) -> None:585        environment: Environment = kwargs.pop("environment")586        cleanup_iperf3(environment)587    def _check_msg_level_change_supported(self, node: Node) -> None:588        msg_level_symbols: Union[str, List[str]]589        uname_tool = node.tools[Uname]590        kernel_version = uname_tool.get_linux_information().kernel_version591        if not node.tools[KernelConfig].is_built_in("CONFIG_HYPERV_NET"):592            modinfo = node.tools[Modinfo]593            netvsc_module = modinfo.get_filename("hv_netvsc")594            # remove any escape character at the end of string595            netvsc_module = netvsc_module.strip()596            decompress_tool = ""597            # if the module is archived as xz, extract it to check symbols598            if netvsc_module.endswith(".xz"):599                decompress_tool = "xz"600            # if the module is archived as zst, extract it to check symbols601            if netvsc_module.endswith(".zst"):602                decompress_tool = "zstd"603            if decompress_tool:604                node.execute(605                    f"cp {netvsc_module} {node.working_path}/", cwd=node.working_path606                )607                node.execute(608                    (609                        f"{decompress_tool} -d {node.working_path}/"610                        f"{netvsc_module.rsplit('/', 1)[-1]}"611                    ),612                    cwd=node.working_path,613                )614                netvsc_module = node.execute(615                    f"ls {node.working_path}/hv_netvsc.ko",616                    shell=True,617                    cwd=node.working_path,618                ).stdout619            assert node.shell.exists(620                PurePosixPath(netvsc_module)621            ), f"{netvsc_module} doesn't exist."622            nm = node.tools[Nm]623            msg_level_symbols = nm.get_symbol_table(netvsc_module)624        else:625            # if the module is builtin626            command = f"grep 'netvsc.*msglevel' '/boot/System.map-{kernel_version}'"627            result = node.execute(628                command,629                shell=True,630                sudo=True,631                expected_exit_code=0,632                expected_exit_code_failure_message="Couldn't get the message level"633                "symbols in System map.",634            )635            msg_level_symbols = result.stdout636        if ("netvsc_get_msglevel" not in msg_level_symbols) or (637            "netvsc_set_msglevel" not in msg_level_symbols638        ):639            raise SkippedException(640                f"Get/Set message level not supported on {kernel_version},"641                " Skipping test."642            )643    def _verify_stats_exists(644        self,645        server_node: RemoteNode,646        client_node: RemoteNode,647    ) -> None:648        ethtool = client_node.tools[Ethtool]649        try:650            devices_statistics = ethtool.get_all_device_statistics()651        except UnsupportedOperationException as identifier:652            raise SkippedException(identifier)653        per_queue_stats = 0654        per_vf_queue_stats = 0655        for device_stats in devices_statistics:656            nic = client_node.nics.get_nic(device_stats.interface)657            if nic.lower:...

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.