3deaa5ef40
Complete Flask → FastAPI migration with: - FastAPI app with session auth, Azure AD SSO, rate limiting - SQLite-backed session store (survives restarts) - Bulk AI metadata generation with SSE progress - Admin panel (user management, audit log, AI usage) - Subpath deployment support (ROOT_PATH config) - Docker + deploy.sh for production deployment - Test suite (auth, upload, templates, imports, admin, sessions) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
95 lines
3.4 KiB
Python
95 lines
3.4 KiB
Python
"""Tests for the SQLite-backed session store."""
|
|
|
|
import tempfile
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.session.store import SessionStore
|
|
|
|
|
|
@pytest.fixture
|
|
def store():
|
|
"""Create a temporary session store."""
|
|
fd, path = tempfile.mkstemp(suffix=".db")
|
|
os.close(fd)
|
|
s = SessionStore(db_path=path)
|
|
yield s
|
|
os.unlink(path)
|
|
|
|
|
|
class TestFileSession:
|
|
def test_create_and_get(self, store):
|
|
"""Create and retrieve a file session."""
|
|
sid = store.create_file_session(user_id=1, metadata_source="manual")
|
|
assert sid
|
|
session = store.get_file_session(sid)
|
|
assert session is not None
|
|
assert session["user_id"] == 1
|
|
assert session["files"] == []
|
|
|
|
def test_add_file_to_session(self, store):
|
|
"""Add files to a session."""
|
|
sid = store.create_file_session(user_id=1)
|
|
store.add_file_to_session(sid, {"filename": "test.pdf", "success": True})
|
|
store.add_file_to_session(sid, {"filename": "img.jpg", "success": True})
|
|
|
|
session = store.get_file_session(sid)
|
|
assert len(session["files"]) == 2
|
|
assert session["files"][0]["filename"] == "test.pdf"
|
|
|
|
def test_update_file_in_session(self, store):
|
|
"""Update a specific file entry."""
|
|
sid = store.create_file_session(user_id=1)
|
|
store.add_file_to_session(sid, {"filename": "test.pdf", "status": "pending"})
|
|
store.update_file_in_session(sid, 0, {"status": "complete", "metadata": {"title": "T"}})
|
|
|
|
session = store.get_file_session(sid)
|
|
assert session["files"][0]["status"] == "complete"
|
|
assert session["files"][0]["metadata"]["title"] == "T"
|
|
|
|
def test_delete_session(self, store):
|
|
"""Delete a file session."""
|
|
sid = store.create_file_session(user_id=1)
|
|
store.delete_file_session(sid)
|
|
assert store.get_file_session(sid) is None
|
|
|
|
def test_session_id_is_secure(self, store):
|
|
"""Session IDs should be cryptographically random."""
|
|
ids = [store.create_file_session(user_id=1) for _ in range(5)]
|
|
assert len(set(ids)) == 5 # All unique
|
|
for sid in ids:
|
|
assert len(sid) > 20 # Long enough for security
|
|
|
|
|
|
class TestImportSession:
|
|
def test_create_import_session(self, store):
|
|
"""Create and retrieve an import session."""
|
|
sid = store.create_import_session(
|
|
user_id=1,
|
|
session_type="import",
|
|
file_info={"path": "/tmp/test.csv", "filename": "test.csv"},
|
|
)
|
|
session = store.get_import_session(sid)
|
|
assert session is not None
|
|
assert session["file_info"]["filename"] == "test.csv"
|
|
|
|
def test_update_import_metadata_map(self, store):
|
|
"""Update import session with metadata map."""
|
|
sid = store.create_import_session(user_id=1, session_type="import")
|
|
metadata_map = {"test": {"title": "Test Title", "subject": "Test Subject"}}
|
|
store.update_import_session(sid, metadata_map=metadata_map)
|
|
|
|
session = store.get_import_session(sid)
|
|
assert session["metadata_map"]["test"]["title"] == "Test Title"
|
|
|
|
|
|
class TestCleanup:
|
|
def test_cleanup_expired(self, store):
|
|
"""Cleanup removes expired sessions."""
|
|
# Create a session with 0 hours expiry (immediately expired)
|
|
sid = store.create_file_session(user_id=1, expires_hours=0)
|
|
count = store.cleanup_expired()
|
|
assert count >= 1
|
|
assert store.get_file_session(sid) is None
|