How to use _execute_rx_drop_test method in lisa

Best Python code snippet using lisa_python

performance.py

Source:performance.py

...147    )148    def perf_xdp_rx_drop_multithread_sriov(149        self, environment: Environment, log: Logger150    ) -> None:151        self._execute_rx_drop_test(152            environment,153            True,154            log,155        )156    @TestCaseMetadata(157        description="""158        This case tests the XDP drop performance by measuring Packets Per Second159        (PPS) and received rate with single send thread.160        see details from perf_xdp_rx_drop_multithread_sriov.161        """,162        priority=3,163        requirement=simple_requirement(164            min_nic_count=2, min_count=2, min_core_count=8, network_interface=Sriov()165        ),166    )167    def perf_xdp_rx_drop_singlethread_sriov(168        self, environment: Environment, log: Logger169    ) -> None:170        self._execute_rx_drop_test(171            environment,172            False,173            log,174        )175    @TestCaseMetadata(176        description="""177        This case compare and record latency impact of XDP component.178        The test use lagscope to send tcp packets. And then compare the latency179        with/without XDP component. If the gap is more than 40%, the test case180        fails.181        """,182        priority=3,183        requirement=simple_requirement(184            min_nic_count=2, min_count=2, min_core_count=8, network_interface=Sriov()185        ),186    )187    def perf_xdp_lagscope_latency(self, result: TestResult, log: Logger) -> None:188        self._execute_latency_test(189            result,190            Lagscope,191            log,192        )193    @TestCaseMetadata(194        description="""195        This case compare and record latency impact of XDP component.196        The test use ntttcp to send tcp packets. And then compare the latency197        with/without XDP component. If the gap is more than 40%, the test case198        fails.199        """,200        priority=3,201        requirement=simple_requirement(202            min_nic_count=2, min_count=2, min_core_count=8, network_interface=Sriov()203        ),204    )205    def perf_xdp_ntttcp_latency(self, result: TestResult, log: Logger) -> None:206        self._execute_latency_test(207            result,208            Ntttcp,209            log,210        )211    def _execute_latency_test(212        self,213        test_result: TestResult,214        tool_type: Type[Tool],215        log: Logger,216    ) -> None:217        environment = test_result.environment218        assert environment, "fail to get environment from testresult"219        server = environment.nodes[0]220        client = environment.nodes[1]221        server_xdpdump = get_xdpdump(server)222        server_xdpdump.make_by_build_type(BuildType.PERF)223        server_nic = server.nics.get_nic_by_index(1)224        # the latency is not stable in cloud environment, test multiple times225        # and aggregate the result.226        tested_runs = 5227        latency_without_xdp: List[float] = []228        latency_with_xdp: List[float] = []229        for _ in range(tested_runs):230            latency_without_xdp.append(231                self._send_packets_for_latency(server, client, test_result, tool_type)232            )233            try:234                server_xdpdump.start_async(nic_name=server_nic.upper, timeout=0)235                latency_with_xdp.append(236                    self._send_packets_for_latency(237                        server, client, test_result, tool_type238                    )239                )240            finally:241                server_kill = server.tools[Kill]242                server_kill.by_name("xdpdump")243        final_without_xdp = calculate_middle_average(latency_without_xdp)244        final_with_xdp = calculate_middle_average(latency_with_xdp)245        log.info(246            f"Latency with XDP: {final_with_xdp}us, "247            f"without XDP: {final_without_xdp}us. "248            f"Raw with XDP: {latency_with_xdp}, "249            f"without XDP: {latency_without_xdp}. "250        )251        assert_that(final_with_xdp / final_without_xdp).described_as(252            f"The XDP latency: {final_with_xdp}us shouldn't slower 40% than "253            f"the normal latency: {final_without_xdp}us."254        ).is_less_than_or_equal_to(_default_latency_threshold)255    def _send_packets_for_latency(256        self,257        server: Node,258        client: Node,259        test_result: TestResult,260        tool_type: Type[Tool],261    ) -> float:262        assert_that(tool_type).described_as("the tool is not supported").is_in(263            Lagscope, Ntttcp264        )265        # mypy doesn't work with generic type method "get". So use a266        # intermidiate variable tools to store it.267        tools: List[Any] = run_in_parallel(268            [269                partial(server.tools.get, tool_type),270                partial(client.tools.get, tool_type),271            ]272        )273        server_nic = server.nics.get_nic_by_index(1)274        if tool_type is Lagscope:275            server_lagscope: Lagscope = tools[0]276            client_lagscope: Lagscope = tools[1]277            try:278                run_in_parallel(279                    [server_lagscope.set_busy_poll, client_lagscope.set_busy_poll]280                )281                server_lagscope.run_as_server(ip=server_nic.ip_addr)282                result = client_lagscope.run_as_client(server_ip=server_nic.ip_addr)283                lagscope_messages = client_lagscope.create_latency_performance_messages(284                    result=result,285                    test_case_name=inspect.stack()[2].function,286                    test_result=test_result,287                )288                assert lagscope_messages289                assert_that(len(lagscope_messages)).described_as(290                    "at least one message is necessary"291                ).is_greater_than(0)292                return float(293                    sum(x.average_latency_us for x in lagscope_messages)294                    / len(lagscope_messages)295                )296            finally:297                for lagscope in [server_lagscope, client_lagscope]:298                    lagscope.kill()299                    lagscope.restore_busy_poll()300        else:301            ntttcp_messages = perf_ntttcp(302                test_result=test_result,303                udp_mode=False,304                connections=[1],305                test_case_name=inspect.stack()[2].function,306            )307            return float(308                # The type is always TCP message, because the above line set udp309                # to False. Ignore type error here, because UDP message has no310                # latency metrics.311                sum(x.latency_us for x in ntttcp_messages)  # type: ignore312                / len(ntttcp_messages)313            )314    def _execute_rx_drop_test(315        self,316        environment: Environment,317        is_multi_thread: bool,318        log: Logger,319        threshold: float = _default_received_threshold,320    ) -> None:321        sender = environment.nodes[0]322        receiver = environment.nodes[1]323        # install pktgen on sender, and xdpdump on receiver.324        try:325            tools: List[Any] = run_in_parallel(326                [partial(sender.tools.get, Pktgen), partial(get_xdpdump, receiver)]327            )328        except UnsupportedKernelException as identifier:...

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.