feat: periodically sweep expired auth sessions
auth_sessions rows were never deleted after expiry, only rejected on read. Adds a background sweep (every 30 min) in the app lifespan, plus a tested pure delete_expired_sessions() function.
This commit is contained in:
37
backend/tests/test_session_cleanup.py
Normal file
37
backend/tests/test_session_cleanup.py
Normal file
@@ -0,0 +1,37 @@
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.models.auth_session import AuthSession
|
||||
from app.models.user import User
|
||||
from app.security import hash_password
|
||||
from app.session_cleanup import delete_expired_sessions
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_expired_sessions_removes_only_expired(db_session):
|
||||
user = User(username="cleanup_medium", password_hash=hash_password("spookyspooky"))
|
||||
db_session.add(user)
|
||||
await db_session.flush()
|
||||
|
||||
expired = AuthSession(
|
||||
user_id=user.id,
|
||||
token_hash="expired-token-hash",
|
||||
expires_at=datetime.now(timezone.utc) - timedelta(days=1),
|
||||
)
|
||||
valid = AuthSession(
|
||||
user_id=user.id,
|
||||
token_hash="valid-token-hash",
|
||||
expires_at=datetime.now(timezone.utc) + timedelta(days=1),
|
||||
)
|
||||
db_session.add_all([expired, valid])
|
||||
await db_session.commit()
|
||||
|
||||
deleted_count = await delete_expired_sessions(db_session)
|
||||
assert deleted_count == 1
|
||||
|
||||
remaining = (await db_session.execute(select(AuthSession))).scalars().all()
|
||||
remaining_ids = {session.id for session in remaining}
|
||||
assert expired.id not in remaining_ids
|
||||
assert valid.id in remaining_ids
|
||||
Reference in New Issue
Block a user