"""Unit tests for cyclone.auth.sessions — Session create/validate/expire/touch.""" from __future__ import annotations from datetime import datetime, timedelta, timezone import pytest from sqlalchemy import delete from cyclone.auth import sessions, users from cyclone.db import Session as DbSession from cyclone.db import SessionLocal, User @pytest.fixture(autouse=True) def _clear(): with SessionLocal()() as db: db.execute(delete(DbSession)) db.execute(delete(User)) db.commit() yield with SessionLocal()() as db: db.execute(delete(DbSession)) db.execute(delete(User)) db.commit() def _make_user(): with SessionLocal()() as db: return users.create(db, username="sessuser", password="hunter2hunter2", role="user") def test_create_session_returns_id_and_session(): user = _make_user() with SessionLocal()() as db: sid, sess = sessions.create(db, user_id=user.id) assert len(sid) >= 32 assert sess.user_id == user.id assert sess.expires_at > datetime.now(timezone.utc) def test_get_valid_returns_session_for_active(): user = _make_user() with SessionLocal()() as db: sid, _ = sessions.create(db, user_id=user.id) with SessionLocal()() as db: got = sessions.get_valid(db, sid) assert got is not None assert got.user_id == user.id def test_get_valid_returns_none_for_missing(): with SessionLocal()() as db: assert sessions.get_valid(db, "does-not-exist") is None def test_get_valid_returns_none_for_expired(): user = _make_user() with SessionLocal()() as db: sid, _ = sessions.create(db, user_id=user.id) with SessionLocal()() as db: sess = sessions.get_valid(db, sid) sess.expires_at = datetime.now(timezone.utc) - timedelta(seconds=1) db.commit() with SessionLocal()() as db: assert sessions.get_valid(db, sid) is None def test_delete_removes_session(): user = _make_user() with SessionLocal()() as db: sid, _ = sessions.create(db, user_id=user.id) sessions.delete(db, sid) with SessionLocal()() as db: assert sessions.get_valid(db, sid) is None def test_touch_extends_expiry(): user = _make_user() with SessionLocal()() as db: sid, sess = sessions.create(db, user_id=user.id) original_expiry = sess.expires_at sessions.touch(db, sid) with SessionLocal()() as db: refreshed = sessions.get_valid(db, sid) assert refreshed.expires_at >= original_expiry