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

How to use get_nic_by_index method in lisa

Best Python code snippet using lisa_python

dpdksuite.py

Source:dpdksuite.py

...124        # enable hugepages needed for dpdk EAL125        init_hugepages(node)126        try:127            # run OVS tests, providing OVS with the NIC info needed for DPDK init128            ovs.setup_ovs(node.nics.get_nic_by_index().pci_slot)129            # validate if OVS was able to initialize DPDK130            node.execute(131                "ovs-vsctl get Open_vSwitch . dpdk_initialized",132                sudo=True,133                expected_exit_code=0,134                expected_exit_code_failure_message=(135                    "OVS repoted that DPDK EAL failed to initialize."136                ),137            )138        finally:139            ovs.stop_ovs()140    @TestCaseMetadata(141        description="""142           Install and run ci test for NFF-Go on ubuntu143        """,144        priority=4,145        requirement=simple_requirement(146            min_core_count=8,147            min_nic_count=2,148            network_interface=Sriov(),149            unsupported_features=[Gpu, Infiniband],150            supported_features=[IsolatedResource],151        ),152    )153    def verify_dpdk_nff_go(154        self, node: Node, log: Logger, variables: Dict[str, Any]155    ) -> None:156        try:157            nff_go = node.tools[DpdkNffGo]158        except UnsupportedDistroException as err:159            raise SkippedException(err)160        # hugepages needed for dpdk tests161        init_hugepages(node)162        # run the nff-go tests163        nff_go.run_test()164    @TestCaseMetadata(165        description="""166           Build and run DPDK multiprocess client/server sample application.167           Requires 3 nics since client/server needs two ports + 1 nic for LISA168        """,169        priority=4,170        requirement=simple_requirement(171            min_nic_count=3,172            network_interface=Sriov(),173            unsupported_features=[Gpu, Infiniband],174            supported_features=[IsolatedResource],175        ),176    )177    def verify_dpdk_multiprocess(178        self, node: Node, log: Logger, variables: Dict[str, Any]179    ) -> None:180        # multiprocess test requires dpdk source.181        self._force_dpdk_default_source(variables)182        kill = node.tools[Kill]183        pmd = "failsafe"184        server_app_name = "dpdk-mp_server"185        client_app_name = "dpdk-mp_client"186        # initialize DPDK with sample applications selected for build187        test_kit = initialize_node_resources(188            node,189            log,190            variables,191            pmd,192            sample_apps=[193                "multi_process/client_server_mp/mp_server",194                "multi_process/client_server_mp/mp_client",195            ],196        )197        if test_kit.testpmd.is_connect_x3:198            raise SkippedException(199                "Unsupported Hardware: ConnectX3 does not support secondary process RX"200            )201        # enable hugepages needed for dpdk EAL202        init_hugepages(node)203        # setup and run mp_server application204        examples_path = test_kit.testpmd.dpdk_build_path.joinpath("examples")205        server_app_path = examples_path.joinpath(server_app_name)206        client_app_path = examples_path.joinpath(client_app_name)207        # EAL -l: start server on cores 1-2,208        # EAL -n: use 4 memory channels209        # APP: -p : set port bitmask to port 0 and 1210        # APP: -n : allow one client to connect211        server_proc = node.execute_async(212            (213                f"{server_app_path} -l 1-2 -n 4 "214                f"-b {node.nics.get_nic_by_index(0).pci_slot} -- -p 3 -n 1"215            ),216            sudo=True,217            shell=True,218        )219        # Wait for server to finish init220        server_proc.wait_output("APP: Finished Process Init.", timeout=5)221        # EAL -l: start client on core 3,222        # EAL --proc-type: client runs as secondary process.223        # APP: -n : client index is 0224        client_result = node.execute(225            (226                f"timeout -s INT 2 {client_app_path} --proc-type=secondary -l 3 -n 4"227                f" -b {node.nics.get_nic_by_index(0).pci_slot} -- -n 0"228            ),229            sudo=True,230            shell=True,231        )232        # client blocks and returns, kill server once client is finished.233        kill.by_name(str(server_app_name), signum=SIGINT)234        server_result = server_proc.wait_result()235        # perform the checks from v2236        assert_that(client_result.stdout).described_as(237            "Secondary process did not finish initialization"238        ).contains("APP: Finished Process Init")239        assert_that(client_result.stdout).described_as(240            "Secondary process did not start accepting packets from server"241        ).contains("Client process 0 handling packets")242        # mp_client returns a nonstandard positive number when killed w signal.243        # one would expect either 0 or 130 (killed by signal w sigint).244        # check that the nonsense number is at least the expected one.245        assert_that(client_result.exit_code).described_as(246            "dpdk-mp client exit code was unexpected"247        ).is_equal_to(124)248        assert_that(server_result.exit_code).is_equal_to(0)249    @TestCaseMetadata(250        description="""251            test sriov failsafe during vf revoke (receive side)252        """,253        priority=2,254        requirement=simple_requirement(255            min_core_count=8,256            min_nic_count=2,257            network_interface=Sriov(),258            min_count=2,259            unsupported_features=[Gpu, Infiniband],260            supported_features=[IsolatedResource],261        ),262    )263    def verify_dpdk_sriov_rescind_failover_receiver(264        self, environment: Environment, log: Logger, variables: Dict[str, Any]265    ) -> None:266        test_kits = init_nodes_concurrent(environment, log, variables, "failsafe")267        try:268            check_send_receive_compatibility(test_kits)269        except UnsupportedPackageVersionException as err:270            raise SkippedException(err)271        sender, receiver = test_kits272        # Want to only switch receiver sriov to avoid timing weirdness273        receiver.switch_sriov = True274        sender.switch_sriov = False275        kit_cmd_pairs = generate_send_receive_run_info("failsafe", sender, receiver)276        run_testpmd_concurrent(277            kit_cmd_pairs, DPDK_VF_REMOVAL_MAX_TEST_TIME, log, rescind_sriov=True278        )279        rescind_tx_pps_set = receiver.testpmd.get_mean_rx_pps_sriov_rescind()280        self._check_rx_or_tx_pps_sriov_rescind("RX", rescind_tx_pps_set)281    @TestCaseMetadata(282        description="""283            test sriov failsafe during vf revoke (send only version)284        """,285        priority=2,286        requirement=simple_requirement(287            min_core_count=8,288            min_nic_count=2,289            network_interface=Sriov(),290            unsupported_features=[Gpu, Infiniband],291            supported_features=[IsolatedResource],292        ),293    )294    def verify_dpdk_sriov_rescind_failover_send_only(295        self, node: Node, log: Logger, variables: Dict[str, Any]296    ) -> None:297        test_kit = initialize_node_resources(node, log, variables, "failsafe")298        testpmd = test_kit.testpmd299        test_nic = node.nics.get_nic_by_index()300        testpmd_cmd = testpmd.generate_testpmd_command(301            test_nic, 0, "txonly", "failsafe"302        )303        kit_cmd_pairs = {304            test_kit: testpmd_cmd,305        }306        run_testpmd_concurrent(307            kit_cmd_pairs, DPDK_VF_REMOVAL_MAX_TEST_TIME, log, rescind_sriov=True308        )309        rescind_tx_pps_set = testpmd.get_mean_tx_pps_sriov_rescind()310        self._check_rx_or_tx_pps_sriov_rescind("TX", rescind_tx_pps_set)311    def _check_rx_or_tx_pps_sriov_rescind(312        self, tx_or_rx: str, pps: Tuple[int, int, int]313    ) -> None:314        before_rescind, during_rescind, after_reenable = pps315        self._check_rx_or_tx_pps(tx_or_rx, before_rescind, sriov_enabled=True)316        self._check_rx_or_tx_pps(tx_or_rx, during_rescind, sriov_enabled=False)317        self._check_rx_or_tx_pps(tx_or_rx, after_reenable, sriov_enabled=True)318    def _check_rx_or_tx_pps(319        self, tx_or_rx: str, pps: int, sriov_enabled: bool = True320    ) -> None:321        if sriov_enabled:322            assert_that(pps).described_as(323                f"{tx_or_rx}-PPS ({pps}) should have been greater "324                "than 2^20 (~1m) PPS before sriov disable."325            ).is_greater_than(2**20)326        else:327            assert_that(pps).described_as(328                f"{tx_or_rx}-PPS ({pps}) should have been less "329                "than 2^20 (~1m) PPS after sriov disable."330            ).is_less_than(2**20)331    @TestCaseMetadata(332        description="""333            verify vpp is able to detect azure network interfaces334            1. run fd.io vpp install scripts335            2. install vpp from their repositories336            3. start vpp service337            4. check that azure interfaces are detected by vpp338        """,339        priority=4,340        requirement=simple_requirement(341            min_core_count=8,342            min_nic_count=2,343            network_interface=Sriov(),344            unsupported_features=[Gpu, Infiniband],345            supported_features=[IsolatedResource],346        ),347    )348    def verify_dpdk_vpp(349        self, node: Node, log: Logger, variables: Dict[str, Any]350    ) -> None:351        vpp = node.tools[DpdkVpp]352        vpp.install()353        net = node.nics354        nic = net.get_nic_by_index()355        # set devices to down and restart vpp service356        ip = node.tools[Ip]357        for dev in [nic.lower, nic.upper]:358            ip.down(dev)359        for dev in [nic.lower, nic.upper]:360            ip.addr_flush(dev)361        vpp.start()362        vpp.run_test()363    @TestCaseMetadata(364        description="""365            This test runs the dpdk ring ping utility from:366            https://github.com/shemminger/dpdk-ring-ping367            to measure the maximum latency for 99.999 percent of packets during368            the test run. The maximum should be under 200000 nanoseconds369            (.2 milliseconds).370            Not dependent on any specific PMD.371        """,372        priority=4,373        requirement=simple_requirement(374            min_core_count=8,375            network_interface=Sriov(),376            unsupported_features=[Gpu, Infiniband],377            supported_features=[IsolatedResource],378        ),379    )380    def verify_dpdk_ring_ping(381        self, node: Node, log: Logger, variables: Dict[str, Any]382    ) -> None:383        # ring ping requires dpdk source to run, since default is package_manager384        # we special case here to use to dpdk-stable as the default.385        self._force_dpdk_default_source(variables)386        # setup and unwrap the resources for this test387        test_kit = initialize_node_resources(node, log, variables, "failsafe")388        testpmd = test_kit.testpmd389        # grab a nic and run testpmd390        git = node.tools[Git]391        make = node.tools[Make]392        echo = node.tools[Echo]393        rping_build_env_vars = [394            "export RTE_TARGET=build",395            f"export RTE_SDK={str(testpmd.dpdk_path)}",396        ]397        echo.write_to_file(398            ";".join(rping_build_env_vars), node.get_pure_path("~/.bashrc"), append=True399        )400        git_path = git.clone(401            "https://github.com/shemminger/dpdk-ring-ping.git", cwd=node.working_path402        )403        make.run(404            shell=True,405            cwd=git_path,406            expected_exit_code=0,407            expected_exit_code_failure_message="make could not build rping project.",408        ).assert_exit_code()409        # run ringping for 30 seconds410        runcmd = "./build/rping -c 0x03 -n 2 --no-pci --no-huge -- -d 5 -t 10"411        result = node.execute(412            runcmd,413            shell=True,414            cwd=git_path,415            expected_exit_code=0,416            expected_exit_code_failure_message="rping program failed to run correctly.",417        )418        result.assert_exit_code()419        # get the max latency for 99.999 percent of enqueued 'packets'.420        result_regex = self._ring_ping_percentile_regex.search(result.stdout)421        if result_regex and len(result_regex.groups()) == 1:422            max_ping_measured = int(result_regex.group(1))423            assert_that(max_ping_measured).described_as(424                (425                    f"RingPing measured {max_ping_measured} as maximum ping latency,"426                    f" maximum should be less than {MAX_RING_PING_LIMIT_NS}"427                )428            ).is_less_than(MAX_RING_PING_LIMIT_NS)429        else:430            fail(431                (432                    "Could not get latency data from rping result. "433                    f"Search was for 'percentile 99.999 = ([0-9]+)'\n{result.stdout}\n"434                )435            )436    @TestCaseMetadata(437        description="""438            Tests a basic sender/receiver setup for default failsafe driver setup.439            Sender sends the packets, receiver receives them.440            We check both to make sure the received traffic is within the expected441            order-of-magnitude.442        """,443        priority=2,444        requirement=simple_requirement(445            min_core_count=8,446            min_nic_count=2,447            network_interface=Sriov(),448            min_count=2,449            unsupported_features=[Gpu, Infiniband],450            supported_features=[IsolatedResource],451        ),452    )453    def verify_dpdk_send_receive_multi_txrx_queue_failsafe(454        self, environment: Environment, log: Logger, variables: Dict[str, Any]455    ) -> None:456        try:457            verify_dpdk_send_receive_multi_txrx_queue(458                environment, log, variables, "failsafe"459            )460        except UnsupportedPackageVersionException as err:461            raise SkippedException(err)462    @TestCaseMetadata(463        description="""464            Tests a basic sender/receiver setup for default failsafe driver setup.465            Sender sends the packets, receiver receives them.466            We check both to make sure the received traffic is within the expected467            order-of-magnitude.468        """,469        priority=2,470        requirement=simple_requirement(471            min_core_count=8,472            min_nic_count=2,473            network_interface=Sriov(),474            min_count=2,475            unsupported_features=[Gpu, Infiniband],476            supported_features=[IsolatedResource],477        ),478    )479    def verify_dpdk_send_receive_multi_txrx_queue_netvsc(480        self, environment: Environment, log: Logger, variables: Dict[str, Any]481    ) -> None:482        try:483            verify_dpdk_send_receive_multi_txrx_queue(484                environment, log, variables, "netvsc"485            )486        except UnsupportedPackageVersionException as err:487            raise SkippedException(err)488    @TestCaseMetadata(489        description="""490            Tests a basic sender/receiver setup for default failsafe driver setup.491            Sender sends the packets, receiver receives them.492            We check both to make sure the received traffic is within the expected493            order-of-magnitude.494        """,495        priority=2,496        requirement=simple_requirement(497            min_core_count=8,498            min_nic_count=2,499            network_interface=Sriov(),500            min_count=2,501            unsupported_features=[Gpu, Infiniband],502            supported_features=[IsolatedResource],503        ),504    )505    def verify_dpdk_send_receive_failsafe(506        self, environment: Environment, log: Logger, variables: Dict[str, Any]507    ) -> None:508        try:509            verify_dpdk_send_receive(environment, log, variables, "failsafe")510        except UnsupportedPackageVersionException as err:511            raise SkippedException(err)512    @TestCaseMetadata(513        description="""514            Tests a basic sender/receiver setup for direct netvsc pmd setup.515            Sender sends the packets, receiver receives them.516            We check both to make sure the received traffic is within the expected517            order-of-magnitude.518        """,519        priority=2,520        requirement=simple_requirement(521            min_core_count=8,522            min_nic_count=2,523            network_interface=Sriov(),524            min_count=2,525            unsupported_features=[Gpu, Infiniband],526            supported_features=[IsolatedResource],527        ),528    )529    def verify_dpdk_send_receive_netvsc(530        self, environment: Environment, log: Logger, variables: Dict[str, Any]531    ) -> None:532        try:533            verify_dpdk_send_receive(environment, log, variables, "netvsc")534        except UnsupportedPackageVersionException as err:535            raise SkippedException(err)536    @TestCaseMetadata(537        description="""538            UIO basic functionality test.539            - Bind interface to uio_hv_generic540            - check that sysfs entry is created541            - unbind542            - check that the driver is unloaded.543            - rebind to original driver544        """,545        priority=2,546        requirement=simple_requirement(547            min_nic_count=2,548            network_interface=Sriov(),549            unsupported_features=[Gpu, Infiniband],550            supported_features=[IsolatedResource],551        ),552    )553    def verify_uio_binding(554        self, node: Node, log: Logger, variables: Dict[str, Any]555    ) -> None:556        lsmod = node.tools[Lsmod]557        modprobe = node.tools[Modprobe]558        nic = node.nics.get_nic_by_index()559        node.nics.get_nic_driver(nic.upper)560        if nic.bound_driver == "hv_netvsc":561            enable_uio_hv_generic_for_nic(node, nic)562        original_driver = nic.driver_sysfs_path563        node.nics.unbind(nic)564        node.nics.bind(nic, UIO_HV_GENERIC_SYSFS_PATH)565        node.execute(566            "test -e /dev/uio0",567            shell=True,568            expected_exit_code=0,569            expected_exit_code_failure_message=(570                "/dev/uio0 did not exist after driver bind"571            ),572        )...

