bot-backend
6ddc9bc661
Co-authored-by: bot-backend <bot-backend@gendsgn.local> Co-committed-by: bot-backend <bot-backend@gendsgn.local>
58 lines
2.3 KiB
Python
58 lines
2.3 KiB
Python
"""Tests for RateLimitMiddleware — per-IP sliding window + auth bypass (#655).
|
||
|
||
Тестируем middleware в изоляции на минимальном FastAPI-приложении, чтобы не
|
||
тянуть тяжёлый app.main (DB / scheduler / sentry). Лимит/окно читаются из
|
||
settings на каждый dispatch → monkeypatch'им их в маленькие значения.
|
||
"""
|
||
|
||
import os
|
||
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||
|
||
import pytest
|
||
from fastapi import FastAPI
|
||
from fastapi.testclient import TestClient
|
||
|
||
from app.core import config
|
||
from app.core.ratelimit import RateLimitMiddleware
|
||
|
||
|
||
@pytest.fixture
|
||
def client(monkeypatch):
|
||
"""Минимальное приложение с лимитом 3 запроса / 60 с (для быстрого 429)."""
|
||
monkeypatch.setattr(config.settings, "rate_limit", 3)
|
||
monkeypatch.setattr(config.settings, "rate_limit_window_s", 60.0)
|
||
|
||
app = FastAPI()
|
||
app.add_middleware(RateLimitMiddleware)
|
||
|
||
@app.get("/api/v1/ping")
|
||
def ping() -> dict[str, bool]:
|
||
return {"ok": True}
|
||
|
||
return TestClient(app)
|
||
|
||
|
||
def test_anonymous_throttled_past_limit(client):
|
||
# Первые rate_limit запросов проходят, следующий — 429.
|
||
for _ in range(3):
|
||
assert client.get("/api/v1/ping").status_code == 200
|
||
resp = client.get("/api/v1/ping")
|
||
assert resp.status_code == 429
|
||
assert resp.json() == {"detail": "Слишком много запросов. Попробуйте позже."}
|
||
assert "Retry-After" in resp.headers
|
||
|
||
|
||
def test_authenticated_user_bypasses_limit(client):
|
||
# X-Authenticated-User (Caddy basic_auth) → лимит не применяется.
|
||
headers = {"X-Authenticated-User": "pilot@example.com"}
|
||
for _ in range(10): # сильно больше rate_limit=3
|
||
assert client.get("/api/v1/ping", headers=headers).status_code == 200
|
||
|
||
|
||
def test_empty_auth_header_does_not_bypass(client):
|
||
# Пустой заголовок не должен no-op'ить лимит (header present но falsy).
|
||
headers = {"X-Authenticated-User": ""}
|
||
for _ in range(3):
|
||
assert client.get("/api/v1/ping", headers=headers).status_code == 200
|
||
assert client.get("/api/v1/ping", headers=headers).status_code == 429
|