tests/test_session.py

"""
tests/test_session.py
─────────────────────
Unit tests for the session architecture.
"""

from __future__ import annotations

import tempfile
from datetime import datetime, timezone
from pathlib import Path

import pytest

from docling_pdf_processor.exceptions import SessionNotFoundError
from docling_pdf_processor.session.manager import SessionManager
from docling_pdf_processor.session.models import SessionStatus
from docling_pdf_processor.session.store import FileSystemSessionStore, InMemorySessionStore


def test_in_memory_store_create_and_get():
    store = InMemorySessionStore()
    session = store.create()
    assert session.status == SessionStatus.CREATED
    assert session.workspace.exists()

    fetched = store.get(session.session_id)
    assert fetched is not None
    assert fetched.session_id == session.session_id


def test_in_memory_store_cleanup_stale():
    store = InMemorySessionStore()
    session = store.create()
    # artificially age the session (do not call save, which would touch it)
    session.last_activity = datetime(2000, 1, 1, tzinfo=timezone.utc)

    removed = store.cleanup_stale(max_age_seconds=1)
    assert removed == 1
    assert store.get(session.session_id) is None


def test_file_system_store_persists_metadata():
    with tempfile.TemporaryDirectory() as tmp:
        base = Path(tmp) / "sessions"
        store = FileSystemSessionStore(base_dir=base)
        session = store.create()
        session.pdf_name = "demo.pdf"
        store.save(session)

        # fresh store instance should read from disk
        store2 = FileSystemSessionStore(base_dir=base)
        fetched = store2.get(session.session_id)
        assert fetched is not None
        assert fetched.pdf_name == "demo.pdf"


def test_session_manager_lifecycle():
    with tempfile.TemporaryDirectory() as tmp:
        base = Path(tmp) / "sessions"
        manager = SessionManager(store=InMemorySessionStore(base_dir=base))
        session = manager.create_session(pdf_name="test.pdf")
        assert session.pdf_name == "test.pdf"

        paths = manager.get_paths(session)
        assert (paths["pdf_dir"]).exists()

        manager.update_status(session.session_id, SessionStatus.READY)
        assert manager.get_session(session.session_id).status == SessionStatus.READY

        manager.cleanup_session(session.session_id)
        with pytest.raises(SessionNotFoundError):
            manager.get_session(session.session_id)