KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
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!!
Was this article helpful?
