98 lines
3.3 KiB
Python
98 lines
3.3 KiB
Python
from datetime import UTC, datetime, timedelta
|
|
|
|
from ai_platform.platform.application.admin_sessions import (
|
|
AdminSessionRecord,
|
|
AdminSessionUseCases,
|
|
AdminUserRecord,
|
|
)
|
|
from ai_platform.platform.security import digest
|
|
|
|
|
|
class FakeAdminSessionRepository:
|
|
def __init__(self) -> None:
|
|
self.user = AdminUserRecord("admin@example.invalid", "stored-hash")
|
|
self.sessions: dict[str, AdminSessionRecord] = {}
|
|
|
|
async def find_user(self, email: str) -> AdminUserRecord | None:
|
|
return self.user if email == self.user.email else None
|
|
|
|
async def find(self, session_id: str) -> AdminSessionRecord | None:
|
|
return self.sessions.get(session_id)
|
|
|
|
async def create(self, session: AdminSessionRecord) -> None:
|
|
self.sessions[session.id] = session
|
|
|
|
async def delete(self, session_id: str) -> None:
|
|
self.sessions.pop(session_id, None)
|
|
|
|
|
|
class Transaction:
|
|
def __init__(self) -> None:
|
|
self.commits = 0
|
|
self.rollbacks = 0
|
|
|
|
async def commit(self) -> None:
|
|
self.commits += 1
|
|
|
|
async def rollback(self) -> None:
|
|
self.rollbacks += 1
|
|
|
|
|
|
def make_use_cases(repository, transaction, tokens) -> AdminSessionUseCases:
|
|
return AdminSessionUseCases(
|
|
repository=repository,
|
|
verify_password=lambda password, stored: password == "correct" and stored == "stored-hash",
|
|
commit=transaction.commit,
|
|
rollback=transaction.rollback,
|
|
token_generator=lambda: next(tokens),
|
|
clock=lambda: datetime(2026, 9, 25, tzinfo=UTC),
|
|
)
|
|
|
|
|
|
async def test_admin_login_stores_only_token_digests_and_logout_commits():
|
|
repository = FakeAdminSessionRepository()
|
|
transaction = Transaction()
|
|
tokens = iter(("one-time-session-token", "one-time-csrf-token"))
|
|
use_cases = make_use_cases(repository, transaction, tokens)
|
|
|
|
login = await use_cases.login(" ADMIN@example.invalid ", "correct", 300)
|
|
assert login is not None
|
|
stored = repository.sessions[digest(login.session_token)]
|
|
assert stored.csrf_hash == digest(login.csrf_token)
|
|
assert login.session_token not in repository.sessions
|
|
assert transaction.commits == 1
|
|
|
|
authenticated = await use_cases.find(login.session_token)
|
|
assert authenticated == stored
|
|
await use_cases.logout(stored.id)
|
|
|
|
assert await use_cases.find(login.session_token) is None
|
|
assert transaction.commits == 2
|
|
assert transaction.rollbacks == 0
|
|
|
|
|
|
async def test_admin_login_rejects_invalid_credentials_without_mutating_session():
|
|
repository = FakeAdminSessionRepository()
|
|
transaction = Transaction()
|
|
use_cases = make_use_cases(repository, transaction, iter(("unused-1", "unused-2")))
|
|
|
|
assert await use_cases.login("admin@example.invalid", "wrong", 300) is None
|
|
assert repository.sessions == {}
|
|
assert transaction.commits == 0
|
|
assert transaction.rollbacks == 0
|
|
|
|
|
|
async def test_expired_admin_session_is_not_authenticated():
|
|
repository = FakeAdminSessionRepository()
|
|
now = datetime(2026, 9, 25, tzinfo=UTC)
|
|
repository.sessions["session-hash"] = AdminSessionRecord(
|
|
id="session-hash",
|
|
admin_email=repository.user.email,
|
|
csrf_hash="csrf-hash",
|
|
expires_at=now - timedelta(seconds=1),
|
|
)
|
|
transaction = Transaction()
|
|
use_cases = make_use_cases(repository, transaction, iter(()))
|
|
|
|
assert await use_cases.find("raw-session") is None
|