134 lines
5.2 KiB
Python
134 lines
5.2 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Quick performance test - 50 users, 100 requests each
|
||
"""
|
||
import asyncio
|
||
import aiohttp
|
||
import time
|
||
import random
|
||
import statistics
|
||
from typing import List, Dict
|
||
|
||
API_URL = "https://echoesoftheashgame.patacuack.net/api"
|
||
NUM_USERS = 50
|
||
REQUESTS_PER_USER = 100
|
||
|
||
async def test_user(session: aiohttp.ClientSession, user_num: int):
|
||
"""Simulate a single user making requests"""
|
||
username = f"loadtest_user_{user_num}"
|
||
password = "TestPassword123!"
|
||
|
||
# Login
|
||
async with session.post(f"{API_URL}/auth/login",
|
||
json={"username": username, "password": password}) as resp:
|
||
if resp.status != 200:
|
||
return []
|
||
data = await resp.json()
|
||
token = data["access_token"]
|
||
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
results = []
|
||
|
||
# Make requests
|
||
for _ in range(REQUESTS_PER_USER):
|
||
start = time.time()
|
||
action = random.choices(
|
||
["location", "inventory", "move"],
|
||
weights=[20, 30, 50] # 50% move, 30% inventory, 20% location
|
||
)[0]
|
||
|
||
success = False
|
||
try:
|
||
if action == "location":
|
||
async with session.get(f"{API_URL}/game/location", headers=headers, timeout=aiohttp.ClientTimeout(total=10)) as resp:
|
||
success = resp.status == 200
|
||
elif action == "inventory":
|
||
async with session.get(f"{API_URL}/game/inventory", headers=headers, timeout=aiohttp.ClientTimeout(total=10)) as resp:
|
||
success = resp.status == 200
|
||
elif action == "move":
|
||
# Get valid directions first
|
||
async with session.get(f"{API_URL}/game/location", headers=headers, timeout=aiohttp.ClientTimeout(total=10)) as resp:
|
||
if resp.status == 200:
|
||
loc_data = await resp.json()
|
||
directions = loc_data.get("directions", [])
|
||
if directions:
|
||
direction = random.choice(directions)
|
||
async with session.post(f"{API_URL}/game/move",
|
||
json={"direction": direction},
|
||
headers=headers,
|
||
timeout=aiohttp.ClientTimeout(total=10)) as move_resp:
|
||
success = move_resp.status == 200
|
||
except:
|
||
pass
|
||
|
||
elapsed = (time.time() - start) * 1000 # Convert to ms
|
||
results.append({
|
||
"action": action,
|
||
"success": success,
|
||
"time_ms": elapsed
|
||
})
|
||
|
||
await asyncio.sleep(0.001) # Small delay
|
||
|
||
return results
|
||
|
||
async def main():
|
||
print("\n" + "="*60)
|
||
print("QUICK PERFORMANCE TEST")
|
||
print(f"{NUM_USERS} users × {REQUESTS_PER_USER} requests = {NUM_USERS * REQUESTS_PER_USER} total")
|
||
print("="*60 + "\n")
|
||
|
||
async with aiohttp.ClientSession() as session:
|
||
start_time = time.time()
|
||
|
||
# Run all users concurrently
|
||
tasks = [test_user(session, i) for i in range(1, NUM_USERS + 1)]
|
||
all_results = await asyncio.gather(*tasks)
|
||
|
||
elapsed = time.time() - start_time
|
||
|
||
# Flatten results
|
||
results = [r for user_results in all_results for r in user_results]
|
||
|
||
# Calculate stats
|
||
total = len(results)
|
||
successful = sum(1 for r in results if r["success"])
|
||
failed = total - successful
|
||
success_rate = (successful / total * 100) if total > 0 else 0
|
||
rps = total / elapsed if elapsed > 0 else 0
|
||
|
||
success_times = [r["time_ms"] for r in results if r["success"]]
|
||
|
||
print("\n" + "="*60)
|
||
print("RESULTS")
|
||
print("="*60)
|
||
print(f"Total Requests: {total}")
|
||
print(f"Successful: {successful} ({success_rate:.1f}%)")
|
||
print(f"Failed: {failed} ({100-success_rate:.1f}%)")
|
||
print(f"Total Time: {elapsed:.2f}s")
|
||
print(f"Requests/Second: {rps:.2f}")
|
||
|
||
if success_times:
|
||
print(f"\nResponse Times (successful):")
|
||
print(f" Min: {min(success_times):.2f}ms")
|
||
print(f" Max: {max(success_times):.2f}ms")
|
||
print(f" Mean: {statistics.mean(success_times):.2f}ms")
|
||
print(f" Median: {statistics.median(success_times):.2f}ms")
|
||
print(f" 95th: {statistics.quantiles(success_times, n=20)[18]:.2f}ms")
|
||
|
||
# Breakdown by action
|
||
print(f"\nBreakdown:")
|
||
for action in ["move", "inventory", "location"]:
|
||
action_results = [r for r in results if r["action"] == action]
|
||
if action_results:
|
||
action_success = sum(1 for r in action_results if r["success"])
|
||
action_rate = (action_success / len(action_results) * 100)
|
||
action_times = [r["time_ms"] for r in action_results if r["success"]]
|
||
avg_time = statistics.mean(action_times) if action_times else 0
|
||
print(f" {action:12s}: {len(action_results):4d} req, {action_rate:5.1f}% success, {avg_time:6.2f}ms avg")
|
||
|
||
print("="*60 + "\n")
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|