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

How to use _check_msg_level_change_supported method in lisa

Best Python code snippet using lisa_python

networksettings.py

Source:networksettings.py

...408        priority=2,409    )410    def validate_device_msg_level_change(self, node: Node, log: Logger) -> None:411        # Check if feature is supported by the kernel412        self._check_msg_level_change_supported(node)413        msg_types: Dict[str, str] = {414            "probe": "0x0002",415            "tx_done": "0x0400",416            "rx_status": "0x0800",417        }418        ethtool = node.tools[Ethtool]419        devices_msg_level = ethtool.get_all_device_msg_level()420        for msg_level_info in devices_msg_level:421            interface = msg_level_info.device_name422            original_msg_level_number = msg_level_info.msg_level_number423            original_msg_level_name = msg_level_info.msg_level_name424            name_test_flag = []425            number_test_flag = 0426            for msg_key, msg_value in msg_types.items():427                if msg_key not in original_msg_level_name:428                    name_test_flag.append(msg_key)429                    number_test_flag += int(msg_value, 16)430            # variable to indicate set or unset431            set = True432            # if test message flags are already set, pick first test flag in list.433            # validate change by first unsetting the flag and then unsetting434            if not name_test_flag and not number_test_flag:435                first_pair = list(msg_types.items())[0]436                name_test_flag.append(first_pair[0])437                number_test_flag = int(first_pair[1], 16)438                set = False439            # Testing set/unset message level by name440            new_settings = ethtool.set_unset_device_message_flag_by_name(441                interface, name_test_flag, set442            )443            if set:444                assert_that(445                    new_settings.msg_level_name,446                    f"Setting msg flags - {' '.join(name_test_flag)} didn't"447                    f" succeed. Current value is {new_settings.msg_level_name}",448                ).contains(" ".join(name_test_flag))449            else:450                assert_that(451                    new_settings.msg_level_name,452                    f"Setting msg flags by name - {' '.join(name_test_flag)} didn't"453                    f" succeed. Current value is {new_settings.msg_level_name}",454                ).does_not_contain(" ".join(name_test_flag))455            reverted_settings = ethtool.set_unset_device_message_flag_by_name(456                interface, name_test_flag, not set457            )458            if not set:459                assert_that(460                    reverted_settings.msg_level_name,461                    f"Setting msg flags by name - {' '.join(name_test_flag)} didn't"462                    f" succeed. Current value is {reverted_settings.msg_level_name}",463                ).contains(" ".join(name_test_flag))464            else:465                assert_that(466                    reverted_settings.msg_level_name,467                    f"Setting msg flags by name - {' '.join(name_test_flag)} didn't"468                    f" succeed. Current value is {reverted_settings.msg_level_name}",469                ).does_not_contain(" ".join(name_test_flag))470            # Testing set message level by number471            new_settings = ethtool.set_device_message_flag_by_num(472                interface, str(hex(number_test_flag))473            )474            assert_that(475                int(new_settings.msg_level_number, 16),476                f"Setting msg flags by number - {str(hex(number_test_flag))} didn't"477                f" succeed. Current value is {new_settings.msg_level_number}",478            ).is_equal_to(number_test_flag)479            reverted_settings = ethtool.set_device_message_flag_by_num(480                interface, original_msg_level_number481            )482            assert_that(483                int(reverted_settings.msg_level_number, 16),484                f"Setting msg flags by number - {original_msg_level_number} didn't"485                f" succeed. Current value is {reverted_settings.msg_level_number}",486            ).is_equal_to(int(original_msg_level_number, 16))487    @TestCaseMetadata(488        description="""489            This test case requires 4 or more cpu cores, so as to validate490            among 4 or more channels(queues), no particular queue is continuously491            starving(not sending/receiving any packets).492            Steps:493            1. Get all the device's statistics.494            2. Validate device statistics lists per queue statistics as well.495            3. Run traffic using iperf3 and check stats for each device.496            4. if the same queue (say queue #0) is inactive repeatitively,497                and the count of channels is >= 4 (total #queues), test should fail498                and require further investigation.499        """,500        priority=2,501        requirement=simple_requirement(502            min_count=2,503            min_core_count=4,504        ),505    )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"):...

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.