""" Offline unit tests for DocumentSyncService vector indexing fixes. Pins the Phase D fixes: - ensure_collection is actually awaited (it used to be a bare coroutine that never ran, so fresh tenants had no collection at upsert time) - None entries from embed_batch are filtered out instead of poisoning the whole batch upsert (one failed chunk aborted the document) - delete-LAST reindex order (same as the wiki fix): new points are upserted with deterministic uuid5 ids BEFORE stale points are pruned, so a failed embedding pass can no longer leave a document with zero vectors (the old order ran delete_by_filter first) """ import uuid from unittest.mock import AsyncMock, MagicMock import pytest from src.services.document_sync_service import DocumentSyncService TENANT = "llm_tester" TENANT_COLLECTION = "library_desk_llm_tester" @pytest.fixture def mock_qdrant(): qdrant = MagicMock() qdrant.ensure_collection = AsyncMock() qdrant.upsert_points = AsyncMock(side_effect=lambda collection_name, points: len(points)) qdrant.scroll_all_points = AsyncMock(return_value=[]) qdrant.delete_by_ids = AsyncMock(side_effect=lambda collection_name, point_ids: len(point_ids)) return qdrant @pytest.fixture def mock_ollama(): ollama = MagicMock() ollama.embed_batch = AsyncMock( side_effect=lambda texts: [[0.1] * 768 for _ in texts] ) return ollama @pytest.fixture def mock_neo4j(): neo4j = MagicMock() neo4j.execute_query = AsyncMock(return_value=[]) neo4j.execute_write = AsyncMock(return_value=[]) return neo4j @pytest.fixture def mock_paperless(): paperless = MagicMock() paperless.get_custom_field_by_name = AsyncMock(return_value=None) return paperless @pytest.fixture def sync_service(mock_paperless, mock_qdrant, mock_ollama, mock_neo4j): return DocumentSyncService( paperless_client=mock_paperless, qdrant_client=mock_qdrant, ollama_client=mock_ollama, neo4j_client=mock_neo4j, wiki_client=MagicMock(), settings=MagicMock(), ) @pytest.mark.unit class TestIndexVectors: async def test_ensure_collection_is_awaited(self, sync_service, mock_qdrant): result = await sync_service.index_document( document_id=1, user=TENANT, content="hello world", title="Doc" ) assert result.success is True mock_qdrant.ensure_collection.assert_awaited_once_with(TENANT_COLLECTION) async def test_point_ids_are_deterministic(self, sync_service, mock_qdrant): await sync_service.index_document( document_id=7, user=TENANT, content="hello world", title="Doc" ) points = mock_qdrant.upsert_points.await_args.kwargs["points"] expected = str(uuid.uuid5(uuid.NAMESPACE_DNS, "document_7_chunk_0")) assert points[0]["id"] == expected async def test_stale_chunks_pruned_after_upsert(self, sync_service, mock_qdrant): """Old points not in the new set are deleted AFTER the new upsert.""" call_order = [] mock_qdrant.upsert_points = AsyncMock( side_effect=lambda collection_name, points: ( call_order.append("upsert"), len(points))[1] ) stale_id = str(uuid.uuid4()) # legacy random-uuid4 point kept_id = str(uuid.uuid5(uuid.NAMESPACE_DNS, "document_7_chunk_0")) mock_qdrant.scroll_all_points = AsyncMock( return_value=[{"id": stale_id, "payload": {}}, {"id": kept_id, "payload": {}}] ) mock_qdrant.delete_by_ids = AsyncMock( side_effect=lambda collection_name, point_ids: ( call_order.append("delete"), len(point_ids))[1] ) await sync_service.index_document( document_id=7, user=TENANT, content="hello world", title="Doc" ) assert call_order == ["upsert", "delete"] mock_qdrant.delete_by_ids.assert_awaited_once_with( collection_name=TENANT_COLLECTION, point_ids=[stale_id], ) mock_qdrant.scroll_all_points.assert_awaited_once_with( collection_name=TENANT_COLLECTION, filter_conditions={"doc_type": "document", "paperless_id": 7}, with_payload=False, ) async def test_upsert_routed_through_wrapper(self, sync_service, mock_qdrant): result = await sync_service.index_document( document_id=1, user=TENANT, content="hello world", title="Doc" ) assert result.chunks_created == 1 kwargs = mock_qdrant.upsert_points.await_args.kwargs assert kwargs["collection_name"] == TENANT_COLLECTION payload = kwargs["points"][0]["payload"] assert payload["doc_type"] == "document" assert payload["paperless_id"] == 1 async def test_failed_chunk_embedding_is_skipped_not_fatal( self, sync_service, mock_qdrant, mock_ollama ): # Three chunks; the middle embedding fails long_content = " ".join(f"word{i}" for i in range(1200)) mock_ollama.embed_batch = AsyncMock( side_effect=lambda texts: [ [0.1] * 768 if i != 1 else None for i in range(len(texts)) ] ) result = await sync_service.index_document( document_id=2, user=TENANT, content=long_content, title="Doc" ) assert result.success is True points = mock_qdrant.upsert_points.await_args.kwargs["points"] assert result.chunks_created == len(points) # The failed chunk (index 1) is absent, the others kept their index indices = [p["payload"]["chunk_index"] for p in points] assert 1 not in indices assert len(indices) >= 2 async def test_all_embeddings_failed_reports_failure( self, sync_service, mock_qdrant, mock_ollama ): mock_ollama.embed_batch = AsyncMock( side_effect=lambda texts: [None for _ in texts] ) result = await sync_service.index_document( document_id=3, user=TENANT, content="hello world", title="Doc" ) assert result.success is False assert "embed" in (result.error or "").lower() mock_qdrant.upsert_points.assert_not_awaited() # Delete-last: a fully failed embedding pass must leave the old # vectors untouched (previously delete_by_filter ran first, leaving # the document with zero vectors until the next successful sync) mock_qdrant.delete_by_ids.assert_not_awaited()