KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use sriov_basic_test method in lisa
Best Python code snippet using lisa_python
sriov.py
Source:sriov.py 
...74 ),75 )76 def sriov_basic_validation(self, environment: Environment) -> None:77 vm_nics = initialize_nic_info(environment)78 sriov_basic_test(environment, vm_nics)79 @TestCaseMetadata(80 description="""81 This case verifies module of sriov network interface is loaded and82 each synthetic nic is paired with one VF, and check rx statistics of source83 and tx statistics of dest increase after send 200 Mb file from source to dest.84 Steps,85 1. Check VF of synthetic nic is paired.86 2. Check module of sriov network device is loaded.87 3. Check VF counts listed from lspci is expected.88 4. Setup SSH connection between source and dest with key authentication.89 5. Ping the dest IP from the source machine to check connectivity.90 6. Generate 200Mb file, copy from source to dest.91 7. Check rx statistics of source VF and tx statistics of dest VF is increased.92 """,93 priority=1,94 requirement=simple_requirement(95 min_count=2,96 network_interface=features.Sriov(),97 ),98 )99 def verify_sriov_single_vf_connection(self, environment: Environment) -> None:100 vm_nics = initialize_nic_info(environment)101 sriov_basic_test(environment, vm_nics)102 sriov_vf_connection_test(environment, vm_nics)103 @TestCaseMetadata(104 description="""105 This case needs 2 nodes and 64 Vcpus. And it verifies module of sriov network106 interface is loaded and each synthetic nic is paired with one VF, and check107 rx statistics of source and tx statistics of dest increase after send 200 Mb108 file from source to dest.109 Steps,110 1. Check VF of synthetic nic is paired.111 2. Check module of sriov network device is loaded.112 3. Check VF counts listed from lspci is expected.113 4. Setup SSH connection between source and dest with key authentication.114 5. Ping the dest IP from the source machine to check connectivity.115 6. Generate 200Mb file, copy from source to dest.116 7. Check rx statistics of source VF and tx statistics of dest VF is increased.117 """,118 priority=2,119 requirement=simple_requirement(120 min_count=2,121 min_core_count=64,122 network_interface=features.Sriov(),123 ),124 )125 def verify_sriov_single_vf_connection_max_cpu(126 self, environment: Environment127 ) -> None:128 vm_nics = initialize_nic_info(environment)129 sriov_basic_test(environment, vm_nics)130 sriov_vf_connection_test(environment, vm_nics)131 @TestCaseMetadata(132 description="""133 This case needs 2 nodes and 8 nics. And it verifies module of sriov network134 interface is loaded and each synthetic nic is paired with one VF, and check135 rx statistics of source and tx statistics of dest increase after send 200 Mb136 file from source to dest.137 Steps,138 1. Check VF of synthetic nic is paired.139 2. Check module of sriov network device is loaded.140 3. Check VF counts listed from lspci is expected.141 4. Setup SSH connection between source and dest with key authentication.142 5. Ping the dest IP from the source machine to check connectivity.143 6. Generate 200Mb file, copy from source to dest.144 7. Check rx statistics of source VF and tx statistics of dest VF is increased.145 """,146 priority=2,147 requirement=simple_requirement(148 min_count=2,149 network_interface=schema.NetworkInterfaceOptionSettings(150 nic_count=8,151 data_path=schema.NetworkDataPath.Sriov,152 ),153 ),154 )155 def verify_sriov_max_vf_connection(self, environment: Environment) -> None:156 vm_nics = initialize_nic_info(environment)157 sriov_basic_test(environment, vm_nics)158 sriov_vf_connection_test(environment, vm_nics)159 @TestCaseMetadata(160 description="""161 This case needs 2 nodes, 8 nics and 64 Vcpus. And it verifies module of sriov162 network interface is loaded and each synthetic nic is paired with one VF, and163 check rx statistics of source and tx statistics of dest increase after send 200164 Mb file from source to dest.165 Steps,166 1. Check VF of synthetic nic is paired.167 2. Check module of sriov network device is loaded.168 3. Check VF counts listed from lspci is expected.169 4. Setup SSH connection between source and dest with key authentication.170 5. Ping the dest IP from the source machine to check connectivity.171 6. Generate 200Mb file, copy from source to dest.172 7. Check rx statistics of source VF and tx statistics of dest VF is increased.173 """,174 priority=2,175 requirement=simple_requirement(176 min_count=2,177 min_core_count=64,178 network_interface=schema.NetworkInterfaceOptionSettings(179 nic_count=8,180 data_path=schema.NetworkDataPath.Sriov,181 ),182 ),183 )184 def verify_sriov_max_vf_connection_max_cpu(self, environment: Environment) -> None:185 vm_nics = initialize_nic_info(environment)186 sriov_basic_test(environment, vm_nics)187 sriov_vf_connection_test(environment, vm_nics)188 @TestCaseMetadata(189 description="""190 This case verify VM works well after disable and enable accelerated network in191 network interface through sdk.192 Steps,193 1. Do the basic sriov check.194 2. Set enable_accelerated_networking as False to disable sriov.195 3. Set enable_accelerated_networking as True to enable sriov.196 4. Do the basic sriov check.197 5. Do step 2 ~ step 4 for 2 times.198 """,199 priority=2,200 requirement=simple_requirement(201 network_interface=features.Sriov(),202 supported_platform_type=[AZURE],203 ),204 )205 def verify_sriov_disable_enable(self, environment: Environment) -> None:206 sriov_disable_enable(environment)207 @TestCaseMetadata(208 description="""209 This case verify VM works well after disable and enable PCI device inside VM.210 Steps,211 1. Disable sriov PCI device inside the VM.212 2. Enable sriov PCI device inside the VM.213 3. Do the basic sriov check.214 4. Do VF connection test.215 """,216 priority=2,217 requirement=simple_requirement(218 min_count=2,219 network_interface=features.Sriov(),220 ),221 )222 def verify_sriov_disable_enable_pci(self, environment: Environment) -> None:223 for node in environment.nodes.list():224 lspci = node.tools[Lspci]225 lspci.disable_devices_by_type(constants.DEVICE_TYPE_SRIOV)226 lspci.enable_devices()227 vm_nics = initialize_nic_info(environment)228 sriov_basic_test(environment, vm_nics)229 sriov_vf_connection_test(environment, vm_nics)230 @TestCaseMetadata(231 description="""232 This case verify VM works well after down the VF nic and up VF nic inside VM.233 Steps,234 1. Do the basic sriov check.235 2. Do network connection test with bring down the VF nic.236 3. After copy 200Mb file from source to desc.237 4. Check rx statistics of source synthetic nic and tx statistics of dest238 synthetic nic is increased.239 5. Bring up VF nic.240 """,241 priority=2,242 requirement=simple_requirement(243 min_count=2,244 network_interface=features.Sriov(),245 ),246 )247 def verify_sriov_disable_enable_on_guest(self, environment: Environment) -> None:248 vm_nics = initialize_nic_info(environment)249 sriov_basic_test(environment, vm_nics)250 sriov_vf_connection_test(environment, vm_nics, turn_off_vf=True)251 @TestCaseMetadata(252 description="""253 This case verify VM works well after attached the max sriov nics after254 provision.255 Steps,256 1. Attach 7 extra sriov nic into the VM.257 2. Do the basic sriov testing.258 """,259 priority=2,260 use_new_environment=True,261 requirement=simple_requirement(262 network_interface=schema.NetworkInterfaceOptionSettings(263 data_path=schema.NetworkDataPath.Sriov,264 max_nic_count=8,265 ),266 ),267 )268 def verify_sriov_add_max_nics(269 self, log_path: Path, log: Logger, environment: Environment270 ) -> None:271 remove_extra_nics(environment)272 try:273 node = cast(RemoteNode, environment.nodes[0])274 network_interface_feature = node.features[NetworkInterface]275 network_interface_feature.attach_nics(extra_nic_count=7)276 is_ready, tcp_error_code = wait_tcp_port_ready(277 node.public_address, node.public_port, log=log, timeout=self.TIME_OUT278 )279 if is_ready:280 vm_nics = initialize_nic_info(environment)281 sriov_basic_test(environment, vm_nics)282 else:283 serial_console = node.features[SerialConsole]284 serial_console.check_panic(285 saved_path=log_path, stage="after_attach_nics"286 )287 raise TcpConnectionException(288 node.public_address,289 node.public_port,290 tcp_error_code,291 "no panic found in serial log after attach nics",292 )293 finally:294 restore_extra_nics(environment)295 @TestCaseMetadata(296 description="""297 This case verify VM works well when provisioning with max (8) sriov nics.298 Steps,299 1. Provision VM with max network interfaces with enabling accelerated network.300 2. Do the basic sriov testing.301 """,302 priority=2,303 requirement=simple_requirement(304 min_nic_count=8,305 network_interface=features.Sriov(),306 ),307 )308 def verify_sriov_provision_with_max_nics(self, environment: Environment) -> None:309 vm_nics = initialize_nic_info(environment)310 sriov_basic_test(environment, vm_nics)311 @TestCaseMetadata(312 description="""313 This case verify VM works well when provisioning with max (8) sriov nics.314 Steps,315 1. Provision VM with max network interfaces with enabling accelerated network.316 2. Do the basic sriov testing.317 3. Reboot VM from guest.318 4. Do the basic sriov testing.319 """,320 priority=2,321 requirement=simple_requirement(322 min_nic_count=8,323 network_interface=features.Sriov(),324 ),325 )326 def verify_sriov_provision_with_max_nics_reboot(327 self, environment: Environment328 ) -> None:329 vm_nics = initialize_nic_info(environment)330 sriov_basic_test(environment, vm_nics)331 for node in environment.nodes.list():332 node.reboot()333 sriov_basic_test(environment, vm_nics)334 @TestCaseMetadata(335 description="""336 This case verify VM works well when provisioning with max (8) sriov nics.337 Steps,338 1. Provision VM with max network interfaces with enabling accelerated network.339 2. Do the basic sriov testing.340 3. Reboot VM from API.341 4. Do the basic sriov testing.342 """,343 priority=2,344 requirement=simple_requirement(345 min_nic_count=8,346 network_interface=features.Sriov(),347 ),348 )349 def verify_sriov_provision_with_max_nics_reboot_from_platform(350 self, environment: Environment351 ) -> None:352 vm_nics = initialize_nic_info(environment)353 sriov_basic_test(environment, vm_nics)354 for node in environment.nodes.list():355 start_stop = node.features[StartStop]356 start_stop.restart()357 sriov_basic_test(environment, vm_nics)358 @TestCaseMetadata(359 description="""360 This case verify VM works well when provisioning with max (8) sriov nics.361 Steps,362 1. Provision VM with max network interfaces with enabling accelerated network.363 2. Do the basic sriov testing.364 3. Stop and Start VM from API.365 4. Do the basic sriov testing.366 """,367 priority=2,368 requirement=simple_requirement(369 min_nic_count=8,370 network_interface=features.Sriov(),371 ),372 )373 def verify_sriov_provision_with_max_nics_stop_start_from_platform(374 self, environment: Environment375 ) -> None:376 vm_nics = initialize_nic_info(environment)377 sriov_basic_test(environment, vm_nics)378 for node in environment.nodes.list():379 start_stop = node.features[StartStop]380 start_stop.stop()381 start_stop.start()382 sriov_basic_test(environment, vm_nics)383 @TestCaseMetadata(384 description="""385 This case verify VM works well during remove and load sriov modules.386 Steps,387 1. Provision VM with max network interfaces with enabling accelerated network.388 2. Do the basic sriov testing.389 3. Remove sriov module, check network traffic through synthetic nic.390 4. Load sriov module, check network traffic through VF.391 """,392 priority=2,393 requirement=simple_requirement(394 min_count=2,395 min_nic_count=8,396 network_interface=features.Sriov(),397 ),398 )399 def verify_sriov_reload_modules(self, environment: Environment) -> None:400 for node in environment.nodes.list():401 if node.tools[KernelConfig].is_built_in(get_used_config(node)):402 raise SkippedException(403 "current VM's mlx driver is built-in, can not reload."404 )405 vm_nics = initialize_nic_info(environment)406 sriov_basic_test(environment, vm_nics)407 module_in_used: Dict[str, str] = {}408 for node in environment.nodes.list():409 module_in_used[node.name] = remove_module(node)410 sriov_vf_connection_test(environment, vm_nics, remove_module=True)411 for node in environment.nodes.list():412 load_module(node, module_in_used[node.name])413 vm_nics = initialize_nic_info(environment)414 sriov_vf_connection_test(environment, vm_nics)415 @TestCaseMetadata(416 description="""417 This case verify below two kernel patches.418 1. hv_netvsc: Sync offloading features to VF NIC419 https://github.com/torvalds/linux/commit/68622d071e555e1528f3e7807f30f73311c1acae#diff-007213ba7199932efdb096be47d209a2f83e4d425c486b3adaba861d0a0c80c5 # noqa: E501420 2. hv_netvsc: Allow scatter-gather feature to be tunable...
stress.py
Source:stress.py
...223 def verify_stress_sriov_with_max_nics_reboot(224 self, environment: Environment225 ) -> None:226 vm_nics = initialize_nic_info(environment)227 sriov_basic_test(environment, vm_nics)228 for _ in range(10):229 for node in environment.nodes.list():230 node.reboot()231 sriov_basic_test(environment, vm_nics)232 @TestCaseMetadata(233 description="""234 This case verify VM works well when provisioning with max (8) sriov nics.235 Steps,236 1. Provision VM with max network interfaces with enabling accelerated network.237 2. Do the basic sriov testing.238 3. Reboot VM from API.239 4. Do the basic sriov testing.240 5. Repeat step 3 and 4 for 10 times.241 """,242 priority=2,243 requirement=simple_requirement(244 min_nic_count=8,245 network_interface=features.Sriov(),246 ),247 )248 def verify_sriov_with_max_nics_reboot_from_platform(249 self, environment: Environment250 ) -> None:251 vm_nics = initialize_nic_info(environment)252 sriov_basic_test(environment, vm_nics)253 for _ in range(10):254 for node in environment.nodes.list():255 start_stop = node.features[StartStop]256 start_stop.restart()257 sriov_basic_test(environment, vm_nics)258 @TestCaseMetadata(259 description="""260 This case verify VM works well when provisioning with max (8) sriov nics.261 Steps,262 1. Provision VM with max network interfaces with enabling accelerated network.263 2. Do the basic sriov testing.264 3. Stop and Start VM from API.265 4. Do the basic sriov testing.266 5. Repeat step 3 and 4 for 10 times.267 """,268 priority=2,269 requirement=simple_requirement(270 min_nic_count=8,271 network_interface=features.Sriov(),272 ),273 )274 def verify_sriov_with_max_nics_stop_start_from_platform(275 self, environment: Environment276 ) -> None:277 vm_nics = initialize_nic_info(environment)278 sriov_basic_test(environment, vm_nics)279 for _ in range(10):280 for node in environment.nodes.list():281 start_stop = node.features[StartStop]282 start_stop.stop()283 start_stop.start()284 sriov_basic_test(environment, vm_nics)285 def after_case(self, log: Logger, **kwargs: Any) -> None:286 environment: Environment = kwargs.pop("environment")...
common.py
Source:common.py
...73def get_packets(node: Node, nic_name: str, name: str = "tx_packets") -> int:74 cat = node.tools[Cat]75 return int(cat.read(f"/sys/class/net/{nic_name}/statistics/{name}", force_run=True))76@retry(exceptions=AssertionError, tries=150, delay=2)77def sriov_basic_test(78 environment: Environment, vm_nics: Dict[str, Dict[str, NicInfo]]79) -> None:80 for node in environment.nodes.list():81 # 1. Check module of sriov network device is loaded.82 used_module = get_used_module(node)83 if not node.tools[KernelConfig].is_built_in(modules_config_dict[used_module]):84 lsmod = node.tools[Lsmod]85 assert_that(lsmod.module_exists(used_module, force_run=True)).described_as(86 "The module of sriov network device isn't loaded."87 ).is_true()88 # 2. Check VF counts listed from lspci is expected.89 lspci = node.tools[Lspci]90 devices_slots = lspci.get_device_names_by_type(91 constants.DEVICE_TYPE_SRIOV, force_run=True92 )93 assert_that(devices_slots).described_as(94 "count of sriov devices listed from lspci is not expected,"95 " please check the driver works properly"96 ).is_length(len([x for x in vm_nics[node.name].values() if x.lower != ""]))97def sriov_vf_connection_test(98 environment: Environment,99 vm_nics: Dict[str, Dict[str, NicInfo]],100 turn_off_vf: bool = False,101 remove_module: bool = False,102) -> None:103 source_node = cast(RemoteNode, environment.nodes[0])104 dest_node = cast(RemoteNode, environment.nodes[1])105 source_ssh = source_node.tools[Ssh]106 dest_ssh = dest_node.tools[Ssh]107 dest_ssh.enable_public_key(source_ssh.generate_key_pairs())108 # generate 200Mb file109 source_node.execute("dd if=/dev/urandom of=large_file bs=100 count=0 seek=2M")110 max_retry_times = 10111 for _, source_nic_info in vm_nics[source_node.name].items():112 matched_dest_nic_name = ""113 for dest_nic_name, dest_nic_info in vm_nics[dest_node.name].items():114 # only when IPs are in the same subnet, IP1 of machine A can connect to115 # IP2 of machine B116 # e.g. eth2 IP is 10.0.2.3 on machine A, eth2 IP is 10.0.3.4 on machine117 # B, use nic name doesn't work in this situation118 if (119 dest_nic_info.ip_addr.rsplit(".", maxsplit=1)[0]120 == source_nic_info.ip_addr.rsplit(".", maxsplit=1)[0]121 ):122 matched_dest_nic_name = dest_nic_name123 break124 assert_that(matched_dest_nic_name).described_as(125 f"can't find the same subnet nic with {source_nic_info.ip_addr} on"126 f" machine {source_node.name}, please check network setting of "127 f"machine {dest_node.name}."128 ).is_not_empty()129 desc_nic_info = vm_nics[dest_node.name][matched_dest_nic_name]130 dest_ip = vm_nics[dest_node.name][matched_dest_nic_name].ip_addr131 source_ip = source_nic_info.ip_addr132 source_synthetic_nic = source_nic_info.upper133 dest_synthetic_nic = desc_nic_info.upper134 source_nic = source_vf_nic = source_nic_info.lower135 dest_nic = dest_vf_nic = desc_nic_info.lower136 if remove_module or turn_off_vf:137 source_nic = source_synthetic_nic138 dest_nic = dest_synthetic_nic139 if turn_off_vf:140 source_node.execute(f"ip link set dev {source_vf_nic} down", sudo=True)141 dest_node.execute(f"ip link set dev {dest_vf_nic} down", sudo=True)142 # get origin tx_packets and rx_packets before copy file143 source_tx_packets_origin = get_packets(source_node, source_nic)144 dest_tx_packets_origin = get_packets(dest_node, dest_nic, "rx_packets")145 # check the connectivity between source and dest machine using ping146 for _ in range(max_retry_times):147 cmd_result = source_node.execute(148 f"ping -c 1 {dest_ip} -I {source_synthetic_nic}", sudo=True149 )150 if cmd_result.exit_code == 0:151 break152 cmd_result.assert_exit_code(153 message=f"fail to ping {dest_ip} from {source_node.name} to "154 f"{dest_node.name} after retry {max_retry_times}"155 )156 # copy 200 Mb file from source ip to dest ip157 cmd_result = source_node.execute(158 f"scp -o BindAddress={source_ip} -i ~/.ssh/id_rsa -o"159 f" StrictHostKeyChecking=no large_file "160 f"$USER@{dest_ip}:/tmp/large_file",161 shell=True,162 expected_exit_code=0,163 expected_exit_code_failure_message="Fail to copy file large_file from"164 f" {source_ip} to {dest_ip}",165 )166 source_tx_packets = get_packets(source_node, source_nic)167 dest_tx_packets = get_packets(dest_node, dest_nic, "rx_packets")168 # verify tx_packets value of source nic is increased after coping 200Mb file169 # from source to dest170 assert_that(171 int(source_tx_packets), "insufficient TX packets sent"172 ).is_greater_than(int(source_tx_packets_origin))173 # verify rx_packets value of dest nic is increased after receiving 200Mb174 # file from source to dest175 assert_that(176 int(dest_tx_packets), "insufficient RX packets received"177 ).is_greater_than(int(dest_tx_packets_origin))178 if turn_off_vf:179 source_node.execute(f"ip link set dev {source_vf_nic} up", sudo=True)180 dest_node.execute(f"ip link set dev {dest_vf_nic} up", sudo=True)181def cleanup_iperf3(environment: Environment) -> None:182 for node in environment.nodes.list():183 kill = node.tools[Kill]184 kill.by_name("iperf3")185def sriov_disable_enable(environment: Environment, times: int = 4) -> None:186 vm_nics = initialize_nic_info(environment)187 sriov_basic_test(environment, vm_nics)188 node = cast(RemoteNode, environment.nodes[0])189 network_interface_feature = node.features[NetworkInterface]190 for _ in range(times):191 sriov_is_enabled = network_interface_feature.is_enabled_sriov()192 if sriov_is_enabled:193 sriov_basic_test(environment, vm_nics)194 network_interface_feature.switch_sriov(enable=(not sriov_is_enabled))195def remove_extra_nics_per_node(node: Node) -> None:196 node = cast(RemoteNode, node)197 network_interface_feature = node.features[NetworkInterface]198 network_interface_feature.remove_extra_nics()199def remove_extra_nics(environment: Environment) -> None:200 for node in environment.nodes.list():201 remove_extra_nics_per_node(node)202def restore_extra_nics_per_node(node: Node) -> None:203 remove_extra_nics_per_node(node)204 network_interface_feature = node.features[NetworkInterface]205 network_interface_feature.attach_nics(206 network_interface_feature.origin_extra_sriov_nics_count,207 enable_accelerated_networking=True,...
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?
