| from __future__ import annotations |
|
|
| import hashlib |
| import uuid |
| from pathlib import Path |
|
|
| import pytest |
| from sqlalchemy import create_engine |
| from sqlalchemy.orm import sessionmaker |
| from sqlalchemy.pool import StaticPool |
|
|
| import persistence.db as persistence_db |
| from backend.app.storage import document_pdf_object_key |
| from persistence.db import Base |
| from persistence.document_ingestion import ( |
| DOCUMENT_STATUS_PROCESSING, |
| DOCUMENT_STATUS_QUEUED, |
| ) |
| from persistence.document_uploads import ( |
| UploadRequestConflictError, |
| create_or_recover_queued_upload, |
| get_owned_document_status, |
| recover_owned_upload_request, |
| should_enqueue_owned_document, |
| ) |
| from persistence.models import Document, User |
| from persistence.ownership import OwnedResourceNotFoundError |
|
|
|
|
| PDF_BYTES = b"%PDF-1.7 persistence upload" |
| SOURCE_HASH = hashlib.sha256(PDF_BYTES).hexdigest() |
| MIGRATION_PATH = ( |
| Path(__file__).resolve().parents[1] |
| / "migrations" |
| / "mysql" |
| / "005_document_upload_idempotency.sql" |
| ) |
|
|
|
|
| @pytest.fixture |
| def session_factory(monkeypatch): |
| engine = create_engine( |
| "sqlite+pysqlite:///:memory:", |
| connect_args={"check_same_thread": False}, |
| poolclass=StaticPool, |
| future=True, |
| ) |
| factory = sessionmaker( |
| bind=engine, |
| autoflush=False, |
| autocommit=False, |
| expire_on_commit=False, |
| future=True, |
| ) |
| Base.metadata.create_all(engine) |
| monkeypatch.setattr(persistence_db, "SessionLocal", factory) |
| try: |
| yield factory |
| finally: |
| Base.metadata.drop_all(engine) |
| engine.dispose() |
|
|
|
|
| def _add_user(session_factory) -> str: |
| user_id = str(uuid.uuid4()) |
| with session_factory() as session: |
| session.add(User(id=user_id, email=f"{user_id}@example.test")) |
| session.commit() |
| return user_id |
|
|
|
|
| def _create(session_factory, *, user_id: str, request_id: str, source_hash: str = SOURCE_HASH): |
| document_id = str(uuid.uuid4()) |
| return create_or_recover_queued_upload( |
| user_id=user_id, |
| document_id=document_id, |
| upload_request_id=request_id, |
| filename="report.pdf", |
| source_hash=source_hash, |
| object_key=document_pdf_object_key(document_id), |
| content_type="application/pdf", |
| byte_size=len(PDF_BYTES), |
| embedding_model="intfloat/e5-small-v2", |
| embedding_format="e5-query-passage-v1", |
| retrieval_mode="faiss_reranker", |
| reranker_model="cross-encoder/test", |
| k_initial=20, |
| k_final=8, |
| ) |
|
|
|
|
| def test_create_queued_upload_persists_owner_object_and_configuration(session_factory) -> None: |
| user_id = _add_user(session_factory) |
| request_id = str(uuid.uuid4()) |
|
|
| result = _create(session_factory, user_id=user_id, request_id=request_id) |
|
|
| assert result.created is True |
| assert result.document.user_id == user_id |
| assert result.document.status == DOCUMENT_STATUS_QUEUED |
| with session_factory() as session: |
| document = session.get(Document, result.document.document_id) |
| assert document.upload_request_id == request_id |
| assert document.object_key == document_pdf_object_key(document.id) |
| assert document.source_hash == SOURCE_HASH |
| assert document.content_type == "application/pdf" |
| assert document.byte_size == len(PDF_BYTES) |
| assert document.embedding_model == "intfloat/e5-small-v2" |
| assert document.embedding_format == "e5-query-passage-v1" |
| assert document.retrieval_mode == "faiss_reranker" |
| assert document.reranker_model == "cross-encoder/test" |
| assert document.k_initial == 20 |
| assert document.k_final == 8 |
|
|
|
|
| def test_upload_idempotency_migration_is_narrow_and_repeatable() -> None: |
| migration = MIGRATION_PATH.read_text(encoding="utf-8").lower() |
|
|
| assert "add column `upload_request_id` varchar(36) null" in migration |
| assert "(`user_id`, `upload_request_id`)" in migration |
| assert "uq_documents_user_upload_request_id" in migration |
| assert "information_schema.columns" in migration |
| assert "information_schema.statistics" in migration |
| assert "object_key" not in migration |
|
|
|
|
| def test_owned_request_recovery_and_hash_conflict(session_factory) -> None: |
| user_id = _add_user(session_factory) |
| request_id = str(uuid.uuid4()) |
| created = _create(session_factory, user_id=user_id, request_id=request_id) |
|
|
| recovered = recover_owned_upload_request( |
| user_id=user_id, |
| upload_request_id=request_id, |
| source_hash=SOURCE_HASH, |
| ) |
|
|
| assert recovered is not None |
| assert recovered.document_id == created.document.document_id |
| with pytest.raises(UploadRequestConflictError): |
| recover_owned_upload_request( |
| user_id=user_id, |
| upload_request_id=request_id, |
| source_hash="0" * 64, |
| ) |
|
|
|
|
| def test_unique_constraint_loser_recovers_one_authoritative_document(session_factory) -> None: |
| user_id = _add_user(session_factory) |
| request_id = str(uuid.uuid4()) |
| winner = _create(session_factory, user_id=user_id, request_id=request_id) |
|
|
| loser = _create(session_factory, user_id=user_id, request_id=request_id) |
|
|
| assert loser.created is False |
| assert loser.document.document_id == winner.document.document_id |
| with session_factory() as session: |
| rows = session.query(Document).filter_by( |
| user_id=user_id, |
| upload_request_id=request_id, |
| ).all() |
| assert [row.id for row in rows] == [winner.document.document_id] |
|
|
|
|
| def test_unique_constraint_loser_with_different_hash_conflicts(session_factory) -> None: |
| user_id = _add_user(session_factory) |
| request_id = str(uuid.uuid4()) |
| _create(session_factory, user_id=user_id, request_id=request_id) |
|
|
| with pytest.raises(UploadRequestConflictError): |
| _create( |
| session_factory, |
| user_id=user_id, |
| request_id=request_id, |
| source_hash="0" * 64, |
| ) |
|
|
|
|
| def test_upload_request_id_is_scoped_to_owner(session_factory) -> None: |
| first_user_id = _add_user(session_factory) |
| second_user_id = _add_user(session_factory) |
| request_id = str(uuid.uuid4()) |
|
|
| first = _create( |
| session_factory, |
| user_id=first_user_id, |
| request_id=request_id, |
| ) |
| second = _create( |
| session_factory, |
| user_id=second_user_id, |
| request_id=request_id, |
| ) |
|
|
| assert first.created is True |
| assert second.created is True |
| assert first.document.document_id != second.document.document_id |
|
|
|
|
| def test_status_lookup_enforces_ownership_and_preserves_current_state(session_factory) -> None: |
| owner_id = _add_user(session_factory) |
| other_id = _add_user(session_factory) |
| created = _create( |
| session_factory, |
| user_id=owner_id, |
| request_id=str(uuid.uuid4()), |
| ) |
| with session_factory() as session: |
| document = session.get(Document, created.document.document_id) |
| document.status = DOCUMENT_STATUS_PROCESSING |
| session.commit() |
|
|
| current = get_owned_document_status( |
| user_id=owner_id, |
| document_id=created.document.document_id, |
| ) |
|
|
| assert current.status == DOCUMENT_STATUS_PROCESSING |
| assert should_enqueue_owned_document( |
| user_id=owner_id, |
| document_id=created.document.document_id, |
| ).status == DOCUMENT_STATUS_PROCESSING |
| with pytest.raises(OwnedResourceNotFoundError): |
| get_owned_document_status( |
| user_id=other_id, |
| document_id=created.document.document_id, |
| ) |
|
|
