test(e2e): cover the ingestion slice against real Postgres, MinIO, and Qdrant

Why:
- the slice's reliability invariants (ADR-0016) had no end-to-end coverage.

Changes:
- 13 Testcontainers-based tests: duplicate upload, retry after a failed job,
  cross-tenant 404, unregistered domain, capacity 503, timeout 504, parse 400,
  real Qdrant 502, missing scope 403, and both readiness states
- only the dense embedders are faked (ADR-0016 bars live providers); they
  return the pinned 768/3072 dimensions
- the capacity test uses a committing sessionmaker, since the shared-connection
  fixture cannot serve concurrent sessions

Impact:
- runs in the default `uv run pytest`; needs Docker, like every integration test

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

tests/e2e/conftest.py

@@ -0,0 +1,244 @@
+"""The vertical slice against real Postgres, MinIO, and Qdrant (ADR-0016).
+
+These are the ADR-0016 end-to-end tests: "a small vertical-slice acceptance
+contract through the real application composition". They run on
+Testcontainers, in the default `uv run pytest` run, because ADR-0016 makes
+Testcontainers "the standard automated integration-test resource mechanism"
+and reserves Docker Compose for manual validation and the separate serialized
+smoke test of the *running web process* (`tests/e2e/test_compose_smoke.py`).
+
+What is real here: routing, auth, scopes, the domain allowlist, the ADR-0008
+error envelope, the two-transaction upload, MinIO object writes, BM25 sparse
+embedding, and Qdrant upserts into a per-test collection created by the
+production `ensure_chunks_collection`.
+
+What is faked, and only this: the two dense embedders. They are paid/remote
+network calls, and ADR-0016 forbids routine runs from touching a live
+provider. Their fakes return the pinned 768/3072 dimensions, because the real
+collection rejects anything else.
+"""
+
+import uuid
+from collections.abc import AsyncIterator, Callable, Coroutine
+from contextlib import AsyncExitStack
+from dataclasses import dataclass
+from typing import Any
+
+import pytest
+import pytest_asyncio
+from asgi_lifespan import LifespanManager
+from httpx import ASGITransport, AsyncClient
+from qdrant_client import AsyncQdrantClient, models
+from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
+
+from src.bootstrap.dependencies import get_dense_embedders, get_sessionmaker
+from src.config import MinioSettings, PostgresSettings, QdrantSettings, Settings
+from src.infrastructure.postgres.database import create_sessionmaker
+from src.infrastructure.qdrant.collection import (
+    DENSE_NOMIC_DIMENSIONS,
+    DENSE_OPENAI_DIMENSIONS,
+    ensure_chunks_collection,
+)
+from src.main import create_app
+from tests.fakes import FakeDenseEmbedder
+from tests.support.factories import create_api_key, create_tenant, create_tenant_domain
+
+CSV_BYTES = b"question,answer\nhow do I file a claim,call the branch\nwhat is covered,see policy\n"
+
+# Past validation's OOXML magic-byte check, so the failure lands in the DOCX
+# parser rather than in upload validation -- the branch this exercises.
+CORRUPT_DOCX_BYTES = b"PK\x03\x04" + b"not really a docx" * 8
+
+
+@dataclass
+class E2EStack:
+    """One configured app instance plus the handles a test asserts against."""
+
+    client: AsyncClient
+    collection: str
+    dense_embedders: list[FakeDenseEmbedder]
+
+
+StackFactory = Callable[..., Coroutine[Any, Any, E2EStack]]
+
+
+@dataclass(frozen=True)
+class Principal:
+    """A provisioned tenant with a bearer token and a registered domain.
+
+    Plain values, not the ORM `Tenant`: tests re-read rows the app committed on
+    a shared connection, and any expiry/refresh of a held ORM instance would
+    lazy-load outside a greenlet context (`MissingGreenlet`).
+    """
+
+    tenant_id: uuid.UUID
+    tenant_slug: str
+    token: str
+    domain: str
+
+    @property
+    def headers(self) -> dict[str, str]:
+        return {"Authorization": f"Bearer {self.token}"}
+
+
+@pytest.fixture
+def e2e_settings(
+    postgres_settings: PostgresSettings,
+    minio_settings: MinioSettings,
+    qdrant_settings: QdrantSettings,
+) -> Settings:
+    """Settings wired to this test's containers.
+
+    Every nested settings object is passed explicitly. `src/config.py` cascades
+    `.env` into each nested class, so omitting one would let a developer's real
+    `EMBEDDING_NOMIC_BASE_URL` (a colleague's Ollama box) into a routine test
+    run. The embedder URLs below point at a closed port for the same reason:
+    the lifespan warms the *real* embedders before the fakes are substituted,
+    and connection-refused is both instant and free.
+    """
+    return Settings(
+        postgres=postgres_settings,
+        minio=minio_settings,
+        qdrant=qdrant_settings,
+        embedding={
+            "nomic": {"base_url": "http://127.0.0.1:1/v1", "timeout_seconds": 0.5},
+            "openai": {"base_url": "http://127.0.0.1:1/v1", "timeout_seconds": 0.5},
+        },
+        app={"readiness_check_timeout_seconds": 2.0},
+    )
+
+
+@pytest_asyncio.fixture(loop_scope="session")
+async def make_stack(
+    e2e_settings: Settings,
+    db_sessionmaker: async_sessionmaker[AsyncSession],
+    qdrant_client: AsyncQdrantClient,
+) -> AsyncIterator[StackFactory]:
+    """Build an app under test, optionally with tightened ingestion bounds.
+
+    A factory rather than a fixture because the capacity and timeout tests need
+    their own `INGESTION_*` values, and those are read at app construction.
+    """
+    async with AsyncExitStack() as exit_stack:
+
+        async def _make(
+            *,
+            ingestion: dict[str, object] | None = None,
+            collection: str | None = None,
+            dense_delay_seconds: float = 0.0,
+            bootstrap_collection: bool = True,
+            sessionmaker: async_sessionmaker[AsyncSession] | None = None,
+        ) -> E2EStack:
+            resolved_collection = collection or e2e_settings.qdrant.collection
+            if bootstrap_collection:
+                await ensure_chunks_collection(qdrant_client, collection=resolved_collection)
+
+            settings = e2e_settings.model_copy(
+                update={
+                    "qdrant": e2e_settings.qdrant.model_copy(
+                        update={"collection": resolved_collection}
+                    ),
+                    "ingestion": e2e_settings.ingestion.model_copy(update=ingestion or {}),
+                }
+            )
+
+            dense = [
+                FakeDenseEmbedder(
+                    name="dense_nomic",
+                    dimensions=DENSE_NOMIC_DIMENSIONS,
+                    value=0.1,
+                    delay_seconds=dense_delay_seconds,
+                ),
+                FakeDenseEmbedder(
+                    name="dense_openai",
+                    dimensions=DENSE_OPENAI_DIMENSIONS,
+                    value=0.2,
+                    delay_seconds=dense_delay_seconds,
+                ),
+            ]
+
+            app = create_app(settings)
+            # The session factory is the test's, so factory-created tenants are
+            # visible to the app and every write rolls back at teardown. Object
+            # storage, point storage, and the sparse embedder stay real.
+            resolved_sessionmaker = sessionmaker or db_sessionmaker
+            app.dependency_overrides[get_sessionmaker] = lambda: resolved_sessionmaker
+            app.dependency_overrides[get_dense_embedders] = lambda: dense
+
+            manager = await exit_stack.enter_async_context(LifespanManager(app))
+            client = await exit_stack.enter_async_context(
+                AsyncClient(
+                    transport=ASGITransport(app=manager.app),
+                    base_url="http://test",
+                    timeout=30.0,
+                )
+            )
+            return E2EStack(client=client, collection=resolved_collection, dense_embedders=dense)
+
+        yield _make
+
+
+@pytest_asyncio.fixture(loop_scope="session")
+async def committed_sessionmaker(
+    postgres_engine: AsyncEngine,
+) -> async_sessionmaker[AsyncSession]:
+    """A session factory whose writes really commit, one connection each.
+
+    The default `db_sessionmaker` pins every session to a single connection
+    inside one rolled-back transaction, which is what isolates a test's writes
+    -- but savepoints on a shared connection cannot interleave, so any test
+    that issues genuinely *concurrent* requests deadlocks or fails with
+    `InvalidSavepointSpecificationError`. Those tests use this instead and rely
+    on their uuid-keyed tenant for isolation (ADR-0016 allows either).
+    """
+    return create_sessionmaker(postgres_engine)
+
+
+@pytest_asyncio.fixture(loop_scope="session")
+async def principal(db_session: AsyncSession) -> Principal:
+    """A tenant with an upload-scoped key and one registered domain.
+
+    Registering the domain is not optional: `POST /v1/files` rejects an
+    unregistered one with `400 unknown_domain` before anything else runs
+    (ADR-0009).
+    """
+    return await provision_principal(db_session)
+
+
+async def provision_principal(
+    session: AsyncSession, *, domain: str = "fire", scopes: list[str] | None = None
+) -> Principal:
+    tenant = await create_tenant(session)
+    _, token = await create_api_key(session, tenant=tenant, scopes=scopes or ["files:write"])
+    await create_tenant_domain(session, tenant=tenant, domain=domain)
+    await session.commit()
+    return Principal(tenant_id=tenant.id, tenant_slug=tenant.slug, token=token, domain=domain)
+
+
+def upload_payload(
+    *, domain: str, filename: str = "faq.csv", data: bytes = CSV_BYTES
+) -> dict[str, Any]:
+    return {"files": {"file": (filename, data, "text/csv")}, "data": {"domain": domain}}
+
+
+async def count_active_points(
+    client: AsyncQdrantClient, collection: str, *, tenant_id: uuid.UUID
+) -> int:
+    """Active points for one tenant, read through the raw client.
+
+    `PointStorage` is deliberately write-only (point reads are plan 002's
+    `/v1/points`), so the verification read here uses the SDK directly.
+    """
+    result = await client.count(
+        collection_name=collection,
+        count_filter=models.Filter(
+            must=[
+                models.FieldCondition(
+                    key="tenant_id", match=models.MatchValue(value=str(tenant_id))
+                ),
+                models.FieldCondition(key="is_active", match=models.MatchValue(value=True)),
+            ]
+        ),
+        exact=True,
+    )
+    return result.count

