RAG/backend/tests/unit/test_documents_persistence.py

from __future__ import annotations

import uuid
from datetime import UTC, datetime
from pathlib import Path
from unittest.mock import MagicMock

import pytest

from app.core.config import Settings
from app.persistence.documents import (
    COMPLETE_UPLOAD_SQL,
    CREATE_UPLOAD_SQL,
    GET_JOB_SQL,
    GET_UPLOAD_SQL,
    LIST_DOCUMENTS_SQL,
    DocumentActor,
    IdempotencyConflictError,
    PostgresDocumentsRepository,
    idempotency_key_hash,
    upload_request_fingerprint,
)

NOW = datetime(2026, 7, 13, 8, 0, tzinfo=UTC)
ACTOR = DocumentActor(
    subject="synthetic-demo-maintainer",
    knowledge_base_id=uuid.UUID("10000000-0000-0000-0000-000000000001"),
    access_scope_id=uuid.UUID("20000000-0000-0000-0000-000000000002"),
)
UPLOAD_ID = uuid.UUID("30000000-0000-0000-0000-000000000003")
STORAGE_KEY = uuid.UUID("40000000-0000-0000-0000-000000000004")
TRACE_ID = uuid.UUID("50000000-0000-0000-0000-000000000005")
EXPECTED_HASH = "a" * 64


def _upload_row(*, fingerprint: str, created: bool = True) -> dict[str, object]:
    return {
        "id": UPLOAD_ID,
        "request_fingerprint": fingerprint,
        "original_filename": "synthetic.md",
        "declared_mime_type": "text/markdown",
        "expected_size": 128,
        "expected_sha256": EXPECTED_HASH,
        "storage_key": STORAGE_KEY,
        "actual_size": None,
        "actual_sha256": None,
        "status": "CREATED",
        "document_id": None,
        "parse_job_id": None,
        "created_at": NOW,
        "updated_at": NOW,
        "completed_at": None,
        "created": created,
    }


def _repository(
    tmp_path: Path, row: dict[str, object] | None
) -> tuple[PostgresDocumentsRepository, MagicMock, MagicMock]:
    password = tmp_path / "password"
    password.write_text("synthetic-password", encoding="utf-8")
    settings = Settings(postgres_password_file=password)
    cursor = MagicMock()
    cursor.fetchone.return_value = row
    connection = MagicMock()
    connection.__enter__.return_value = connection
    connection.execute.return_value = cursor
    factory = MagicMock(return_value=connection)
    return (
        PostgresDocumentsRepository(settings, connection_factory=factory),
        connection,
        factory,
    )


def test_create_upload_uses_short_transaction_actor_scope_and_hashed_idempotency(
    tmp_path: Path,
) -> None:
    key = uuid.UUID("60000000-0000-0000-0000-000000000006")
    key_hash = idempotency_key_hash(ACTOR, key)
    fingerprint = upload_request_fingerprint(
        filename="synthetic.md",
        declared_mime_type="text/markdown",
        expected_size=128,
        expected_sha256=EXPECTED_HASH,
    )
    repository, connection, factory = _repository(tmp_path, _upload_row(fingerprint=fingerprint))

    upload, created = repository.create_upload(
        actor=ACTOR,
        idempotency_key_hash=key_hash,
        request_fingerprint=fingerprint,
        filename="synthetic.md",
        declared_mime_type="text/markdown",
        expected_size=128,
        expected_sha256=EXPECTED_HASH,
        storage_key=STORAGE_KEY,
        trace_id=TRACE_ID,
    )

    assert created is True
    assert upload.id == UPLOAD_ID
    assert upload.storage_key == STORAGE_KEY
    assert len(key_hash) == 64
    assert str(key) not in key_hash
    factory.assert_called_once()
    connection.transaction.return_value.__enter__.assert_called_once_with()
    statement, parameters = connection.execute.call_args.args
    assert statement == CREATE_UPLOAD_SQL
    assert parameters[0:3] == (
        ACTOR.subject,
        ACTOR.knowledge_base_id,
        ACTOR.access_scope_id,
    )
    assert parameters[3] == key_hash
    assert parameters[-4:] == (
        ACTOR.subject,
        ACTOR.knowledge_base_id,
        ACTOR.access_scope_id,
        key_hash,
    )


def test_replayed_key_with_different_fingerprint_is_a_safe_conflict(tmp_path: Path) -> None:
    repository, _, _ = _repository(
        tmp_path,
        _upload_row(fingerprint="b" * 64, created=False),
    )

    with pytest.raises(IdempotencyConflictError) as captured:
        repository.create_upload(
            actor=ACTOR,
            idempotency_key_hash="c" * 64,
            request_fingerprint="d" * 64,
            filename="synthetic.md",
            declared_mime_type="text/markdown",
            expected_size=128,
            expected_sha256=EXPECTED_HASH,
            storage_key=STORAGE_KEY,
            trace_id=TRACE_ID,
        )

    assert "synthetic.md" not in str(captured.value)
    assert EXPECTED_HASH not in str(captured.value)


def test_all_public_reads_apply_server_actor_scope_and_job_projection_is_safe() -> None:
    normalized_upload = " ".join(GET_UPLOAD_SQL.lower().split())
    normalized_job = " ".join(GET_JOB_SQL.lower().split())
    normalized_list = " ".join(LIST_DOCUMENTS_SQL.lower().split())
    normalized_complete = " ".join(COMPLETE_UPLOAD_SQL.lower().split())

    for query in (normalized_upload, normalized_job):
        assert "actor_subject = %s" in query
        assert "knowledge_base_id = %s" in query
        assert "access_scope_id = %s" in query
    assert "document.knowledge_base_id = %s" in normalized_list
    assert "document.access_scope_id = %s" in normalized_list

    for forbidden in ("lease_owner", "lease_token", "lease_until", "payload"):
        assert forbidden not in normalized_job
    assert "upload.parse_job_id = job.id" in normalized_job
    assert "job.job_type = 'embed_document'" in normalized_job
    assert "version.document_id = upload.document_id" in normalized_job
    assert "'parse_document'" in normalized_complete
    assert "'document_parse'" in normalized_complete
    assert "on conflict (job_type, idempotency_key)" in normalized_complete
    assert "rag.documents.access_scope_id = excluded.access_scope_id" in normalized_complete
    assert (
        "storage_key" not in normalized_complete.split("jsonb_build_object", 1)[1].split("),", 1)[0]
    )