Antoni Nuñez Romeu
95 lines
3.1 KiB
Python
95 lines
3.1 KiB
Python
"""E2E tests for Rate Limiting middleware.
|
|
|
|
Tests that rate limiting returns 429 with proper headers
|
|
when request count exceeds the configured limit.
|
|
"""
|
|
|
|
import asyncio
|
|
import os
|
|
|
|
import httpx
|
|
|
|
BASE_URL = os.environ.get("PIP_BASE_URL", "http://localhost:8000")
|
|
ADMIN_CREDS = {"username": os.environ.get("PIP_ADMIN_USER", "synctest"), "password": os.environ.get("PIP_ADMIN_PASS", "Test123!")}
|
|
|
|
|
|
async def get_admin_token(client: httpx.AsyncClient) -> str:
|
|
r = await client.post("/api/v1/auth/token", json=ADMIN_CREDS)
|
|
assert r.status_code == 200
|
|
return r.json()["access_token"]
|
|
|
|
|
|
def bearer_headers(token: str) -> dict:
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
async def test_rate_limit_headers():
|
|
async with httpx.AsyncClient(base_url=BASE_URL) as c:
|
|
token = await get_admin_token(c)
|
|
h = bearer_headers(token)
|
|
|
|
r = await c.get("/api/v1/pharmacies", headers=h)
|
|
assert r.status_code == 200
|
|
assert "x-ratelimit-limit" in r.headers, f"Missing X-RateLimit-Limit header. Got: {list(r.headers.keys())}"
|
|
assert "x-ratelimit-remaining" in r.headers
|
|
assert "x-ratelimit-reset" in r.headers
|
|
|
|
limit = int(r.headers["x-ratelimit-limit"])
|
|
assert limit > 0
|
|
|
|
print(f" PASS: test_rate_limit_headers (limit={limit}/min)")
|
|
|
|
|
|
async def test_rate_limit_429():
|
|
async with httpx.AsyncClient(base_url=BASE_URL) as c:
|
|
token = await get_admin_token(c)
|
|
h = bearer_headers(token)
|
|
|
|
r0 = await c.get("/api/v1/pharmacies", headers=h)
|
|
limit = int(r0.headers["x-ratelimit-limit"])
|
|
|
|
successes = 0
|
|
blocked = 0
|
|
for i in range(limit + 10):
|
|
r = await c.get("/api/v1/pharmacies", headers=h)
|
|
if r.status_code == 200:
|
|
successes += 1
|
|
elif r.status_code == 429:
|
|
blocked += 1
|
|
assert "retry-after" in r.headers, "429 response missing Retry-After header"
|
|
break
|
|
|
|
assert blocked > 0, f"Rate limit was not triggered after {limit + 10} requests"
|
|
|
|
print(f" PASS: test_rate_limit_429 (blocked after {successes} requests)")
|
|
|
|
|
|
async def test_rate_limit_exempt_paths():
|
|
async with httpx.AsyncClient(base_url=BASE_URL) as c:
|
|
r = await c.get("/docs")
|
|
assert r.status_code == 200
|
|
assert "x-ratelimit-limit" not in r.headers, "Docs should be exempt from rate limiting"
|
|
|
|
print(" PASS: test_rate_limit_exempt_paths")
|
|
|
|
|
|
async def test_rate_limit_per_client():
|
|
async with httpx.AsyncClient(base_url=BASE_URL, timeout=30.0) as c:
|
|
r2 = await c.get("/api/v1/system/health")
|
|
assert r2.status_code == 200, "Health endpoint should work regardless of rate limit on other paths (it is exempt)"
|
|
|
|
print(" PASS: test_rate_limit_per_client")
|
|
|
|
|
|
async def run_all():
|
|
print("=== Rate Limiting E2E Tests ===")
|
|
await test_rate_limit_headers()
|
|
await test_rate_limit_429()
|
|
await test_rate_limit_exempt_paths()
|
|
await test_rate_limit_per_client()
|
|
print("=== All Rate Limiting tests passed ===")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(run_all())
|