"""`index_chunks`: payload correctness, bounded batching, and the ordering that keeps a failed attempt from damaging a working index (ADR-0001, ADR-0017). """ import uuid import pytest from anyio import CapacityLimiter from src.application.ingestion.chunking import chunk_id_for from src.application.ingestion.errors import PointIndexingError from src.application.ingestion.models import Chunk, ContentType, EmbeddedChunk, SparseVector from src.application.points import index_chunks from src.config import QdrantSettings from tests.fakes import FakeDenseEmbedder, FakePointStorage, FakeSparseEmbedder pytestmark = [pytest.mark.unit, pytest.mark.asyncio] _TENANT_ID = uuid.UUID("11111111-1111-1111-1111-111111111111") _FILE_ID = uuid.UUID("22222222-2222-2222-2222-222222222222") _API_KEY_ID = uuid.UUID("33333333-3333-3333-3333-333333333333") _ACTOR = f"api_key:{_API_KEY_ID}" def _embedded(count: int) -> list[EmbeddedChunk]: chunks = [ Chunk( chunk_id=chunk_id_for(_FILE_ID, index), chunk_index=index, order_id=float(index + 1), content=f"chunk {index}", content_type=ContentType.PARAGRAPH, token_count=2, character_count=7, ) for index in range(count) ] for position, chunk in enumerate(chunks): if position > 0: chunk.previous_chunk_id = chunks[position - 1].chunk_id if position < len(chunks) - 1: chunk.next_chunk_id = chunks[position + 1].chunk_id return [ EmbeddedChunk( chunk=chunk, dense={"dense_nomic": [0.0] * 4, "dense_openai": [1.0] * 4}, sparse=SparseVector(indices=[7], values=[0.5]), ) for chunk in chunks ] async def _index( storage: FakePointStorage, *, count: int = 3, settings: QdrantSettings | None = None, domain: str = "fire", ): return await index_chunks( _embedded(count), storage=storage, tenant_id=_TENANT_ID, domain=domain, file_id=_FILE_ID, source_filename="policy.docx", source_type="docx", actor=_ACTOR, dense_embedders=[ FakeDenseEmbedder(name="dense_nomic", model_version="nomic-embed-text-v2-moe"), FakeDenseEmbedder(name="dense_openai", model_version="text-embedding-3-large"), ], sparse_embedder=FakeSparseEmbedder(model_version="bm25-fa_norm_stop"), settings=settings or QdrantSettings(), thread_limiter=CapacityLimiter(2), ) async def test_index_chunks_writes_every_adr_0001_payload_field() -> None: storage = FakePointStorage() await _index(storage, count=3) payload = storage.points[str(chunk_id_for(_FILE_ID, 1))].payload assert payload["tenant_id"] == str(_TENANT_ID) assert payload["domain"] == "fire" assert payload["file_id"] == str(_FILE_ID) assert payload["chunk_id"] == str(chunk_id_for(_FILE_ID, 1)) assert payload["content"] == "chunk 1" assert payload["content_type"] == "paragraph" assert payload["source_filename"] == "policy.docx" assert payload["source_type"] == "docx" assert payload["order_id"] == 2.0 assert payload["chunk_index"] == 1 assert payload["previous_chunk_id"] == str(chunk_id_for(_FILE_ID, 0)) assert payload["next_chunk_id"] == str(chunk_id_for(_FILE_ID, 2)) assert payload["is_active"] is True assert payload["deleted_at"] is None assert payload["created_by"] == _ACTOR assert payload["updated_by"] == _ACTOR assert payload["version"] == 1 assert payload["created_at"] == payload["updated_at"] assert isinstance(payload["content_hash"], str) # Sorted, so wiring order cannot change the value (ADR-0001). assert payload["embedding_model_version"] == ( "bm25-fa_norm_stop+nomic-embed-text-v2-moe+text-embedding-3-large" ) async def test_index_chunks_uses_null_neighbours_at_the_file_ends() -> None: storage = FakePointStorage() await _index(storage, count=3) assert storage.points[str(chunk_id_for(_FILE_ID, 0))].payload["previous_chunk_id"] is None assert storage.points[str(chunk_id_for(_FILE_ID, 2))].payload["next_chunk_id"] is None async def test_index_chunks_derives_tenant_and_domain_from_the_caller_not_the_chunk() -> None: """Tenant identity is server-derived; nothing in the chunk can assert it.""" storage = FakePointStorage() await _index(storage, count=1, domain="car") payload = storage.points[str(chunk_id_for(_FILE_ID, 0))].payload assert payload["tenant_id"] == str(_TENANT_ID) assert payload["domain"] == "car" async def test_index_chunks_uses_deterministic_point_ids() -> None: storage = FakePointStorage() result = await _index(storage, count=4) assert result.points_upserted == 4 assert set(storage.points) == {str(chunk_id_for(_FILE_ID, i)) for i in range(4)} async def test_index_chunks_repeated_run_produces_no_duplicate_points() -> None: storage = FakePointStorage() await _index(storage, count=4) await _index(storage, count=4) assert len(storage.points) == 4 async def test_index_chunks_batches_at_the_configured_size() -> None: storage = FakePointStorage() settings = QdrantSettings(upsert_batch_size=2, upsert_concurrency=4) await _index(storage, count=5, settings=settings) assert storage.upsert_batches == [2, 2, 1] async def test_index_chunks_bounds_in_flight_batches() -> None: storage = FakePointStorage() settings = QdrantSettings(upsert_batch_size=1, upsert_concurrency=2) await _index(storage, count=8, settings=settings) assert len(storage.upsert_batches) == 8 assert storage.max_in_flight <= 2 async def test_index_chunks_soft_deletes_only_points_past_the_new_chunk_count() -> None: storage = FakePointStorage() await _index(storage, count=5) result = await _index(storage, count=2) assert result.points_soft_deleted == 3 assert storage.points[str(chunk_id_for(_FILE_ID, 1))].payload["is_active"] is True assert storage.points[str(chunk_id_for(_FILE_ID, 2))].payload["is_active"] is False assert storage.points[str(chunk_id_for(_FILE_ID, 4))].payload["is_active"] is False async def test_index_chunks_does_not_soft_delete_when_an_upsert_batch_fails() -> None: """A failed attempt must never remove content from a working index.""" storage = FakePointStorage() await _index(storage, count=5) storage.deactivate_calls.clear() storage.fail_on_batch = 1 with pytest.raises(PointIndexingError): await _index(storage, count=2, settings=QdrantSettings(upsert_batch_size=1)) assert storage.deactivate_calls == [] assert all(point.payload["is_active"] is True for point in storage.points.values()) async def test_index_chunks_raises_point_indexing_error_when_a_batch_fails() -> None: storage = FakePointStorage(fail_on_batch=0) with pytest.raises(PointIndexingError, match="upserting"): await _index(storage, count=2) async def test_index_chunks_raises_point_indexing_error_when_the_sweep_fails() -> None: storage = FakePointStorage(fail_deactivate=True) with pytest.raises(PointIndexingError, match="soft-deleting"): await _index(storage, count=2) async def test_index_chunks_on_empty_input_touches_no_storage() -> None: storage = FakePointStorage() result = await index_chunks( [], storage=storage, tenant_id=_TENANT_ID, domain="fire", file_id=_FILE_ID, source_filename="empty.csv", source_type="csv", actor=_ACTOR, dense_embedders=[FakeDenseEmbedder(name="dense_nomic")], sparse_embedder=FakeSparseEmbedder(), settings=QdrantSettings(), thread_limiter=CapacityLimiter(2), ) assert result.points_upserted == 0 assert storage.upsert_batches == [] assert storage.deactivate_calls == []