KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use get_dropped_count method in lisa

Best Python code snippet using lisa_python

performance.py

Source:performance.py

...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(410            node=forwarder,411            nic=forwarder_nic,412            previous_count=0,413            log=log,414        )415        original_dropped_count = get_dropped_count(416            node=receiver,417            nic=receiver_nic,418            previous_count=0,419            log=log,420        )421        try:422            # start xdpdump423            forwarder_xdpdump.start_async(nic_name=forwarder_nic.upper, timeout=0)424            receiver_xdpdump.start_async(nic_name=receiver_nic.upper, timeout=0)425            pktgen_result = self._send_packets(426                is_multi_threads, sender, pktgen, sender_nic, forwarder_nic427            )428            self._wait_packets_proceeded(429                log, receiver, receiver_nic, original_dropped_count430            )431        finally:432            # kill xdpdump processes.433            forwarder_kill = forwarder.tools[Kill]434            forwarder_kill.by_name("xdpdump")435            receiver_kill = receiver.tools[Kill]436            receiver_kill.by_name("xdpdump")437        # capture stats to calculate delta438        dropped_count = get_dropped_count(439            node=receiver,440            nic=receiver_nic,441            previous_count=original_dropped_count,442            log=log,443        )444        forwarded_count = get_forwarded_count(445            node=forwarder,446            nic=forwarder_nic,447            previous_count=original_forwarded_count,448            log=log,449        )450        # In some nodes like synthetic nic, there is no forward counter,451        # so count it by dropped count.452        validate_count = forwarded_count453        if not validate_count:454            validate_count = dropped_count455        log.debug(456            f"sender pktgen result: {pktgen_result}, "457            f"on forwarder: {forwarded_count}, "458            f"on receiver: {dropped_count}"459        )460        self._check_threshold(461            pktgen_result.sent_count, validate_count, threshold, "forwarded packets"462        )463    def _check_threshold(464        self, expected_count: int, actual_count: int, threshold: float, packet_name: str465    ) -> None:466        assert_that(actual_count / expected_count).described_as(467            f"{packet_name} rate should be above the threshold. "468            f"expected count: {expected_count}, actual count: {actual_count}"469        ).is_greater_than_or_equal_to(threshold)470    def _wait_packets_proceeded(471        self, log: Logger, receiver: Node, receiver_nic: NicInfo, original_count: int472    ) -> int:473        # wait until the forwarded count is not increased, it means there is474        # no more packets in queue.475        current_count = 0476        delta_count = 1477        while delta_count:478            sleep(1)479            previous_count = current_count480            current_count = get_dropped_count(481                node=receiver,482                nic=receiver_nic,483                previous_count=original_count,484                log=log,485            )486            delta_count = current_count - previous_count487            message = f"received {delta_count} new dropped packets in the 0.5 second."488            if delta_count > 0:489                message += " keep checking."490            log.debug(message)491        return previous_count492    def _send_packets(493        self,494        is_multi_threads: bool,...

functional.py

Source:functional.py

...125    ) -> 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="""...

requests_threshold.py

Source:requests_threshold.py

...127 at index 20, 21, and 22 is dropped because it breaches rule number 2.1311 at index 24, 25 and 26 is dropped because it breaches rule number 2.14Though the sliding window for 11 is considered from sec 2.15"""16def get_dropped_count(requests):17    size = len(requests)18    dropped_count = 019    dropped_dict = {}20    for i in range(size):21        if i > 2 and requests[i] == requests[i-3]:22            if requests[i] not in dropped_dict or dropped_dict[requests[i]] != i:23                dropped_dict[requests[i]] = i24                dropped_count += 125                print(f"Packet {requests[i]} at {i}th index dropped")26        elif i > 19 and requests[i] - requests[i-20] < 10:27            if requests[i] not in dropped_dict or dropped_dict[requests[i]] != i:28                dropped_dict[requests[i]] = i29                dropped_count += 130                print(f"Packet {requests[i]} at {i}th index dropped")31        elif i > 59 and requests[i] - requests[i-60] < 60:32            if requests[i] not in dropped_dict or dropped_dict[requests[i]] != i:33                dropped_dict[requests[i]] = i34                dropped_count += 135                print(f"Packet {requests[i]} at {i}th index dropped")36    return dropped_count37if __name__ == "__main__":38    requests_time = [1, 1, 1, 1, 2, 2, 2, 3, 3, 3, 4, 4, 4, 5, 5, 5, 6, 6, 6, 7, 7, 7, 7, 11, 11, 11, 11]...

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.