96 lines
2.4 KiB
Python
96 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from django.core.cache import cache
|
|
from django.test import RequestFactory
|
|
|
|
from apps.core.rate_limit import clear_rate_limit, is_rate_limited, register_rate_limit_failure
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear_rate_limit_cache():
|
|
cache.clear()
|
|
yield
|
|
cache.clear()
|
|
|
|
|
|
def _post_request(payload: dict[str, str], *, username: str | None = None):
|
|
factory = RequestFactory()
|
|
request = factory.post("/login/", data=payload)
|
|
request.META["REMOTE_ADDR"] = "203.0.113.10"
|
|
if username is None:
|
|
request.user = SimpleNamespace(is_authenticated=False, is_anonymous=True)
|
|
else:
|
|
request.user = SimpleNamespace(is_authenticated=True, pk=42)
|
|
payload["username"] = username
|
|
request.user.username = username
|
|
return request
|
|
|
|
|
|
def test_rate_limit_blocks_after_max_attempts_and_reports_block_state():
|
|
request = _post_request({"username": "alice"}, username="alice")
|
|
|
|
state = is_rate_limited(
|
|
request,
|
|
namespace="manual_import",
|
|
max_attempts=2,
|
|
window_seconds=300,
|
|
block_seconds=120,
|
|
)
|
|
assert not state.is_blocked
|
|
assert state.attempts == 0
|
|
|
|
state = register_rate_limit_failure(
|
|
request,
|
|
namespace="manual_import",
|
|
max_attempts=2,
|
|
window_seconds=300,
|
|
block_seconds=120,
|
|
)
|
|
assert not state.is_blocked
|
|
assert state.attempts == 1
|
|
|
|
state = register_rate_limit_failure(
|
|
request,
|
|
namespace="manual_import",
|
|
max_attempts=2,
|
|
window_seconds=300,
|
|
block_seconds=120,
|
|
)
|
|
assert state.is_blocked
|
|
assert state.remaining_seconds > 0
|
|
|
|
assert is_rate_limited(
|
|
request,
|
|
namespace="manual_import",
|
|
max_attempts=2,
|
|
window_seconds=300,
|
|
block_seconds=120,
|
|
).is_blocked
|
|
|
|
|
|
def test_rate_limit_clear_resets_state_for_authenticated_identity():
|
|
request = _post_request({"username": "alice"}, username="alice")
|
|
|
|
state = register_rate_limit_failure(
|
|
request,
|
|
namespace="login",
|
|
max_attempts=1,
|
|
window_seconds=300,
|
|
block_seconds=10,
|
|
)
|
|
assert state.is_blocked
|
|
|
|
clear_rate_limit(request, namespace="login")
|
|
state = is_rate_limited(
|
|
request,
|
|
namespace="login",
|
|
max_attempts=1,
|
|
window_seconds=300,
|
|
block_seconds=10,
|
|
)
|
|
assert not state.is_blocked
|
|
assert state.attempts == 0
|