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

How to use _run_clients method in autotest

Best Python code snippet using autotest_python

stress_test.py

Source:stress_test.py

...122        with timer('Client invocation'):123            result = subprocess.run(cmd, stdout=PIPE, stderr=PIPE, input=stdin,124                                    universal_newlines=True, check=True)125        return Output(result.stdout)126def _run_clients(numof_clients, client, stdin):127    with Pool(numof_clients) as pool:128        results = pool.starmap(_run_client, [(client, stdin) for _ in range(numof_clients)])129        if len(results) != numof_clients:130            raise RuntimeError(f'expected {numof_clients} results, got {len(results)}')131        return results132def _run_stress_test(args):133    client = Client(args.client, args.host, args.port)134    input = Input.generate(args.numof_expressions)135    stdin = input.stdin()136    expected_output = input.expected_output()137    actual_outputs = _run_clients(args.numof_processes, client, stdin)138    # Check that all outputs are equal to each other:139    assert actual_outputs140    actual_output, rest = actual_outputs[0], actual_outputs[1:]141    for other in rest:142        if actual_output == other:143            continue144        logging.error("Client outputs don't match, this should never happen")145        logging.error('For example, this:\n%s', actual_output)146        logging.error('... is not equal to this:\n%s', other)147        return False148    # Check that the first output is equal to the expected output:149    return actual_output.equal_to_expected(expected_output)150def _parse_positive_int(s):151    try:...

run_clients.py

Source:run_clients.py

...23        return False24    finally:25        request_time = time.time() - start26        print(f"#{id_} Request time:{request_time:.2f}")27async def _run_clients(port: int, num_clients: int) -> None:28    clients = [_make_request(id_, port) for id_ in range(num_clients)]29    results = await asyncio.gather(*clients)30    counter = Counter(results)31    print(f"Successful:{counter[True]} Failed:{counter[False]}")32def main() -> None:33    args = _parse_args()34    asyncio.run(_run_clients(args.port, args.num_clients))35if __name__ == "__main__":...

multiple-client.py

Source:multiple-client.py

...4def _close_clients(processes):5    for proc in processes:6        proc.kill()7    return True8def _run_clients(count=1):9    process_list = []10    for _ in range(count):11        cmd = ["python", "client.py"] + sys.argv[1:]12        process_list.append(Popen(" ".join(cmd), shell=True))13    return process_list14if __name__ == "__main__":15    process_list = []16    while True:17        action = input(18            "(r)un new process / (c)lose all processes / (s)how clients / (q)uit\n"19        )20        if action == "r":21            process_list.extend(_run_clients())22            continue23        if action == "c":24            _close_clients(process_list)25            process_list = []26            continue27        if action == "s":28            for proc in process_list:29                print(f"pid={proc.pid}, args='{proc.args}'")30            continue31        if action == "q":...

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.