7f0775c8c3
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.
38 lines
1.2 KiB
Python
38 lines
1.2 KiB
Python
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
|