Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use set_max_session method in lisa

Best Python code snippet using lisa_python

common.py

Source:common.py

...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_nic248            )249            dev_differentiator = "Hypervisor callback interrupts"250        server_lagscope.run_as_server(ip=server.internal_address)251        max_server_threads = 64252        perf_ntttcp_message_list: List[253            Union[NetworkTCPPerformanceMessage, NetworkUDPPerformanceMessage]254        ] = []255        for test_thread in connections:256            if test_thread < max_server_threads:257                num_threads_p = test_thread258                num_threads_n = 1259            else:260                num_threads_p = max_server_threads261                num_threads_n = int(test_thread / num_threads_p)262            if 1 == num_threads_n and 1 == num_threads_p:263                buffer_size = int(1048576 / 1024)264            else:265                buffer_size = int(65536 / 1024)266            if udp_mode:267                buffer_size = int(1024 / 1024)268            server_result = server_ntttcp.run_as_server_async(269                server_nic_name,270                ports_count=num_threads_p,271                buffer_size=buffer_size,272                dev_differentiator=dev_differentiator,273                udp_mode=udp_mode,274            )275            client_lagscope_process = client_lagscope.run_as_client_async(276                server_ip=server.internal_address,277                ping_count=0,278                run_time_seconds=10,279                print_histogram=False,280                print_percentile=False,281                histogram_1st_interval_start_value=0,282                length_of_histogram_intervals=0,283                count_of_histogram_intervals=0,284                dump_csv=False,285            )286            client_ntttcp_result = client_ntttcp.run_as_client(287                client_nic_name,288                server.internal_address,289                buffer_size=buffer_size,290                threads_count=num_threads_n,291                ports_count=num_threads_p,292                dev_differentiator=dev_differentiator,293                udp_mode=udp_mode,294            )295            server_ntttcp_result = server_result.wait_result()296            server_result_temp = server_ntttcp.create_ntttcp_result(297                server_ntttcp_result298            )299            client_result_temp = client_ntttcp.create_ntttcp_result(300                client_ntttcp_result, role="client"301            )302            client_sar_result = client_lagscope_process.wait_result()303            client_average_latency = client_lagscope.get_average(client_sar_result)304            if udp_mode:305                ntttcp_message: Union[306                    NetworkTCPPerformanceMessage, NetworkUDPPerformanceMessage307                ] = client_ntttcp.create_ntttcp_udp_performance_message(308                    server_result_temp,309                    client_result_temp,310                    str(test_thread),311                    buffer_size,312                    test_case_name,313                    test_result,314                )315            else:316                ntttcp_message = client_ntttcp.create_ntttcp_tcp_performance_message(317                    server_result_temp,318                    client_result_temp,319                    client_average_latency,320                    str(test_thread),321                    buffer_size,322                    test_case_name,323                    test_result,324                )325            notifier.notify(ntttcp_message)326            perf_ntttcp_message_list.append(ntttcp_message)327    finally:328        for ntttcp in [client_ntttcp, server_ntttcp]:329            ntttcp.restore_system(udp_mode)330        for lagscope in [client_lagscope, server_lagscope]:331            lagscope.restore_busy_poll()332    return perf_ntttcp_message_list333def perf_iperf(334    test_result: TestResult,335    connections: List[int],336    buffer_length_list: List[int],337    udp_mode: bool = False,338) -> None:339    environment = test_result.environment340    assert environment, "fail to get environment from testresult"341    client = cast(RemoteNode, environment.nodes[0])342    server = cast(RemoteNode, environment.nodes[1])343    client_iperf3 = client.tools[Iperf3]344    server_iperf3 = server.tools[Iperf3]345    test_case_name = inspect.stack()[1][3]346    iperf3_messages_list: List[Any] = []347    if udp_mode:348        for node in [client, server]:349            ssh = node.tools[Ssh]350            ssh.set_max_session()351            node.close()352    for buffer_length in buffer_length_list:353        for connection in connections:354            server_iperf3_process_list: List[Process] = []355            client_iperf3_process_list: List[Process] = []356            client_result_list: List[ExecutableResult] = []357            server_result_list: List[ExecutableResult] = []358            if connection < 64:359                num_threads_p = connection360                num_threads_n = 1361            else:362                num_threads_p = 64363                num_threads_n = int(connection / 64)364            server_start_port = 750...

ssh.py

Source:ssh.py

...50        if result and result[0]:51            return result[0]52        else:53            raise LisaException("not find sshd_config")54    def set_max_session(self, count: int = 200) -> None:55        config_path = self.get_default_sshd_config_path()56        sed = self.node.tools[Sed]57        sed.append(f"MaxSessions {count}", config_path, sudo=True)58        service = self.node.tools[Service]59        service.restart_service("sshd")60    def get(self, setting: str) -> str:61        config_path = self.get_default_sshd_config_path()62        settings = self.node.tools[Cat].read(config_path, True, True)63        if isinstance(self.node.os, Ubuntu):64            extra_sshd_config = "/etc/ssh/sshd_config.d/50-cloudimg-settings.conf"65            path_exist = self.node.execute(f"ls -lt {extra_sshd_config}", sudo=True)66            if path_exist.exit_code == 0:67                settings += self.node.tools[Cat].read(extra_sshd_config, True, True)68        pattern = re.compile(rf"^{setting}\s+(?P<value>.*)", re.M)...

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.