test(integration): add coverage for repositories, auth resolution, and the two-phase upload

tests/integration/minio/conftest.py

@@ -0,0 +1,39 @@
+from collections.abc import Iterator
+
+import pytest
+from testcontainers.community.minio import MinioContainer
+
+from src.config import MinioSettings
+from src.infrastructure.minio.client import create_client
+
+_BUCKET = "test-source-files"
+
+
+@pytest.fixture(scope="session")
+def minio_container() -> Iterator[MinioContainer]:
+    with MinioContainer() as container:
+        yield container
+
+
+@pytest.fixture(scope="session")
+def minio_settings(minio_container: MinioContainer) -> MinioSettings:
+    config = minio_container.get_config()
+    # Pinned to IPv4 for the same reason as postgres_url (see
+    # tests/integration/postgres/conftest.py): `localhost` resolves to `::1`
+    # first, but Docker only publishes the mapped port on IPv4, so the
+    # connection hangs instead of failing.
+    endpoint = config["endpoint"].replace("localhost:", "127.0.0.1:")
+    return MinioSettings(
+        endpoint=endpoint,
+        access_key=config["access_key"],
+        secret_key=config["secret_key"],
+        secure=False,
+        bucket=_BUCKET,
+    )
+
+
+@pytest.fixture(scope="session", autouse=True)
+def _ensure_bucket(minio_settings: MinioSettings) -> None:
+    client = create_client(minio_settings)
+    if not client.bucket_exists(minio_settings.bucket):
+        client.make_bucket(minio_settings.bucket)

tests/integration/minio/test_storage.py

@@ -0,0 +1,39 @@
+import uuid
+
+import pytest
+from anyio import CapacityLimiter
+
+from src.application.files.storage_keys import source_file_object_key
+from src.config import MinioSettings
+from src.infrastructure.minio.client import create_client
+from src.infrastructure.minio.storage import MinioObjectStorage
+
+pytestmark = [pytest.mark.integration, pytest.mark.minio, pytest.mark.asyncio]
+
+
+async def test_put_object_stores_bytes_under_server_derived_key(
+    minio_settings: MinioSettings,
+) -> None:
+    client = create_client(minio_settings)
+    storage = MinioObjectStorage(client, bucket=minio_settings.bucket, limiter=CapacityLimiter(2))
+    key = source_file_object_key(uuid.uuid4(), uuid.uuid4())
+    data = b"tenant-scoped, id-addressed bytes"
+
+    await storage.put_object(key=key, data=data, content_type="text/csv")
+
+    stored = client.get_object(minio_settings.bucket, key).read()
+    assert stored == data
+
+
+async def test_put_object_overwrites_existing_object_at_same_key(
+    minio_settings: MinioSettings,
+) -> None:
+    client = create_client(minio_settings)
+    storage = MinioObjectStorage(client, bucket=minio_settings.bucket, limiter=CapacityLimiter(2))
+    key = source_file_object_key(uuid.uuid4(), uuid.uuid4())
+
+    await storage.put_object(key=key, data=b"first", content_type="text/csv")
+    await storage.put_object(key=key, data=b"second", content_type="text/csv")
+
+    stored = client.get_object(minio_settings.bucket, key).read()
+    assert stored == b"second"

tests/integration/postgres/test_auth_service.py

@@ -0,0 +1,65 @@
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from src.application.auth.errors import InvalidApiKeyError, TenantInactiveError
+from src.application.auth.service import resolve_auth_context
+from tests.support.factories import create_api_key, create_tenant
+
+pytestmark = [
+    pytest.mark.integration,
+    pytest.mark.postgres,
+    pytest.mark.asyncio(loop_scope="session"),
+]
+
+
+async def test_resolve_auth_context_accepts_valid_key(
+    db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
+) -> None:
+    tenant = await create_tenant(db_session)
+    _, full_key = await create_api_key(db_session, tenant=tenant, scopes=["files:write"])
+    await db_session.commit()
+
+    auth = await resolve_auth_context(db_sessionmaker, full_key)
+
+    assert auth.tenant_id == tenant.id
+    assert auth.has_scope("files:write")
+
+
+async def test_resolve_auth_context_rejects_wrong_secret(
+    db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
+) -> None:
+    tenant = await create_tenant(db_session)
+    api_key, _ = await create_api_key(db_session, tenant=tenant)
+    await db_session.commit()
+
+    with pytest.raises(InvalidApiKeyError):
+        await resolve_auth_context(db_sessionmaker, f"sk_{api_key.key_prefix}_wrong-secret")
+
+
+async def test_resolve_auth_context_rejects_unknown_prefix(
+    db_sessionmaker: async_sessionmaker[AsyncSession],
+) -> None:
+    with pytest.raises(InvalidApiKeyError):
+        await resolve_auth_context(db_sessionmaker, "sk_doesnotexist_secret")
+
+
+async def test_resolve_auth_context_rejects_revoked_key(
+    db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
+) -> None:
+    tenant = await create_tenant(db_session)
+    _, full_key = await create_api_key(db_session, tenant=tenant, status="revoked")
+    await db_session.commit()
+
+    with pytest.raises(InvalidApiKeyError):
+        await resolve_auth_context(db_sessionmaker, full_key)
+
+
+async def test_resolve_auth_context_rejects_suspended_tenant(
+    db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
+) -> None:
+    tenant = await create_tenant(db_session, status="suspended")
+    _, full_key = await create_api_key(db_session, tenant=tenant)
+    await db_session.commit()
+
+    with pytest.raises(TenantInactiveError):
+        await resolve_auth_context(db_sessionmaker, full_key)

