2136286275
tests/stress/live_accuracy.mjs: classroom-scale accuracy + latency test that targets the deployed server (single-session, sid=main). Logs in as admin via /admin/login, resets the session, joins N students serially over HTTP, opens N student WebSockets in batches of 8 (250ms apart) plus the instructor WS, then drives every question through the admin "next" command. Each student picks uniformly random A-D, sends the submit, waits for the submit_ack, and records the round-trip latency. After session_ended, the script verifies that every student whose pick == correct got score > 0, every other submission got score == 0, and reports p50/p95/p99 ack latency. First live run: 50 students, 100 submits, 100% acks, 100% accuracy match, p99 555ms (≈intercontinental RTT to HK). tests/stress/live_loop.sh: tmux-friendly loop that runs the live test every 60s and appends a JSONL summary line per cycle to runs/live_summary.jsonl. Mirrors the morning's api_stress run_loop shape so per-cycle aggregates are easy to scrape. app/rate_limit.py: tiny in-memory token bucket. Capacity + refill in tokens/minute, keyed by client IP via X-Forwarded-For (with a fallback to request.client.host). Process-local state — admin login is the only user. POST /admin/login: rate-limited at 10 attempts/minute/IP. Generous for the legit instructor (who succeeds in 1-2 tries) and prohibitive for brute force from a single attacker IP. Student endpoints deliberately NOT rate-limited because campus students share NAT gateways and IP-level limits would false-positive a whole class. The bucket is per-app-instance (instantiated inside the router factory), so test apps each get a fresh one and tests don't poison each other.
28 lines
1.1 KiB
Python
28 lines
1.1 KiB
Python
from app.rate_limit import TokenBucket
|
|
from conftest import admin_login
|
|
|
|
|
|
def test_token_bucket_allows_then_denies_then_refills():
|
|
bucket = TokenBucket(capacity=3, refill_per_minute=60) # 1 token/sec
|
|
assert bucket.take("ip1") is True
|
|
assert bucket.take("ip1") is True
|
|
assert bucket.take("ip1") is True
|
|
assert bucket.take("ip1") is False # exhausted
|
|
# Different key has its own bucket
|
|
assert bucket.take("ip2") is True
|
|
|
|
|
|
def test_admin_login_rate_limits_after_burst(client):
|
|
# Default config: 10 attempts/min/IP. Eleventh attempt should 429.
|
|
# Exhaust on wrong-password attempts so the test doesn't depend on
|
|
# the right password being unknown.
|
|
for _ in range(10):
|
|
response = client.post("/admin/login", json={"password": "wrong"})
|
|
assert response.status_code == 401
|
|
# Eleventh attempt: throttled
|
|
response = client.post("/admin/login", json={"password": "wrong"})
|
|
assert response.status_code == 429
|
|
# Even a correct password is throttled until the bucket refills.
|
|
response = client.post("/admin/login", json={"password": "admin-pass"})
|
|
assert response.status_code == 429
|