tests/e2e/test_ingestion_slice.py

@@ -0,0 +1,329 @@
+"""The ingestion vertical slice end to end (plan 001, Phase 6).
+
+Every test here asserts a reliability invariant ADR-0016 lists as a reusable
+contract, not a happy path: the bound maps to its status code, the failure
+still writes a terminal job row, the retry converges, and the tenant filter
+holds. Failures additionally assert the ADR-0008 error envelope, because the
+status code alone is not the contract a client integrates against.
+"""
+
+import asyncio
+import uuid
+
+import pytest
+from httpx import Response
+from qdrant_client import AsyncQdrantClient
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from src.infrastructure.postgres.models.ingestion_job import IngestionJob
+from tests.e2e.conftest import (
+    CORRUPT_DOCX_BYTES,
+    Principal,
+    StackFactory,
+    count_active_points,
+    provision_principal,
+    upload_payload,
+)
+
+pytestmark = [
+    pytest.mark.e2e,
+    pytest.mark.postgres,
+    pytest.mark.minio,
+    pytest.mark.qdrant,
+    pytest.mark.asyncio(loop_scope="session"),
+]
+
+
+async def _latest_job(session: AsyncSession, *, tenant_id: uuid.UUID) -> IngestionJob:
+    """The tenant's most recent job row, read fresh.
+
+    `populate_existing` matters: this session shares a connection with the
+    app's sessions, so without it the identity map can answer from a row
+    loaded before the request under test committed its update.
+    """
+    result = await session.execute(
+        select(IngestionJob)
+        .where(IngestionJob.tenant_id == tenant_id)
+        .order_by(IngestionJob.created_at.desc(), IngestionJob.id)
+        .execution_options(populate_existing=True)
+    )
+    return result.scalars().first() or pytest.fail("no ingestion job was written")
+
+
+def _envelope(response: Response) -> dict[str, object]:
+    """The ADR-0008 error body. Asserting the code here, not just the HTTP
+    status, is the difference between testing the contract and testing the
+    number."""
+    error: dict[str, object] = response.json()["error"]
+    assert error["request_id"], "every error envelope carries its correlation id"
+    return error
+
+
+async def test_upload_succeeds_and_indexes_points_under_the_tenant_filter(
+    make_stack: StackFactory, principal: Principal, qdrant_client: AsyncQdrantClient
+) -> None:
+    stack = await make_stack()
+
+    response = await stack.client.post(
+        "/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
+    )
+
+    assert response.status_code == 201
+    body = response.json()
+    assert body["status"] == "succeeded"
+    assert body["chunks_indexed"] > 0
+    indexed = await count_active_points(
+        qdrant_client, stack.collection, tenant_id=principal.tenant_id
+    )
+    assert indexed == body["chunks_indexed"]
+
+
+async def test_get_file_reports_the_terminal_status_after_upload(
+    make_stack: StackFactory, principal: Principal
+) -> None:
+    stack = await make_stack()
+    created = await stack.client.post(
+        "/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
+    )
+
+    status = await stack.client.get(
+        f"/v1/files/{created.json()['file_id']}", headers=principal.headers
+    )
+
+    assert status.status_code == 200
+    assert status.json()["ingestion_status"] == "succeeded"
+    assert status.json()["domain"] == principal.domain
+
+
+async def test_upload_duplicate_content_returns_the_existing_job_without_reingesting(
+    make_stack: StackFactory, principal: Principal, qdrant_client: AsyncQdrantClient
+) -> None:
+    stack = await make_stack()
+    first = await stack.client.post(
+        "/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
+    )
+    points_after_first = await count_active_points(
+        qdrant_client, stack.collection, tenant_id=principal.tenant_id
+    )
+
+    second = await stack.client.post(
+        "/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
+    )
+
+    assert first.status_code == 201
+    # 200, not 201: nothing was created the second time (ADR-0017 idempotency).
+    assert second.status_code == 200
+    assert second.json()["file_id"] == first.json()["file_id"]
+    assert second.json()["ingestion_job_id"] == first.json()["ingestion_job_id"]
+    assert (
+        await count_active_points(qdrant_client, stack.collection, tenant_id=principal.tenant_id)
+        == points_after_first
+    )
+
+
+async def test_upload_retried_after_a_failed_job_succeeds_without_duplicate_points(
+    make_stack: StackFactory,
+    principal: Principal,
+    db_session: AsyncSession,
+    qdrant_client: AsyncQdrantClient,
+) -> None:
+    """ADR-0017, "Re-running an ingestion stays safe": a failed attempt is
+    retried by re-uploading the same bytes, and deterministic point ids make
+    the retry converge instead of duplicating.
+    """
+    stack = await make_stack()
+    stack.dense_embedders[0].fail_next = True
+
+    failed = await stack.client.post(
+        "/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
+    )
+    failed_job = await _latest_job(db_session, tenant_id=principal.tenant_id)
+    retried = await stack.client.post(
+        "/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
+    )
+
+    assert failed.status_code == 502
+    assert failed_job.status == "failed"
+    assert failed_job.error_code == "embedding_failed"
+    assert retried.status_code == 201
+    assert retried.json()["status"] == "succeeded"
+    # Same source file, new attempt -- a terminal job is never reused or
+    # transitioned back to running.
+    assert retried.json()["ingestion_job_id"] != str(failed_job.id)
+    indexed = await count_active_points(
+        qdrant_client, stack.collection, tenant_id=principal.tenant_id
+    )
+    assert indexed == retried.json()["chunks_indexed"]
+
+
+async def test_get_file_for_another_tenants_file_returns_404_not_403(
+    make_stack: StackFactory, principal: Principal, db_session: AsyncSession
+) -> None:
+    """404, never 403: a 403 would confirm the file exists to a stranger."""
+    stack = await make_stack()
+    created = await stack.client.post(
+        "/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
+    )
+    other = await provision_principal(db_session)
+
+    response = await stack.client.get(
+        f"/v1/files/{created.json()['file_id']}", headers=other.headers
+    )
+
+    assert response.status_code == 404
+    assert _envelope(response)["code"] == "not_found"
+
+
+async def test_upload_to_an_unregistered_domain_is_rejected_before_any_job_is_written(
+    make_stack: StackFactory, principal: Principal, db_session: AsyncSession
+) -> None:
+    stack = await make_stack()
+
+    response = await stack.client.post(
+        "/v1/files", **upload_payload(domain="never-registered"), headers=principal.headers
+    )
+
+    assert response.status_code == 400
+    assert _envelope(response)["code"] == "unknown_domain"
+    jobs = await db_session.execute(
+        select(IngestionJob).where(IngestionJob.tenant_id == principal.tenant_id)
+    )
+    assert jobs.scalars().all() == []
+
+
+async def test_upload_at_capacity_is_rejected_with_503_and_retry_after(
+    make_stack: StackFactory, committed_sessionmaker: async_sessionmaker[AsyncSession]
+) -> None:
+    """ADR-0017 rejects rather than queues, so the second concurrent upload
+    must fail fast with a backoff hint instead of waiting for a slot.
+
+    This is the one test that needs two requests genuinely in flight at once,
+    so it runs on `committed_sessionmaker` -- see that fixture for why the
+    shared-connection default cannot serve concurrent sessions.
+    """
+    async with committed_sessionmaker() as session:
+        principal = await provision_principal(session)
+    stack = await make_stack(
+        ingestion={"max_concurrency": 1},
+        dense_delay_seconds=0.4,
+        sessionmaker=committed_sessionmaker,
+    )
+
+    first, second = await asyncio.gather(
+        stack.client.post(
+            "/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
+        ),
+        stack.client.post(
+            "/v1/files",
+            **upload_payload(domain=principal.domain, filename="other.csv", data=b"a,b\n1,2\n"),
+            headers=principal.headers,
+        ),
+    )
+
+    statuses = sorted([first.status_code, second.status_code])
+    assert statuses == [201, 503]
+    rejected = first if first.status_code == 503 else second
+    assert _envelope(rejected)["code"] == "ingestion_at_capacity"
+    assert rejected.headers["Retry-After"] == "1"
+
+
+async def test_upload_exceeding_the_timeout_returns_504_and_writes_a_terminal_job(
+    make_stack: StackFactory, principal: Principal, db_session: AsyncSession
+) -> None:
+    """A timeout must never leave a job stuck in `running` (ADR-0016)."""
+    stack = await make_stack(ingestion={"timeout_seconds": 0.05}, dense_delay_seconds=0.5)
+
+    response = await stack.client.post(
+        "/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
+    )
+
+    assert response.status_code == 504
+    assert _envelope(response)["code"] == "ingestion_timeout"
+    job = await _latest_job(db_session, tenant_id=principal.tenant_id)
+    assert job.status == "failed"
+    assert job.error_code == "timeout"
+
+
+async def test_upload_of_an_unparseable_document_returns_400_and_a_failed_job(
+    make_stack: StackFactory, principal: Principal, db_session: AsyncSession
+) -> None:
+    stack = await make_stack()
+
+    response = await stack.client.post(
+        "/v1/files",
+        **upload_payload(domain=principal.domain, filename="broken.docx", data=CORRUPT_DOCX_BYTES),
+        headers=principal.headers,
+    )
+
+    assert response.status_code == 400
+    assert _envelope(response)["code"] == "validation_error"
+    job = await _latest_job(db_session, tenant_id=principal.tenant_id)
+    assert job.status == "failed"
+    assert job.error_code == "parse_failed"
+
+
+async def test_upload_returns_502_when_qdrant_indexing_fails(
+    make_stack: StackFactory, principal: Principal, db_session: AsyncSession
+) -> None:
+    """A real Qdrant error, not a fake: the collection the deployment step
+    should have created is missing, which is the failure an operator actually
+    hits when `qdrant_bootstrap` was skipped.
+    """
+    stack = await make_stack(
+        collection=f"never_bootstrapped_{uuid.uuid4().hex}", bootstrap_collection=False
+    )
+
+    response = await stack.client.post(
+        "/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
+    )
+
+    assert response.status_code == 502
+    assert _envelope(response)["code"] == "index_error"
+    job = await _latest_job(db_session, tenant_id=principal.tenant_id)
+    assert job.status == "failed"
+    assert job.error_code == "index_failed"
+
+
+async def test_upload_without_the_files_write_scope_is_forbidden(
+    make_stack: StackFactory, db_session: AsyncSession
+) -> None:
+    stack = await make_stack()
+    reader = await provision_principal(db_session, scopes=["domains:read"])
+
+    response = await stack.client.post(
+        "/v1/files", **upload_payload(domain=reader.domain), headers=reader.headers
+    )
+
+    assert response.status_code == 403
+    assert _envelope(response)["code"] == "missing_scope"
+
+
+async def test_readyz_reports_every_dependency_ready_against_real_services(
+    make_stack: StackFactory,
+) -> None:
+    """`/readyz` is dependency readiness, `/healthz` is process health."""
+    stack = await make_stack()
+
+    ready = await stack.client.get("/readyz")
+    healthy = await stack.client.get("/healthz")
+
+    assert healthy.status_code == 200
+    assert ready.status_code == 200
+    assert ready.json() == {"postgres": True, "minio": True, "qdrant": True}
+
+
+async def test_readyz_is_unready_when_the_chunks_collection_is_missing(
+    make_stack: StackFactory,
+) -> None:
+    """A reachable but unbootstrapped Qdrant is deliberately not ready --
+    uploads to it would 502 (see the indexing test above).
+    """
+    stack = await make_stack(
+        collection=f"never_bootstrapped_{uuid.uuid4().hex}", bootstrap_collection=False
+    )
+
+    response = await stack.client.get("/readyz")
+
+    assert response.status_code == 503
+    assert response.json()["qdrant"] is False

tests/fakes.py

@@ -32,6 +32,9 @@ class FakeDenseEmbedder:
 
     name: str
     dimensions: int = 4
+    value: float = 0.0
+    """Component value of every returned vector. Non-zero where a real Qdrant
+    has to score the result, since a zero vector has no direction to compare."""
     model_version: str = "fake-dense-v1"
     calls: list[list[str]] = field(default_factory=list)
     fail_next: bool = False
@@ -45,7 +48,7 @@ class FakeDenseEmbedder:
         if self.fail_next:
             self.fail_next = False
             raise RuntimeError("simulated embedder failure")
-        return [[0.0] * self.dimensions for _ in texts]
+        return [[self.value] * self.dimensions for _ in texts]
 
 
 @dataclass