How to use get_dropped_count method in lisa

Best Python code snippet using lisa_python

performance.py

Source:performance.py Github

copy

Full Screen

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

Full Screen

Full Screen

functional.py

Source:functional.py Github

copy

Full Screen

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

Full Screen

Full Screen

requests_threshold.py

Source:requests_threshold.py Github

copy

Full Screen

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

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run lisa automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful