DocumentSyncService._index_vectors ran delete_by_filter on the
document's existing chunks FIRST and only then embedded; if the
embedding pass failed (Ollama down) the Paperless document was left
with zero vectors until the next successful sync - the same
zero-vector hazard already fixed for wiki pages in
VectorService.update_from_page.
Chunk ids are now deterministic uuid5 (document_{id}_chunk_{i}) so
re-upserting overwrites in place; new points are upserted first, then
stale points (including legacy random-uuid4 ones) are pruned via
scroll + delete_by_ids, and only after a successful upsert.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QbFZyDvYksazX6nYQYZ67L
177 lines
6.4 KiB
Python
177 lines
6.4 KiB
Python
"""
|
|
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()
|