Ebbe BaßandClaude Sonnet 5.5
73505ac717
Audit entries written only inside the caller's transaction, database throttle counters that refuse to run inside a transaction that could roll back, and a database session store with the user key in its own column, end_sessions and idle and absolute limits. Ticket 1.3. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Sk72AxLwQnYttF6VFSxQLt
92 lines
2.7 KiB
Python
92 lines
2.7 KiB
Python
from datetime import UTC, datetime, timedelta
|
|
|
|
import pytest
|
|
from django.db import transaction
|
|
|
|
from portal.accounts import throttle
|
|
from portal.accounts.models import ThrottleCounter
|
|
|
|
T0 = datetime(2026, 1, 1, 12, 0, tzinfo=UTC)
|
|
EP = "sign-in"
|
|
|
|
pytestmark = pytest.mark.django_db(transaction=True)
|
|
|
|
|
|
def fail(n, at=T0, **kw):
|
|
for _ in range(n):
|
|
throttle.hit(EP, clock=lambda: at, **kw)
|
|
|
|
|
|
def test_below_limit_passes():
|
|
fail(4, user="alice")
|
|
throttle.check(EP, user="alice", clock=lambda: T0)
|
|
|
|
|
|
def test_user_is_throttled_but_others_pass():
|
|
fail(5, user="Alice", ip="1.1.1.1")
|
|
with pytest.raises(throttle.Throttled) as exc:
|
|
throttle.check(EP, user="alice", clock=lambda: T0)
|
|
assert exc.value.retry_after == 900
|
|
throttle.check(EP, user="bob", ip="2.2.2.2", clock=lambda: T0)
|
|
|
|
|
|
def test_ip_is_throttled_across_users():
|
|
fail(5, user="a", ip="1.1.1.1")
|
|
with pytest.raises(throttle.Throttled):
|
|
throttle.check(EP, user="other", ip="1.1.1.1", clock=lambda: T0)
|
|
throttle.check(EP, user="other", ip="3.3.3.3", clock=lambda: T0)
|
|
|
|
|
|
def test_other_endpoint_is_unaffected():
|
|
fail(5, user="alice")
|
|
throttle.check("password-change", user="alice", clock=lambda: T0)
|
|
|
|
|
|
def test_pause_elapses():
|
|
fail(5, user="alice")
|
|
later = T0 + timedelta(seconds=901)
|
|
throttle.check(EP, user="alice", clock=lambda: later)
|
|
# a new failure after the pause starts a fresh count
|
|
throttle.hit(EP, user="alice", clock=lambda: later)
|
|
assert ThrottleCounter.objects.get(subject="user:alice").count == 1
|
|
|
|
|
|
def test_stale_window_restarts_count():
|
|
fail(4, user="alice")
|
|
later = T0 + timedelta(minutes=16)
|
|
throttle.hit(EP, user="alice", clock=lambda: later)
|
|
assert ThrottleCounter.objects.get().count == 1
|
|
|
|
|
|
def test_reset_clears_user_but_never_the_ip():
|
|
fail(5, user="alice", ip="1.1.1.1")
|
|
throttle.reset(EP, user="alice")
|
|
throttle.check(EP, user="alice", clock=lambda: T0)
|
|
with pytest.raises(throttle.Throttled):
|
|
throttle.check(EP, ip="1.1.1.1", clock=lambda: T0)
|
|
|
|
|
|
def test_reset_has_no_ip_parameter():
|
|
with pytest.raises(TypeError):
|
|
throttle.reset(EP, user="alice", ip="1.1.1.1")
|
|
|
|
|
|
def test_hit_inside_transaction_raises_and_counts_nothing():
|
|
with pytest.raises(RuntimeError), transaction.atomic():
|
|
throttle.hit(EP, user="alice", clock=lambda: T0)
|
|
assert ThrottleCounter.objects.count() == 0
|
|
|
|
|
|
def test_no_subject_raises_for_all_functions():
|
|
with pytest.raises(ValueError):
|
|
throttle.check(EP)
|
|
with pytest.raises(ValueError):
|
|
throttle.hit(EP)
|
|
with pytest.raises(ValueError):
|
|
throttle.reset(EP, user="")
|
|
|
|
|
|
def test_check_never_writes():
|
|
throttle.check(EP, user="alice", ip="1.1.1.1", clock=lambda: T0)
|
|
assert ThrottleCounter.objects.count() == 0
|