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

How to use test_by_ping method in lisa

Best Python code snippet using lisa_python

functional.py

Source:functional.py

...44    )45    def verify_xdp_basic(self, node: Node) -> None:46        for _ in range(3):47            xdpdump = get_xdpdump(node)48            output = xdpdump.test_by_ping()49            self._verify_xdpdump_result(output)50    @TestCaseMetadata(51        description="""52        It validates the XDP works with Synthetic network when SRIOV is53        disabled.54        1. Test in SRIOV mode.55        2. Disable the SRIOV.56        3. Test in Synthetic mode.57        """,58        priority=2,59        requirement=simple_requirement(min_count=2, network_interface=Sriov()),60    )61    def verify_xdp_sriov_failsafe(self, environment: Environment) -> None:62        xdp_node = environment.nodes[0]63        xdpdump = get_xdpdump(xdp_node)64        remote_address = self._get_ping_address(environment)65        # test in SRIOV mode.66        output = xdpdump.test_by_ping(67            xdp_node.nics.default_nic, remote_address=remote_address68        )69        self._verify_xdpdump_result(output)70        try:71            # disable SRIOV72            network = xdp_node.features[NetworkInterface]73            assert_that(network.is_enabled_sriov()).described_as(74                "SRIOV must be enabled when start the test."75            ).is_true()76            network.switch_sriov(False)77            # test in synthetic mode78            output = xdpdump.test_by_ping(79                xdp_node.nics.default_nic, remote_address=remote_address80            )81            self._verify_xdpdump_result(output)82        finally:83            # enable SRIOV back to recover environment84            network.switch_sriov(True)85    @TestCaseMetadata(86        description="""87        It validates the XDP works with Synthetic network.88        The test step is the same as verify_xdp_basic, but it run once only.89        """,90        priority=2,91        requirement=simple_requirement(network_interface=Synthetic()),92    )93    def verify_xdp_synthetic(self, node: Node) -> None:94        xdpdump = get_xdpdump(node)95        output = xdpdump.test_by_ping()96        self._verify_xdpdump_result(output)97    @TestCaseMetadata(98        description="""99        It validates XDP with multiple nics.100        1. Check current image supports XDP or not.101        2. Install and validate xdpdump.102        """,103        priority=3,104        requirement=simple_requirement(min_nic_count=3),105    )106    def verify_xdp_multiple_nics(self, node: Node) -> None:107        xdpdump = get_xdpdump(node)108        for i in range(3):109            nic_info = node.nics.get_nic_by_index(i)110            output = xdpdump.test_by_ping(nic_name=nic_info.upper)111            self._verify_xdpdump_result(output)112    @TestCaseMetadata(113        description="""114        It validates the XDP with action DROP.115        1. start tcpdump with icmp filter.116        2. start xdpdump.117        3. run ping 5 times.118        4. check tcpdump with 5 packets.119        """,120        priority=2,121        requirement=simple_requirement(min_count=2),122    )123    def verify_xdp_action_drop(124        self, environment: Environment, case_name: str, log: Logger125    ) -> None:126        # expect no response from the ping source side.127        captured_node = environment.nodes[0]128        default_nic = captured_node.nics.get_nic_by_index(0)129        original_count = get_dropped_count(130            node=captured_node,131            nic=default_nic,132            previous_count=0,133            log=log,134        )135        self._test_with_build_type(136            environment=environment,137            captured_node=captured_node,138            case_name=case_name,139            build_type=BuildType.ACTION_DROP,140            expected_tcp_packet_count=5,141            failure_message="DROP mode must have and only have sent packets "142            "at the send side.",143            expected_ping_success=False,144        )145        drop_count = get_dropped_count(146            node=captured_node,147            nic=default_nic,148            previous_count=original_count,149            log=log,150        )151        assert_that(drop_count).described_as(152            "the source side should have 5 dropped packets."153        ).is_equal_to(5)154        # expect no packet from the ping target side155        captured_node = environment.nodes[1]156        default_nic = captured_node.nics.get_nic_by_index(0)157        original_count = get_dropped_count(158            node=captured_node,159            nic=default_nic,160            previous_count=0,161            log=log,162        )163        self._test_with_build_type(164            environment=environment,165            captured_node=captured_node,166            case_name=case_name,167            build_type=BuildType.ACTION_DROP,168            expected_tcp_packet_count=0,169            failure_message="DROP mode must have no packet at target side.",170            expected_ping_success=False,171        )172        drop_count = get_dropped_count(173            node=captured_node,174            nic=default_nic,175            previous_count=original_count,176            log=log,177        )178        assert_that(drop_count).described_as(179            "the target side should have 5 dropped packets."180        ).is_equal_to(5)181    @TestCaseMetadata(182        description="""183        It validates the XDP with action TX.184        1. start tcpdump with icmp filter.185        2. start xdpdump.186        3. run ping 5 times.187        4. check tcpdump with 5 packets, because the icmp is replied in xdp188           level.189        """,190        priority=2,191        requirement=simple_requirement(min_count=2),192    )193    def verify_xdp_action_tx(self, environment: Environment, case_name: str) -> None:194        # tx has response packet from ping source side195        self._test_with_build_type(196            environment=environment,197            captured_node=environment.nodes[0],198            case_name=case_name,199            build_type=BuildType.ACTION_TX,200            expected_tcp_packet_count=10,201            failure_message="TX mode should receive response from ping source side.",202            expected_ping_success=True,203        )204        # tx has no packet from target side205        self._test_with_build_type(206            environment=environment,207            captured_node=environment.nodes[1],208            case_name=case_name,209            build_type=BuildType.ACTION_TX,210            expected_tcp_packet_count=0,211            failure_message="TX mode shouldn't capture any packets "212            "from the ping target node in tcp dump.",213            expected_ping_success=True,214        )215    @TestCaseMetadata(216        description="""217        It validates the XDP with action ABORT.218        1. start tcpdump with icmp filter.219        2. start xdpdump.220        3. run ping 5 times.221        4. check tcpdump with 5 packets.222        """,223        priority=3,224        requirement=simple_requirement(min_count=2),225    )226    def verify_xdp_action_aborted(227        self, environment: Environment, case_name: str228    ) -> None:229        # expect no response from the ping source side.230        self._test_with_build_type(231            environment=environment,232            captured_node=environment.nodes[0],233            case_name=case_name,234            build_type=BuildType.ACTION_ABORTED,235            expected_tcp_packet_count=5,236            failure_message="DROP mode must have and only have sent packets "237            "at the send side.",238            expected_ping_success=False,239        )240        # expect no packet from the ping target side241        self._test_with_build_type(242            environment=environment,243            captured_node=environment.nodes[1],244            case_name=case_name,245            build_type=BuildType.ACTION_ABORTED,246            expected_tcp_packet_count=0,247            failure_message="ABORT mode must have no packet at target side.",248            expected_ping_success=False,249        )250    @TestCaseMetadata(251        description="""252        It validates XDP with different MTU253        1. Check current image supports XDP or not.254        2. change MTU to 1500, 2000, 3506 to test XDP.255        """,256        priority=3,257        requirement=simple_requirement(min_count=2),258    )259    def verify_xdp_with_different_mtu(self, environment: Environment) -> None:260        xdp_node = environment.nodes[0]261        remote_node = environment.nodes[1]262        xdpdump = get_xdpdump(xdp_node)263        remote_address = self._get_ping_address(environment)264        tested_mtu: List[int] = [1500, 2000, 3506]265        xdp_node_nic_name = xdp_node.nics.default_nic266        remote_nic_name = remote_node.nics.default_nic267        xdp_node_ip = xdp_node.tools[Ip]268        remote_ip = remote_node.tools[Ip]269        original_xdp_node_mtu = xdp_node_ip.get_mtu(xdp_node_nic_name)270        original_remote_mtu = remote_ip.get_mtu(remote_nic_name)271        try:272            for mtu in tested_mtu:273                xdp_node_ip.set_mtu(xdp_node_nic_name, mtu)274                remote_ip.set_mtu(remote_nic_name, mtu)275                # tested mtu equals (ping mtu - IP headr (20) - ICMP header (8))276                xdpdump.test_by_ping(277                    xdp_node_nic_name,278                    remote_address=remote_address,279                    ping_package_size=mtu - 28,280                )281        finally:282            xdp_node_ip.set_mtu(xdp_node_nic_name, original_xdp_node_mtu)283            remote_ip.set_mtu(remote_nic_name, original_remote_mtu)284    @TestCaseMetadata(285        description="""286        It validates the XDP works with VF hot add/remove from API.287        1. Run xdp dump to drop and count packets.288        2. Remove VF from API.289        3. Run xdp dump to drop and count packets.290        5. Add VF back from API.291        6. Run xdp dump to drop and count packets.292        """,293        priority=3,294        requirement=simple_requirement(network_interface=Sriov()),295    )296    def verify_xdp_remove_add_vf(self, node: Node, log: Logger) -> None:297        xdpdump = get_xdpdump(node)298        nic_name = node.nics.default_nic299        nic_feature = node.features[NetworkInterface]300        default_nic = node.nics.get_nic_by_index(0)301        try:302            # validate xdp works with VF303            original_count = get_dropped_count(304                node=node,305                nic=default_nic,306                previous_count=0,307                log=log,308            )309            output = xdpdump.test_by_ping(310                nic_name=nic_name,311                build_type=BuildType.ACTION_DROP,312                expected_ping_success=False,313                remote_address=INTERNET_PING_ADDRESS,314            )315            self._verify_xdpdump_result(output)316            drop_count = get_dropped_count(317                node=node,318                nic=default_nic,319                previous_count=original_count,320                log=log,321            )322            assert_that(drop_count).described_as(323                "the source side should have 5 dropped packets when VF is enabled."324            ).is_equal_to(5)325            # disable VF326            nic_feature.switch_sriov(False)327            default_nic = node.nics.get_nic_by_index(0)328            # validate xdp works with synthetic329            original_count = get_dropped_count(330                node=node,331                nic=default_nic,332                previous_count=0,333                log=log,334            )335            output = xdpdump.test_by_ping(336                nic_name=nic_name,337                build_type=BuildType.ACTION_DROP,338                expected_ping_success=False,339                remote_address=INTERNET_PING_ADDRESS,340            )341            self._verify_xdpdump_result(output)342            drop_count = get_dropped_count(343                node=node,344                nic=default_nic,345                previous_count=original_count,346                log=log,347            )348            assert_that(drop_count).described_as(349                "There should be 5 dropped packets when VF is disabled."350            ).is_equal_to(5)351            # enable VF and validate xdp works with VF again352            nic_feature.switch_sriov(True)353            default_nic = node.nics.get_nic_by_index(0)354            original_count = get_dropped_count(355                node=node,356                nic=default_nic,357                previous_count=0,358                log=log,359            )360            output = xdpdump.test_by_ping(361                nic_name=nic_name,362                build_type=BuildType.ACTION_DROP,363                expected_ping_success=False,364                remote_address=INTERNET_PING_ADDRESS,365            )366            self._verify_xdpdump_result(output)367            drop_count = get_dropped_count(368                node=node,369                nic=default_nic,370                previous_count=original_count,371                log=log,372            )373            assert_that(drop_count).described_as(374                "the source side should have 5 dropped packets when VF back again."375            ).is_equal_to(5)376        finally:377            # recover sriov to on, it prevents the test fails when the sriov is378            # off.379            nic_feature.switch_sriov(True)380    @TestCaseMetadata(381        description="""382        It runs all tests of xdp-tools. Check the official site for more383        details.384        https://github.com/xdp-project/xdp-tools385        """,386        priority=3,387    )388    def verify_xdp_community_test(self, node: Node) -> None:389        try:390            xdptool = node.tools[XdpTool]391        except UnsupportedDistroException as identifier:392            raise SkippedException(identifier)393        xdptool.run_full_test()394    def _test_with_build_type(395        self,396        environment: Environment,397        captured_node: Node,398        case_name: str,399        build_type: BuildType,400        expected_tcp_packet_count: int,401        failure_message: str,402        expected_ping_success: bool,403    ) -> None:404        ping_source_node = environment.nodes[0]405        xdpdump = get_xdpdump(captured_node)406        tcpdump = captured_node.tools[TcpDump]407        ping_address = self._get_ping_address(environment)408        pcap_filename = f"{case_name}.pcap"409        tcpdump.dump_async(410            ping_source_node.nics.default_nic,411            filter=f'"icmp and host {ping_address}"',412            packet_filename=pcap_filename,413        )414        xdpdump.test_by_ping(415            ping_source_node.nics.default_nic,416            remote_address=ping_address,417            build_type=build_type,418            expected_ping_success=expected_ping_success,419            ping_source_node=ping_source_node,420        )421        kill = captured_node.tools[Kill]422        kill.by_name("tcpdump", SIGINT)423        packets = tcpdump.parse(pcap_filename)424        ping_node = cast(RemoteNode, ping_source_node)425        packets = [426            x427            for x in packets428            if x.destination == ping_node.internal_address...

xdpdump.py

Source:xdpdump.py

...108            result = process.wait_result()109        finally:110            self._restore_lro(nic_name)111        return result112    def test_by_ping(113        self,114        nic_name: str = "",115        timeout: int = 5,116        build_type: Optional[BuildType] = None,117        remote_address: str = "",118        expected_ping_success: bool = True,119        ping_package_size: Optional[int] = None,120        # the ping command can be triggered from different node121        ping_source_node: Optional[Node] = None,122    ) -> str:123        """124        Test with ICMP ping packets125        """126        if not nic_name:...

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.