chatbot_v3/tests/integration/postgres/test_upload_service.py

"""ADR-0017's two-transaction upload shape, exercised against real Postgres.

Object storage is faked (`FakeObjectStorage`) -- it's a port, not the thing
under test here. `MinioObjectStorage` itself is covered in
`tests/integration/minio/`.
"""

import pytest
from anyio import CapacityLimiter
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from src.application.auth.context import AuthContext
from src.application.files.upload import upload_source_file
from src.infrastructure.postgres.models.ingestion_job import IngestionJob
from tests.fakes import FakeObjectStorage
from tests.support.factories import create_api_key, create_tenant

pytestmark = [
    pytest.mark.integration,
    pytest.mark.postgres,
    pytest.mark.asyncio(loop_scope="session"),
]

_CSV_BYTES = b"name,value\nfirst,1\n"


async def _auth_for(db_session: AsyncSession) -> AuthContext:
    tenant = await create_tenant(db_session)
    api_key, _ = await create_api_key(db_session, tenant=tenant)
    await db_session.commit()
    return AuthContext(
        tenant_id=tenant.id,
        tenant_slug=tenant.slug,
        api_key_id=api_key.id,
        scopes=frozenset({"files:write"}),
        actor_type="backend",
    )


async def test_upload_source_file_commits_running_job_before_storage_write(
    db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
) -> None:
    """After phase 1 commits, a *separate* session must already see the
    running job -- proving txn A committed and released before phase 2 work,
    per ADR-0017.
    """
    auth = await _auth_for(db_session)
    storage = FakeObjectStorage()

    result = await upload_source_file(
        sessionmaker=db_sessionmaker,
        storage=storage,
        auth=auth,
        domain="general",
        filename="report.csv",
        data=_CSV_BYTES,
        max_upload_size_bytes=1_000_000,
        chunking_strategy="fixed_size",
        validation_limiter=CapacityLimiter(2),
    )

    assert result.status == "succeeded"
    assert result.chunks_indexed == 0
    assert result.is_new_attempt

    async with db_sessionmaker() as verify_session:
        job = await verify_session.get(IngestionJob, result.ingestion_job_id)
    assert job is not None
    assert job.status == "succeeded"
    assert storage.objects  # bytes were actually written


async def test_upload_source_file_duplicate_hash_returns_existing_without_reingesting(
    db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
) -> None:
    auth = await _auth_for(db_session)
    storage = FakeObjectStorage()

    first = await upload_source_file(
        sessionmaker=db_sessionmaker,
        storage=storage,
        auth=auth,
        domain="general",
        filename="report.csv",
        data=_CSV_BYTES,
        max_upload_size_bytes=1_000_000,
        chunking_strategy="fixed_size",
        validation_limiter=CapacityLimiter(2),
    )
    second = await upload_source_file(
        sessionmaker=db_sessionmaker,
        storage=storage,
        auth=auth,
        domain="general",
        filename="report.csv",
        data=_CSV_BYTES,
        max_upload_size_bytes=1_000_000,
        chunking_strategy="fixed_size",
        validation_limiter=CapacityLimiter(2),
    )

    assert second.file_id == first.file_id
    assert second.ingestion_job_id == first.ingestion_job_id
    assert not second.is_new_attempt
    assert len(storage.objects) == 1  # no second write


async def test_upload_source_file_retries_after_failed_job(
    db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
) -> None:
    """A duplicate upload whose last attempt failed must get a fresh job,
    not be treated as already-succeeded (ADR-0017: safe to run more than
    once).
    """
    auth = await _auth_for(db_session)
    failing_storage = FakeObjectStorage(fail_next=True)

    with pytest.raises(OSError):
        await upload_source_file(
            sessionmaker=db_sessionmaker,
            storage=failing_storage,
            auth=auth,
            domain="general",
            filename="report.csv",
            data=_CSV_BYTES,
            max_upload_size_bytes=1_000_000,
            chunking_strategy="fixed_size",
            validation_limiter=CapacityLimiter(2),
        )

    async with db_sessionmaker() as verify_session:
        jobs = (
            (
                await verify_session.execute(
                    select(IngestionJob).where(IngestionJob.tenant_id == auth.tenant_id)
                )
            )
            .scalars()
            .all()
        )
    assert len(jobs) == 1
    assert jobs[0].status == "failed"

    retry = await upload_source_file(
        sessionmaker=db_sessionmaker,
        storage=FakeObjectStorage(),
        auth=auth,
        domain="general",
        filename="report.csv",
        data=_CSV_BYTES,
        max_upload_size_bytes=1_000_000,
        chunking_strategy="fixed_size",
        validation_limiter=CapacityLimiter(2),
    )

    assert retry.status == "succeeded"
    assert retry.is_new_attempt
    assert retry.file_id == jobs[0].source_file_id


async def test_upload_source_file_is_tenant_isolated_for_identical_content(
    db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
) -> None:
    """Two different tenants uploading byte-identical files must not collide
    on the (tenant_id, domain, content_sha256) idempotency key.
    """
    auth_a = await _auth_for(db_session)
    auth_b = await _auth_for(db_session)

    result_a = await upload_source_file(
        sessionmaker=db_sessionmaker,
        storage=FakeObjectStorage(),
        auth=auth_a,
        domain="general",
        filename="report.csv",
        data=_CSV_BYTES,
        max_upload_size_bytes=1_000_000,
        chunking_strategy="fixed_size",
        validation_limiter=CapacityLimiter(2),
    )
    result_b = await upload_source_file(
        sessionmaker=db_sessionmaker,
        storage=FakeObjectStorage(),
        auth=auth_b,
        domain="general",
        filename="report.csv",
        data=_CSV_BYTES,
        max_upload_size_bytes=1_000_000,
        chunking_strategy="fixed_size",
        validation_limiter=CapacityLimiter(2),
    )

    assert result_a.file_id != result_b.file_id