From 1b873e5a6f0b0acbe98668272e9c00058458ffa9 Mon Sep 17 00:00:00 2001 From: Ali Zarinkolah Date: Thu, 20 Aug 2026 22:26:36 +0330 Subject: [PATCH] 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 --- tests/e2e/conftest.py | 244 ++++++++++++++++++++++ tests/e2e/test_ingestion_slice.py | 329 ++++++++++++++++++++++++++++++ tests/fakes.py | 5 +- 3 files changed, 577 insertions(+), 1 deletion(-) create mode 100644 tests/e2e/conftest.py create mode 100644 tests/e2e/test_ingestion_slice.py diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py new file mode 100644 index 0000000..6e78340 --- /dev/null +++ b/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 diff --git a/tests/e2e/test_ingestion_slice.py b/tests/e2e/test_ingestion_slice.py new file mode 100644 index 0000000..c0d62ac --- /dev/null +++ b/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 diff --git a/tests/fakes.py b/tests/fakes.py index 3a80dfb..18a2a01 100644 --- a/tests/fakes.py +++ b/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