stress_test_concurrent()

None test_stress.stress_test_concurrent	(	StressTestConfig	config,
		StressTestResults	results )

Test: Multiple concurrent connections.

Definition at line 147 of file test_stress.py.

) -> 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})"
    )
 
 

Referenced by main().

Here is the caller graph for this function: