How to use _check_msg_level_change_supported method in lisa

Best Python code snippet using lisa_python

networksettings.py

Source:networksettings.py Github

copy

Full Screen

...408 priority=2,409 )410 def validate_device_msg_level_change(self, node: Node, log: Logger) -> None:411 # Check if feature is supported by the kernel412 self._check_msg_level_change_supported(node)413 msg_types: Dict[str, str] = {414 "probe": "0x0002",415 "tx_done": "0x0400",416 "rx_status": "0x0800",417 }418 ethtool = node.tools[Ethtool]419 devices_msg_level = ethtool.get_all_device_msg_level()420 for msg_level_info in devices_msg_level:421 interface = msg_level_info.device_name422 original_msg_level_number = msg_level_info.msg_level_number423 original_msg_level_name = msg_level_info.msg_level_name424 name_test_flag = []425 number_test_flag = 0426 for msg_key, msg_value in msg_types.items():427 if msg_key not in original_msg_level_name:428 name_test_flag.append(msg_key)429 number_test_flag += int(msg_value, 16)430 # variable to indicate set or unset431 set = True432 # if test message flags are already set, pick first test flag in list.433 # validate change by first unsetting the flag and then unsetting434 if not name_test_flag and not number_test_flag:435 first_pair = list(msg_types.items())[0]436 name_test_flag.append(first_pair[0])437 number_test_flag = int(first_pair[1], 16)438 set = False439 # Testing set/unset message level by name440 new_settings = ethtool.set_unset_device_message_flag_by_name(441 interface, name_test_flag, set442 )443 if set:444 assert_that(445 new_settings.msg_level_name,446 f"Setting msg flags - {' '.join(name_test_flag)} didn't"447 f" succeed. Current value is {new_settings.msg_level_name}",448 ).contains(" ".join(name_test_flag))449 else:450 assert_that(451 new_settings.msg_level_name,452 f"Setting msg flags by name - {' '.join(name_test_flag)} didn't"453 f" succeed. Current value is {new_settings.msg_level_name}",454 ).does_not_contain(" ".join(name_test_flag))455 reverted_settings = ethtool.set_unset_device_message_flag_by_name(456 interface, name_test_flag, not set457 )458 if not set:459 assert_that(460 reverted_settings.msg_level_name,461 f"Setting msg flags by name - {' '.join(name_test_flag)} didn't"462 f" succeed. Current value is {reverted_settings.msg_level_name}",463 ).contains(" ".join(name_test_flag))464 else:465 assert_that(466 reverted_settings.msg_level_name,467 f"Setting msg flags by name - {' '.join(name_test_flag)} didn't"468 f" succeed. Current value is {reverted_settings.msg_level_name}",469 ).does_not_contain(" ".join(name_test_flag))470 # Testing set message level by number471 new_settings = ethtool.set_device_message_flag_by_num(472 interface, str(hex(number_test_flag))473 )474 assert_that(475 int(new_settings.msg_level_number, 16),476 f"Setting msg flags by number - {str(hex(number_test_flag))} didn't"477 f" succeed. Current value is {new_settings.msg_level_number}",478 ).is_equal_to(number_test_flag)479 reverted_settings = ethtool.set_device_message_flag_by_num(480 interface, original_msg_level_number481 )482 assert_that(483 int(reverted_settings.msg_level_number, 16),484 f"Setting msg flags by number - {original_msg_level_number} didn't"485 f" succeed. Current value is {reverted_settings.msg_level_number}",486 ).is_equal_to(int(original_msg_level_number, 16))487 @TestCaseMetadata(488 description="""489 This test case requires 4 or more cpu cores, so as to validate490 among 4 or more channels(queues), no particular queue is continuously491 starving(not sending/receiving any packets).492 Steps:493 1. Get all the device's statistics.494 2. Validate device statistics lists per queue statistics as well.495 3. Run traffic using iperf3 and check stats for each device.496 4. if the same queue (say queue #0) is inactive repeatitively,497 and the count of channels is >= 4 (total #queues), test should fail498 and require further investigation.499 """,500 priority=2,501 requirement=simple_requirement(502 min_count=2,503 min_core_count=4,504 ),505 )506 def validate_device_statistics(self, environment: Environment, log: Logger) -> None:507 server_node = cast(RemoteNode, environment.nodes[0])508 client_node = cast(RemoteNode, environment.nodes[1])509 ethtool = client_node.tools[Ethtool]510 self._verify_stats_exists(server_node, client_node)511 starving_queues: Tuple[List[int], List[int]] = ([], [])512 prev_starved_queues = starving_queues513 an_enabled = False514 device = client_node.nics.default_nic515 nic = client_node.nics.get_nic(device)516 if nic.lower:517 # If AN is enabled on this interface then use VF nic stats.518 an_enabled = True519 device = nic.lower520 timeout = 300521 timer = create_timer()522 self._run_iperf3(server_node, client_node, run_time_seconds=timeout)523 while timer.elapsed(False) < timeout:524 per_tx_queue_packets: List[int] = []525 per_rx_queue_packets: List[int] = []526 device_stats = ethtool.get_device_statistics(device, True)527 if an_enabled:528 per_tx_queue_packets = [529 v530 for (k, v) in device_stats.counters.items()531 if self._vf_tx_stats_regex.search(k)532 ]533 per_rx_queue_packets = [534 v535 for (k, v) in device_stats.counters.items()536 if self._vf_rx_stats_regex.search(k)537 ]538 else:539 per_tx_queue_packets = [540 v541 for (k, v) in device_stats.counters.items()542 if self._tx_queue_stats_regex.search(k)543 ]544 per_rx_queue_packets = [545 v546 for (k, v) in device_stats.counters.items()547 if self._rx_queue_stats_regex.search(k)548 ]549 starving_queues = (550 [queue for queue, pkts in enumerate(per_tx_queue_packets) if pkts == 0],551 [queue for queue, pkts in enumerate(per_rx_queue_packets) if pkts == 0],552 )553 if (not prev_starved_queues) and (not starving_queues):554 # This means there is no queue that was starved in last check555 # and this check. It establishes no single queue is being starved.556 # Hence Test would PASS.557 return558 prev_starved_queues = (559 [560 element561 for element in prev_starved_queues[0]562 if element in starving_queues[0]563 ],564 [565 element566 for element in prev_starved_queues[1]567 if element in starving_queues[1]568 ],569 )570 starving_queues = ([], [])571 time.sleep(2)572 assert_that(573 prev_starved_queues[0],574 f"The tx stats for queue/queues {prev_starved_queues[0]} is/are 0."575 " This can have perf impact, please ensure all tx queues are used for"576 " traffic distribution.",577 ).is_empty()578 assert_that(579 prev_starved_queues[1],580 f"The rx stats for queue/queues {prev_starved_queues[1]} is/are 0"581 " This can have perf impact, please ensure all rx queues are used for"582 " traffic distribution.",583 ).is_empty()584 def after_case(self, log: Logger, **kwargs: Any) -> None:585 environment: Environment = kwargs.pop("environment")586 cleanup_iperf3(environment)587 def _check_msg_level_change_supported(self, node: Node) -> None:588 msg_level_symbols: Union[str, List[str]]589 uname_tool = node.tools[Uname]590 kernel_version = uname_tool.get_linux_information().kernel_version591 if not node.tools[KernelConfig].is_built_in("CONFIG_HYPERV_NET"):592 modinfo = node.tools[Modinfo]593 netvsc_module = modinfo.get_filename("hv_netvsc")594 # remove any escape character at the end of string595 netvsc_module = netvsc_module.strip()596 decompress_tool = ""597 # if the module is archived as xz, extract it to check symbols598 if netvsc_module.endswith(".xz"):599 decompress_tool = "xz"600 # if the module is archived as zst, extract it to check symbols601 if netvsc_module.endswith(".zst"):...

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