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

How to use get_nic_datapath method in lisa

Best Python code snippet using lisa_python

common.py

Source:common.py

...91        other_fields=other_fields,92    )93    for fio_message in fio_messages:94        notifier.notify(fio_message)95def get_nic_datapath(node: Node) -> str:96    data_path: str = ""97    assert (98        node.capability.network_interface99        and node.capability.network_interface.data_path100    )101    if isinstance(node.capability.network_interface.data_path, NetworkDataPath):102        data_path = node.capability.network_interface.data_path.value103    return data_path104def cleanup_process(environment: Environment, process_name: str) -> None:105    for node in environment.nodes.list():106        kill = node.tools[Kill]107        kill.by_name(process_name)108def reset_partitions(109    node: Node,110    disk_names: List[str],111) -> List[str]:112    fdisk = node.tools[Fdisk]113    partition_disks: List[str] = []114    for data_disk in disk_names:115        fdisk.delete_partitions(data_disk)116        partition_disks.append(fdisk.make_partition(data_disk, format=False))117    return partition_disks118def stop_raid(node: Node) -> None:119    mdadm = node.tools[Mdadm]120    mdadm.stop_raid()121def reset_raid(node: Node, disk_list: List[str]) -> None:122    stop_raid(node)123    mdadm = node.tools[Mdadm]124    mdadm.create_raid(disk_list)125def perf_tcp_latency(test_result: TestResult) -> List[NetworkLatencyPerformanceMessage]:126    environment = test_result.environment127    assert environment, "fail to get environment from testresult"128    client = cast(RemoteNode, environment.nodes[0])129    server = cast(RemoteNode, environment.nodes[1])130    client_lagscope = client.tools[Lagscope]131    server_lagscope = server.tools[Lagscope]132    try:133        for lagscope in [client_lagscope, server_lagscope]:134            lagscope.set_busy_poll()135        server_lagscope.run_as_server(ip=server.internal_address)136        latency_perf_messages = client_lagscope.create_latency_performance_messages(137            client_lagscope.run_as_client(server_ip=server.internal_address),138            inspect.stack()[1][3],139            test_result,140        )141    finally:142        for lagscope in [client_lagscope, server_lagscope]:143            lagscope.restore_busy_poll()144    return latency_perf_messages145def perf_tcp_pps(146    test_result: TestResult,147    test_type: str,148    server: Optional[RemoteNode] = None,149    client: Optional[RemoteNode] = None,150) -> None:151    # Either server and client are set explicitly or we use the first two nodes152    # from the environment. We never combine the two options. We need to specify153    # server and client explicitly for nested VM's which are not part of the154    # `environment` and are created during the test.155    if server is not None or client is not None:156        assert server is not None, "server need to be specified, if client is set"157        assert client is not None, "client need to be specified, if server is set"158    else:159        environment = test_result.environment160        assert environment, "fail to get environment from testresult"161        # set server and client from environment, if not set explicitly162        server = cast(RemoteNode, environment.nodes[1])163        client = cast(RemoteNode, environment.nodes[0])164    client_netperf, server_netperf = run_in_parallel(165        [lambda: client.tools[Netperf], lambda: server.tools[Netperf]]  # type: ignore166    )167    cpu = client.tools[Lscpu]168    core_count = cpu.get_core_count()169    if "maxpps" == test_type:170        ssh = client.tools[Ssh]171        ssh.set_max_session()172        client.close()173        ports = range(30000, 30032)174    else:175        ports = range(30000, 30001)176    for port in ports:177        server_netperf.run_as_server(port)178    for port in ports:179        client_netperf.run_as_client_async(server.internal_address, core_count, port)180    client_sar = client.tools[Sar]181    server_sar = server.tools[Sar]182    server_sar.get_statistics_async()183    result = client_sar.get_statistics()184    pps_message = client_sar.create_pps_performance_messages(185        result, inspect.stack()[1][3], test_type, test_result186    )187    notifier.notify(pps_message)188def perf_ntttcp(189    test_result: TestResult,190    server: Optional[RemoteNode] = None,191    client: Optional[RemoteNode] = None,192    udp_mode: bool = False,193    connections: Optional[List[int]] = None,194    test_case_name: str = "",195    server_nic_name: Optional[str] = None,196    client_nic_name: Optional[str] = None,197) -> List[Union[NetworkTCPPerformanceMessage, NetworkUDPPerformanceMessage]]:198    # Either server and client are set explicitly or we use the first two nodes199    # from the environment. We never combine the two options. We need to specify200    # server and client explicitly for nested VM's which are not part of the201    # `environment` and are created during the test.202    if server is not None or client is not None:203        assert server is not None, "server need to be specified, if client is set"204        assert client is not None, "client need to be specified, if server is set"205    else:206        environment = test_result.environment207        assert environment, "fail to get environment from testresult"208        # set server and client from environment, if not set explicitly209        server = cast(RemoteNode, environment.nodes[1])210        client = cast(RemoteNode, environment.nodes[0])211    if not test_case_name:212        # if it's not filled, assume it's called by case directly.213        test_case_name = inspect.stack()[1][3]214    if connections is None:215        if udp_mode:216            connections = NTTTCP_UDP_CONCURRENCY217        else:218            connections = NTTTCP_TCP_CONCURRENCY219    client_ntttcp, server_ntttcp = run_in_parallel(220        [lambda: client.tools[Ntttcp], lambda: server.tools[Ntttcp]]  # type: ignore221    )222    try:223        client_lagscope, server_lagscope = run_in_parallel(224            [225                lambda: client.tools[Lagscope],  # type: ignore226                lambda: server.tools[Lagscope],  # type: ignore227            ]228        )229        for ntttcp in [client_ntttcp, server_ntttcp]:230            ntttcp.setup_system(udp_mode)231        for lagscope in [client_lagscope, server_lagscope]:232            lagscope.set_busy_poll()233        data_path = get_nic_datapath(client)234        if NetworkDataPath.Sriov.value == data_path:235            server_nic_name = (236                server_nic_name if server_nic_name else server.nics.get_lower_nics()[0]237            )238            client_nic_name = (239                client_nic_name if client_nic_name else client.nics.get_lower_nics()[0]240            )241            dev_differentiator = "mlx"242        else:243            server_nic_name = (244                server_nic_name if server_nic_name else server.nics.default_nic245            )246            client_nic_name = (247                client_nic_name if client_nic_name else client.nics.default_nic...

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.