32 lines
1.1 KiB
Python
32 lines
1.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Test: find DPI threshold — how many WS frames before DPI kills connection."""
|
|
import asyncio, json, websockets, base64, os
|
|
|
|
async def test_chunks(n_chunks, delay=0.15):
|
|
uri = "wss://proxy.kube5s.ru/ws"
|
|
chunk_data = base64.b64encode(os.urandom(5000)).decode() # ~6.7KB per chunk
|
|
try:
|
|
async with websockets.connect(uri) as ws:
|
|
# Send data chunks
|
|
for i in range(n_chunks):
|
|
msg = json.dumps({"c": i, "n": n_chunks, "t": "test", "b": chunk_data, "m": "audio/webm"})
|
|
await ws.send(msg)
|
|
await asyncio.sleep(delay)
|
|
# Final
|
|
await ws.send(json.dumps({"c": -1}))
|
|
try:
|
|
resp = await asyncio.wait_for(ws.recv(), timeout=15)
|
|
data = json.loads(resp)
|
|
return f"OK: {data.get('error','?')}"
|
|
except asyncio.TimeoutError:
|
|
return "TIMEOUT"
|
|
except Exception as e:
|
|
return f"CONN_ERR: {e}"
|
|
|
|
async def main():
|
|
for n in [1, 2, 3, 4, 5, 8]:
|
|
result = await test_chunks(n, delay=0.2)
|
|
print(f" chunks={n}: {result}")
|
|
|
|
asyncio.run(main())
|