Function stress_test_concurrent

examples/Asio/Server/test_stress.py:147–183

Test: Multiple concurrent connections.

(
    config: StressTestConfig, results: StressTestResults
)

def stress_test_concurrent(
    config: StressTestConfig, results: StressTestResults
) -> None:
    """Test: Multiple concurrent connections."""
    console.print("\n[bold cyan]Test 1: Concurrent Connections[/bold cyan]")
    console.print(f"Making {config.num_connections} concurrent requests...")

    with Progress(
        SpinnerColumn(),
        TextColumn("[progress.description]{task.description}"),
        BarColumn(),
        TaskProgressColumn(),
        console=console,
    ) as progress:
        task = progress.add_task("Requesting...", total=config.num_connections)

        threads: list[threading.Thread] = []
        with httpx.Client() as client:
            for _ in range(config.num_connections):
                thread: threading.Thread = threading.Thread(
                    target=make_request, args=(config, results, client)
                )
                thread.start()
                threads.append(thread)

            for thread in threads:
                thread.join()
                progress.update(task, advance=1)

    success_rate = (
        (results.successful_requests / results.total_requests * 100)
        if results.total_requests > 0
        else 0
    )
    console.print(
        f"  Success rate: {success_rate:.1f}% ({results.successful_requests}/{results.total_requests})"
    )


def stress_test_rapid(config: StressTestConfig, results: StressTestResults) -> None: