File size: 2,757 Bytes
2f073d3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 | """
Lightweight load test script for the API.
Runs concurrent requests against the /api/analyze endpoint.
Usage: python -m backend.tests.test_load [--url URL] [--concurrency N] [--requests N]
"""
import argparse
import asyncio
import time
import httpx
async def single_request(client: httpx.AsyncClient, url: str, text: str) -> dict:
start = time.time()
try:
resp = await client.post(
f"{url}/api/analyze",
json={"text": text},
timeout=30.0,
)
return {
"status": resp.status_code,
"latency_ms": int((time.time() - start) * 1000),
"success": resp.status_code == 200,
}
except Exception as e:
return {
"status": 0,
"latency_ms": int((time.time() - start) * 1000),
"success": False,
"error": str(e),
}
async def run_load_test(url: str, concurrency: int, total_requests: int):
text = (
"The Federal Reserve announced today that it will maintain current interest "
"rates through the end of the quarter, citing stable employment numbers."
)
results = []
semaphore = asyncio.Semaphore(concurrency)
async def bounded_request(client):
async with semaphore:
return await single_request(client, url, text)
start = time.time()
async with httpx.AsyncClient() as client:
tasks = [bounded_request(client) for _ in range(total_requests)]
results = await asyncio.gather(*tasks)
total_time = time.time() - start
successes = sum(1 for r in results if r["success"])
latencies = [r["latency_ms"] for r in results if r["success"]]
print(f"\n{'='*50}")
print(f"Load Test Results")
print(f"{'='*50}")
print(f"Total requests: {total_requests}")
print(f"Concurrency: {concurrency}")
print(f"Total time: {total_time:.2f}s")
print(f"Successes: {successes}/{total_requests}")
print(f"RPS: {total_requests/total_time:.1f}")
if latencies:
latencies.sort()
print(f"Avg latency: {sum(latencies)/len(latencies):.0f}ms")
print(f"P50 latency: {latencies[len(latencies)//2]}ms")
print(f"P95 latency: {latencies[int(len(latencies)*0.95)]}ms")
print(f"P99 latency: {latencies[int(len(latencies)*0.99)]}ms")
print(f"{'='*50}\n")
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--url", default="http://localhost:8000")
parser.add_argument("--concurrency", type=int, default=10)
parser.add_argument("--requests", type=int, default=50)
args = parser.parse_args()
asyncio.run(run_load_test(args.url, args.concurrency, args.requests))
|