tests/integration/postgres/test_repositories.py

@@ -0,0 +1,113 @@
+import uuid
+
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.infrastructure.postgres.repositories import ingestion_jobs as jobs_repo
+from src.infrastructure.postgres.repositories import source_files as source_files_repo
+from tests.support.factories import create_tenant
+
+pytestmark = [
+    pytest.mark.integration,
+    pytest.mark.postgres,
+    pytest.mark.asyncio(loop_scope="session"),
+]
+
+
+async def test_source_files_get_by_id_is_tenant_scoped(db_session: AsyncSession) -> None:
+    owner = await create_tenant(db_session)
+    other = await create_tenant(db_session)
+    source_file_id = uuid.uuid4()
+    source_files_repo.create(
+        db_session,
+        source_file_id=source_file_id,
+        tenant_id=owner.id,
+        domain="general",
+        source_filename="report.csv",
+        source_type="csv",
+        content_sha256="a" * 64,
+        byte_size=10,
+        storage_uri="tenants/x/source-files/y/original",
+        created_by_api_key_id=None,
+    )
+    await db_session.commit()
+
+    found_for_owner = await source_files_repo.get_by_id(
+        db_session, tenant_id=owner.id, source_file_id=source_file_id
+    )
+    found_for_other = await source_files_repo.get_by_id(
+        db_session, tenant_id=other.id, source_file_id=source_file_id
+    )
+
+    assert found_for_owner is not None
+    assert found_for_other is None
+
+
+async def test_source_files_find_active_by_content_hash_matches_tenant_domain_hash(
+    db_session: AsyncSession,
+) -> None:
+    tenant = await create_tenant(db_session)
+    source_files_repo.create(
+        db_session,
+        source_file_id=uuid.uuid4(),
+        tenant_id=tenant.id,
+        domain="general",
+        source_filename="report.csv",
+        source_type="csv",
+        content_sha256="b" * 64,
+        byte_size=10,
+        storage_uri="tenants/x/source-files/y/original",
+        created_by_api_key_id=None,
+    )
+    await db_session.commit()
+
+    found = await source_files_repo.find_active_by_content_hash(
+        db_session, tenant_id=tenant.id, domain="general", content_sha256="b" * 64
+    )
+    not_found_other_domain = await source_files_repo.find_active_by_content_hash(
+        db_session, tenant_id=tenant.id, domain="other", content_sha256="b" * 64
+    )
+
+    assert found is not None
+    assert not_found_other_domain is None
+
+
+async def test_ingestion_jobs_mark_terminal_rejects_non_running_job(
+    db_session: AsyncSession,
+) -> None:
+    """A job already in a terminal state must not be re-marked (ADR-0017)."""
+    tenant = await create_tenant(db_session)
+    source_file_id = uuid.uuid4()
+    source_files_repo.create(
+        db_session,
+        source_file_id=source_file_id,
+        tenant_id=tenant.id,
+        domain="general",
+        source_filename="report.csv",
+        source_type="csv",
+        content_sha256="c" * 64,
+        byte_size=10,
+        storage_uri="tenants/x/source-files/y/original",
+        created_by_api_key_id=None,
+    )
+    await db_session.flush()
+    job = jobs_repo.create_running(
+        db_session,
+        tenant_id=tenant.id,
+        source_file_id=source_file_id,
+        requested_by_api_key_id=None,
+        chunking_strategy="fixed_size",
+    )
+    await db_session.commit()
+
+    first = await jobs_repo.mark_terminal(
+        db_session, tenant_id=tenant.id, ingestion_job_id=job.id, status="succeeded"
+    )
+    await db_session.commit()
+    second = await jobs_repo.mark_terminal(
+        db_session, tenant_id=tenant.id, ingestion_job_id=job.id, status="failed"
+    )
+
+    assert first is not None
+    assert first.status == "succeeded"
+    assert second is None

tests/integration/postgres/test_upload_service.py

@@ -0,0 +1,194 @@
+"""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