performance.py

Source:performance.py

...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:329            raise SkippedException(identifier)330        # type annotations331        pktgen: Pktgen = tools[0]332        xdpdump: XdpDump = tools[1]333        sender_nic = sender.nics.get_nic_by_index(1)334        receiver_nic = receiver.nics.get_nic_by_index(1)335        xdpdump.make_by_build_type(build_type=BuildType.PERF_DROP)336        original_dropped_count = get_dropped_count(337            node=receiver,338            nic=receiver_nic,339            previous_count=0,340            log=log,341        )342        try:343            xdpdump.start_async(nic_name=receiver_nic.upper, timeout=0)344            pktgen_result = self._send_packets(345                is_multi_thread, sender, pktgen, sender_nic, receiver_nic346            )347            self._wait_packets_proceeded(348                log, receiver, receiver_nic, original_dropped_count349            )350        finally:351            receiver_kill = receiver.tools[Kill]352            receiver_kill.by_name("xdpdump")353        # capture stats to calculate delta354        dropped_count = get_dropped_count(355            node=receiver,356            nic=receiver_nic,357            previous_count=original_dropped_count,358            log=log,359        )360        log.debug(361            f"sender pktgen result: {pktgen_result}, "362            f"dropped on receiver: {dropped_count}"363        )364        self._check_threshold(365            pktgen_result.sent_count, dropped_count, threshold, "dropped packets"366        )367        assert_that(pktgen_result.pps).described_as(368            "pps must be greater than 1M."369        ).is_greater_than_or_equal_to(1000000)370    def _execute_tx_forward_test(371        self,372        environment: Environment,373        log: Logger,374        is_multi_threads: bool = False,375        threshold: float = _default_received_threshold,376    ) -> None:377        sender = environment.nodes[0]378        forwarder = environment.nodes[1]379        receiver = environment.nodes[2]380        # install pktgen on sender381        try:382            pktgen = sender.tools[Pktgen]383        except UnsupportedKernelException as identifier:384            raise SkippedException(identifier)385        # install xdp dump on forwarder and receiver386        forwarder_xdpdump, receiver_xdpdump = run_in_parallel(387            [388                partial(get_xdpdump, forwarder),389                partial(get_xdpdump, receiver),390            ],391            log=log,392        )393        sender_nic = sender.nics.get_nic_by_index(1)394        forwarder_nic = forwarder.nics.get_nic_by_index(1)395        receiver_nic = receiver.nics.get_nic_by_index(1)396        run_in_parallel(397            [398                partial(399                    forwarder_xdpdump.make_on_forwarder_role,400                    forwarder_nic=forwarder_nic,401                    receiver_nic=receiver_nic,402                ),403                partial(404                    receiver_xdpdump.make_by_build_type, build_type=BuildType.PERF_DROP405                ),406            ]407        )408        # capture existing stats to calculate delta409        original_forwarded_count = get_forwarded_count(...

functional.py

Source:functional.py

...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(...

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.