zzstoatzz.io
audio streaming app
plyr.fm
1import asyncio
2import sys
3import time
4
5import httpx
6
7BASE_URL = "https://api-stg.plyr.fm"
8
9
10async def check_health(client):
11 print(f"Checking {BASE_URL}/health...")
12 try:
13 resp = await client.get(f"{BASE_URL}/health")
14 if resp.status_code == 200:
15 print("✅ Health check passed")
16 else:
17 print(f"❌ Health check failed: {resp.status_code}")
18 sys.exit(1)
19 except Exception as e:
20 print(f"❌ Connection failed: {e}")
21 sys.exit(1)
22
23
24async def verify_rate_limit(client):
25 print("\nVerifying rate limits (target: >100 reqs)...")
26 # Global limit is 100/min per instance.
27 # With N instances, capacity is N * 100.
28 count = 250
29 rate_limited = 0
30 success = 0
31
32 start = time.time()
33 # Send in batches
34 batch_size = 50
35 for _ in range(0, count, batch_size):
36 tasks = []
37 for _ in range(batch_size):
38 tasks.append(client.get(f"{BASE_URL}/health"))
39
40 responses = await asyncio.gather(*tasks, return_exceptions=True)
41
42 for r in responses:
43 if isinstance(r, httpx.Response):
44 if r.status_code == 429:
45 rate_limited += 1
46 elif r.status_code == 200:
47 success += 1
48 else:
49 print(f"Unexpected status: {r.status_code}")
50 else:
51 print(f"Request error: {r}")
52
53 if rate_limited > 0:
54 print(f"Hit limit after {success + rate_limited} requests!")
55 break
56
57 duration = time.time() - start
58 print(f"Summary: {success} OK, {rate_limited} Limited in {duration:.2f}s")
59
60 if rate_limited > 0:
61 print("✅ Rate limiting confirmed active")
62 else:
63 print("⚠️ No rate limits hit - capacity might be higher than tested")
64
65
66async def main():
67 async with httpx.AsyncClient(timeout=10.0) as client:
68 await check_health(client)
69 await verify_rate_limit(client)
70
71
72if __name__ == "__main__":
73 asyncio.run(main())