wehub-resource-sync
7a0da7932b
OSV-Scanner (Scheduled) / scan-scheduled (push) Failing after 0s
Create Release / test-gate (push) Has been cancelled
Create Release / release-gate (push) Has been cancelled
Create Release / ci-gate (push) Has been cancelled
Create Release / version-check (push) Has been cancelled
Create Release / e2e-test-gate (push) Has been cancelled
Create Release / responsive-test-gate (push) Has been cancelled
Create Release / compat-test-gate (push) Has been cancelled
Create Release / compose-integration-gate (push) Has been cancelled
Create Release / vulture-gate (push) Has been cancelled
Create Release / build (push) Has been cancelled
Create Release / provenance (push) Has been cancelled
Create Release / prerelease-docker (push) Has been cancelled
Create Release / publish-docker (push) Has been cancelled
Create Release / create-release (push) Has been cancelled
Create Release / cleanup-changelog (push) Has been cancelled
Create Release / trigger-pypi (push) Has been cancelled
Create Release / monitor-pypi (push) Has been cancelled
Create Release / Clean up orphan prerelease tags and signatures (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-form] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-metrics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-workflow] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-core] (push) Has been cancelled
CodeQL Advanced / Analyze (javascript-typescript) (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [history-news] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [library] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [link-analytics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-core] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-lifecycle] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [error-benchmark] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) (push) Has been cancelled
Docker Tests (Consolidated) / Accessibility Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Unit Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Example Tests (push) Has been cancelled
Docker Tests (Consolidated) / Production Image Smoke Test (push) Has been cancelled
Docker Tests (Consolidated) / Infrastructure Tests (push) Has been cancelled
OSSF Scorecard / OSSF Security Scorecard Analysis (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [mobile] (push) Has been cancelled
Backwards Compatibility / Verify Encryption Constants (push) Has been cancelled
Backwards Compatibility / PyPI Version Compatibility (push) Has been cancelled
Backwards Compatibility / Database Migration Tests (push) Has been cancelled
CodeQL Advanced / Analyze (python) (push) Has been cancelled
Docker Tests (Consolidated) / detect-changes (push) Has been cancelled
Docker Tests (Consolidated) / Build Test Image (push) Has been cancelled
Docker Tests (Consolidated) / All Pytest Tests + Coverage (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [accessibility] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [api-crud] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-login] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-register] (push) Has been cancelled
1573 lines
60 KiB
Python
1573 lines
60 KiB
Python
"""
|
|
Coverage tests for LibraryRAGService.
|
|
|
|
Focuses on logic paths not exercised by the existing test_library_rag_service.py:
|
|
- _get_index_hash edge cases
|
|
- _get_index_path cache directory details
|
|
- _deduplicate_chunks order preservation and empty input
|
|
- _get_or_create_rag_index new vs existing
|
|
- load_or_create_faiss_index HNSW/IVF/L2/IP variants, integrity failure, dimension mismatch,
|
|
load failure, corrupted unlink failure
|
|
- index_document chunk indexing, empty text, skip already indexed
|
|
- index_all_documents with progress callback, stores settings
|
|
"""
|
|
|
|
import hashlib
|
|
import threading
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from langchain_core.documents import Document as LangchainDocument
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Module-level patch path prefix
|
|
# ---------------------------------------------------------------------------
|
|
_MOD = "local_deep_research.research_library.services.library_rag_service"
|
|
|
|
|
|
def _make_service(**overrides):
|
|
"""Create a LibraryRAGService with all external deps mocked out."""
|
|
with (
|
|
patch(f"{_MOD}.LocalEmbeddingManager") as _lem,
|
|
patch(f"{_MOD}.get_user_db_session"),
|
|
patch(f"{_MOD}.FileIntegrityManager") as _fim,
|
|
patch(f"{_MOD}.get_text_splitter") as _gts,
|
|
):
|
|
_lem.return_value.embeddings = MagicMock()
|
|
from local_deep_research.research_library.services.library_rag_service import (
|
|
LibraryRAGService,
|
|
)
|
|
|
|
defaults = dict(username="testuser", db_password="pw")
|
|
defaults.update(overrides)
|
|
svc = LibraryRAGService(**defaults)
|
|
return svc
|
|
|
|
|
|
# =========================================================================
|
|
# _get_index_hash
|
|
# =========================================================================
|
|
class TestGetIndexHash:
|
|
"""Tests for _get_index_hash determinism and sensitivity."""
|
|
|
|
def test_hash_is_deterministic(self):
|
|
svc = _make_service()
|
|
h1 = svc._get_index_hash("col_a", "model_x", "sentence_transformers")
|
|
h2 = svc._get_index_hash("col_a", "model_x", "sentence_transformers")
|
|
assert h1 == h2
|
|
|
|
def test_hash_changes_with_collection_name(self):
|
|
svc = _make_service()
|
|
h1 = svc._get_index_hash("col_a", "model_x", "sentence_transformers")
|
|
h2 = svc._get_index_hash("col_b", "model_x", "sentence_transformers")
|
|
assert h1 != h2
|
|
|
|
def test_hash_changes_with_model(self):
|
|
svc = _make_service()
|
|
h1 = svc._get_index_hash("col_a", "model_x", "sentence_transformers")
|
|
h2 = svc._get_index_hash("col_a", "model_y", "sentence_transformers")
|
|
assert h1 != h2
|
|
|
|
def test_hash_changes_with_provider(self):
|
|
svc = _make_service()
|
|
h1 = svc._get_index_hash("col_a", "model_x", "sentence_transformers")
|
|
h2 = svc._get_index_hash("col_a", "model_x", "ollama")
|
|
assert h1 != h2
|
|
|
|
def test_hash_is_sha256_hex(self):
|
|
svc = _make_service()
|
|
h = svc._get_index_hash("c", "m", "p")
|
|
expected = hashlib.sha256("c:m:p".encode()).hexdigest()
|
|
assert h == expected
|
|
|
|
def test_hash_length_is_64_chars(self):
|
|
svc = _make_service()
|
|
h = svc._get_index_hash("x", "y", "z")
|
|
assert len(h) == 64
|
|
|
|
|
|
# =========================================================================
|
|
# _get_index_path
|
|
# =========================================================================
|
|
class TestGetIndexPath:
|
|
"""Tests for _get_index_path."""
|
|
|
|
@patch(f"{_MOD}.get_cache_directory")
|
|
def test_path_under_rag_indices_subdir(self, mock_cache_dir, tmp_path):
|
|
mock_cache_dir.return_value = tmp_path
|
|
svc = _make_service()
|
|
p = svc._get_index_path("abc123")
|
|
assert p.parent.name == "rag_indices"
|
|
|
|
@patch(f"{_MOD}.get_cache_directory")
|
|
def test_path_filename_contains_hash(self, mock_cache_dir, tmp_path):
|
|
mock_cache_dir.return_value = tmp_path
|
|
svc = _make_service()
|
|
p = svc._get_index_path("abc123")
|
|
assert p.name == "abc123.faiss"
|
|
|
|
@patch(f"{_MOD}.get_cache_directory")
|
|
def test_path_creates_directory(self, mock_cache_dir, tmp_path):
|
|
mock_cache_dir.return_value = tmp_path
|
|
svc = _make_service()
|
|
svc._get_index_path("abc123")
|
|
assert (tmp_path / "rag_indices").is_dir()
|
|
|
|
@patch(f"{_MOD}.get_cache_directory")
|
|
def test_path_returns_path_object(self, mock_cache_dir, tmp_path):
|
|
mock_cache_dir.return_value = tmp_path
|
|
svc = _make_service()
|
|
p = svc._get_index_path("somehash")
|
|
assert isinstance(p, Path)
|
|
|
|
|
|
# =========================================================================
|
|
# _deduplicate_chunks (static method)
|
|
# =========================================================================
|
|
class TestDeduplicateChunks:
|
|
"""Tests for _deduplicate_chunks."""
|
|
|
|
def _doc(self, text):
|
|
return LangchainDocument(page_content=text)
|
|
|
|
def test_empty_input_returns_empty(self):
|
|
from local_deep_research.research_library.services.library_rag_service import (
|
|
LibraryRAGService,
|
|
)
|
|
|
|
chunks, ids = LibraryRAGService._deduplicate_chunks([], [])
|
|
assert chunks == []
|
|
assert ids == []
|
|
|
|
def test_no_duplicates_preserved(self):
|
|
from local_deep_research.research_library.services.library_rag_service import (
|
|
LibraryRAGService,
|
|
)
|
|
|
|
d1, d2 = self._doc("a"), self._doc("b")
|
|
chunks, ids = LibraryRAGService._deduplicate_chunks(
|
|
[d1, d2], ["id1", "id2"]
|
|
)
|
|
assert ids == ["id1", "id2"]
|
|
assert chunks == [d1, d2]
|
|
|
|
def test_duplicate_ids_keeps_first(self):
|
|
from local_deep_research.research_library.services.library_rag_service import (
|
|
LibraryRAGService,
|
|
)
|
|
|
|
d1, d2, d3 = self._doc("a"), self._doc("b"), self._doc("c")
|
|
chunks, ids = LibraryRAGService._deduplicate_chunks(
|
|
[d1, d2, d3], ["id1", "id1", "id2"]
|
|
)
|
|
assert ids == ["id1", "id2"]
|
|
assert chunks == [d1, d3]
|
|
|
|
def test_existing_ids_excluded(self):
|
|
from local_deep_research.research_library.services.library_rag_service import (
|
|
LibraryRAGService,
|
|
)
|
|
|
|
d1, d2 = self._doc("a"), self._doc("b")
|
|
chunks, ids = LibraryRAGService._deduplicate_chunks(
|
|
[d1, d2], ["id1", "id2"], existing_ids={"id1"}
|
|
)
|
|
assert ids == ["id2"]
|
|
assert chunks == [d2]
|
|
|
|
def test_none_existing_ids_allows_all(self):
|
|
from local_deep_research.research_library.services.library_rag_service import (
|
|
LibraryRAGService,
|
|
)
|
|
|
|
d1 = self._doc("a")
|
|
chunks, ids = LibraryRAGService._deduplicate_chunks(
|
|
[d1], ["id1"], existing_ids=None
|
|
)
|
|
assert ids == ["id1"]
|
|
|
|
def test_order_preservation(self):
|
|
from local_deep_research.research_library.services.library_rag_service import (
|
|
LibraryRAGService,
|
|
)
|
|
|
|
docs = [self._doc(str(i)) for i in range(5)]
|
|
id_list = ["e", "d", "c", "b", "a"]
|
|
chunks, ids = LibraryRAGService._deduplicate_chunks(docs, id_list)
|
|
assert ids == ["e", "d", "c", "b", "a"]
|
|
|
|
def test_all_existing_returns_empty(self):
|
|
from local_deep_research.research_library.services.library_rag_service import (
|
|
LibraryRAGService,
|
|
)
|
|
|
|
d1, d2 = self._doc("a"), self._doc("b")
|
|
chunks, ids = LibraryRAGService._deduplicate_chunks(
|
|
[d1, d2], ["id1", "id2"], existing_ids={"id1", "id2"}
|
|
)
|
|
assert chunks == []
|
|
assert ids == []
|
|
|
|
|
|
# =========================================================================
|
|
# _get_or_create_rag_index
|
|
# =========================================================================
|
|
class TestGetOrCreateRagIndex:
|
|
"""Tests for _get_or_create_rag_index."""
|
|
|
|
@patch(f"{_MOD}.get_user_db_session")
|
|
def test_returns_existing_index(self, mock_session_ctx):
|
|
svc = _make_service()
|
|
mock_session = MagicMock()
|
|
mock_session_ctx.return_value.__enter__ = MagicMock(
|
|
return_value=mock_session
|
|
)
|
|
mock_session_ctx.return_value.__exit__ = MagicMock(return_value=None)
|
|
|
|
existing_index = MagicMock()
|
|
mock_session.query.return_value.filter_by.return_value.first.return_value = existing_index
|
|
|
|
result = svc._get_or_create_rag_index("coll-123")
|
|
assert result is existing_index
|
|
# Should NOT call session.add since index already existed
|
|
mock_session.add.assert_not_called()
|
|
|
|
@patch(f"{_MOD}.get_cache_directory")
|
|
@patch(f"{_MOD}.get_user_db_session")
|
|
def test_creates_new_index_when_none_exists(
|
|
self, mock_session_ctx, mock_cache_dir, tmp_path
|
|
):
|
|
mock_cache_dir.return_value = tmp_path
|
|
svc = _make_service()
|
|
svc.embedding_manager = MagicMock()
|
|
svc.embedding_manager.embeddings.embed_query.return_value = [0.0] * 384
|
|
|
|
mock_session = MagicMock()
|
|
mock_session_ctx.return_value.__enter__ = MagicMock(
|
|
return_value=mock_session
|
|
)
|
|
mock_session_ctx.return_value.__exit__ = MagicMock(return_value=None)
|
|
|
|
# No existing index found
|
|
mock_session.query.return_value.filter_by.return_value.first.return_value = None
|
|
|
|
svc._get_or_create_rag_index("coll-456")
|
|
|
|
# Should have called session.add for the new RAGIndex
|
|
mock_session.add.assert_called_once()
|
|
mock_session.commit.assert_called_once()
|
|
mock_session.refresh.assert_called_once()
|
|
|
|
@patch(f"{_MOD}.get_cache_directory")
|
|
@patch(f"{_MOD}.get_user_db_session")
|
|
def test_embeds_test_string_for_dimension(
|
|
self, mock_session_ctx, mock_cache_dir, tmp_path
|
|
):
|
|
mock_cache_dir.return_value = tmp_path
|
|
svc = _make_service()
|
|
svc.embedding_manager = MagicMock()
|
|
svc.embedding_manager.embeddings.embed_query.return_value = [0.1] * 768
|
|
|
|
mock_session = MagicMock()
|
|
mock_session_ctx.return_value.__enter__ = MagicMock(
|
|
return_value=mock_session
|
|
)
|
|
mock_session_ctx.return_value.__exit__ = MagicMock(return_value=None)
|
|
mock_session.query.return_value.filter_by.return_value.first.return_value = None
|
|
|
|
svc._get_or_create_rag_index("coll-789")
|
|
|
|
svc.embedding_manager.embeddings.embed_query.assert_called_once_with(
|
|
"test"
|
|
)
|
|
|
|
|
|
# =========================================================================
|
|
# load_or_create_faiss_index
|
|
# =========================================================================
|
|
class TestLoadOrCreateFaissIndex:
|
|
"""Tests for load_or_create_faiss_index."""
|
|
|
|
def _patch_get_or_create(self, svc, rag_index):
|
|
"""Patch _get_or_create_rag_index on a service instance."""
|
|
svc._get_or_create_rag_index = MagicMock(return_value=rag_index)
|
|
|
|
def _make_rag_index(self, index_path="/tmp/test.faiss", dim=384):
|
|
idx = MagicMock()
|
|
idx.index_path = index_path
|
|
idx.embedding_dimension = dim
|
|
idx.id = "rag-idx-1"
|
|
return idx
|
|
|
|
@patch(f"{_MOD}.FAISS")
|
|
@patch(f"{_MOD}.InMemoryDocstore")
|
|
@patch(f"{_MOD}.IndexFlatIP")
|
|
def test_creates_flat_ip_for_cosine(
|
|
self, mock_flat_ip, mock_docstore, mock_faiss
|
|
):
|
|
svc = _make_service(distance_metric="cosine", index_type="flat")
|
|
rag_idx = self._make_rag_index(index_path="/nonexistent/test.faiss")
|
|
self._patch_get_or_create(svc, rag_idx)
|
|
svc.embedding_manager = MagicMock()
|
|
|
|
svc.load_or_create_faiss_index("coll-1")
|
|
mock_flat_ip.assert_called_once_with(384)
|
|
|
|
@patch(f"{_MOD}.FAISS")
|
|
@patch(f"{_MOD}.InMemoryDocstore")
|
|
@patch(f"{_MOD}.IndexFlatL2")
|
|
def test_creates_flat_l2_for_l2_metric(
|
|
self, mock_flat_l2, mock_docstore, mock_faiss
|
|
):
|
|
svc = _make_service(distance_metric="l2", index_type="flat")
|
|
rag_idx = self._make_rag_index(index_path="/nonexistent/test.faiss")
|
|
self._patch_get_or_create(svc, rag_idx)
|
|
svc.embedding_manager = MagicMock()
|
|
|
|
svc.load_or_create_faiss_index("coll-1")
|
|
mock_flat_l2.assert_called_once_with(384)
|
|
|
|
@patch(f"{_MOD}.FAISS")
|
|
@patch(f"{_MOD}.InMemoryDocstore")
|
|
@patch(f"{_MOD}.IndexHNSWFlat")
|
|
def test_creates_hnsw_index(self, mock_hnsw, mock_docstore, mock_faiss):
|
|
svc = _make_service(index_type="hnsw")
|
|
rag_idx = self._make_rag_index(index_path="/nonexistent/test.faiss")
|
|
self._patch_get_or_create(svc, rag_idx)
|
|
svc.embedding_manager = MagicMock()
|
|
|
|
svc.load_or_create_faiss_index("coll-1")
|
|
mock_hnsw.assert_called_once_with(384, 32)
|
|
|
|
@patch(f"{_MOD}.FAISS")
|
|
@patch(f"{_MOD}.InMemoryDocstore")
|
|
@patch(f"{_MOD}.IndexFlatIP")
|
|
def test_ivf_falls_back_to_flat_ip_for_cosine(
|
|
self, mock_flat_ip, mock_docstore, mock_faiss
|
|
):
|
|
svc = _make_service(index_type="ivf", distance_metric="cosine")
|
|
rag_idx = self._make_rag_index(index_path="/nonexistent/test.faiss")
|
|
self._patch_get_or_create(svc, rag_idx)
|
|
svc.embedding_manager = MagicMock()
|
|
|
|
svc.load_or_create_faiss_index("coll-1")
|
|
mock_flat_ip.assert_called_once_with(384)
|
|
|
|
@patch(f"{_MOD}.FAISS")
|
|
@patch(f"{_MOD}.InMemoryDocstore")
|
|
@patch(f"{_MOD}.IndexFlatL2")
|
|
def test_ivf_falls_back_to_flat_l2_for_l2(
|
|
self, mock_flat_l2, mock_docstore, mock_faiss
|
|
):
|
|
svc = _make_service(index_type="ivf", distance_metric="l2")
|
|
rag_idx = self._make_rag_index(index_path="/nonexistent/test.faiss")
|
|
self._patch_get_or_create(svc, rag_idx)
|
|
svc.embedding_manager = MagicMock()
|
|
|
|
svc.load_or_create_faiss_index("coll-1")
|
|
mock_flat_l2.assert_called_once_with(384)
|
|
|
|
@patch(f"{_MOD}.safe_load_faiss")
|
|
@patch(f"{_MOD}.FAISS")
|
|
@patch(f"{_MOD}.InMemoryDocstore")
|
|
@patch(f"{_MOD}.IndexFlatIP")
|
|
def test_verified_load_returns_existing_index(
|
|
self, mock_flat_ip, mock_docstore, mock_faiss_cls, mock_safe_load
|
|
):
|
|
svc = _make_service()
|
|
svc.embedding_manager = MagicMock()
|
|
svc.embedding_manager.embeddings.embed_query.return_value = [0.0] * 384
|
|
svc.integrity_manager = MagicMock()
|
|
svc.integrity_manager.verify_file.return_value = (True, None)
|
|
|
|
rag_idx = self._make_rag_index(dim=384)
|
|
rag_idx.index_path = "/tmp/existing.faiss"
|
|
self._patch_get_or_create(svc, rag_idx)
|
|
|
|
mock_loaded = MagicMock()
|
|
mock_safe_load.return_value = mock_loaded
|
|
|
|
with patch.object(Path, "exists", return_value=True):
|
|
result = svc.load_or_create_faiss_index("coll-1")
|
|
|
|
# Verified index is loaded via the restricted-unpickler loader,
|
|
# never via the dangerous FAISS.load_local.
|
|
assert result is mock_loaded
|
|
mock_safe_load.assert_called_once()
|
|
mock_faiss_cls.load_local.assert_not_called()
|
|
|
|
@patch(f"{_MOD}.FAISS")
|
|
@patch(f"{_MOD}.InMemoryDocstore")
|
|
@patch(f"{_MOD}.IndexFlatIP")
|
|
def test_integrity_failure_quarantines_and_creates_new_index(
|
|
self, mock_flat_ip, mock_docstore, mock_faiss_cls, tmp_path
|
|
):
|
|
"""Verify-failure path: corrupt .faiss + .pkl are RENAMED to
|
|
.corrupt-<ns>, NOT unlinked. Then fresh index is created.
|
|
Regression for #4197 data-loss bug.
|
|
"""
|
|
svc = _make_service()
|
|
svc.embedding_manager = MagicMock()
|
|
svc.integrity_manager = MagicMock()
|
|
svc.integrity_manager.verify_file.return_value = (
|
|
False,
|
|
"hash mismatch",
|
|
)
|
|
|
|
# Use real on-disk files so .exists() in the quarantine
|
|
# collision check returns False naturally for the .corrupt-*
|
|
# paths, not True-for-everything.
|
|
idx_path = tmp_path / "corrupt.faiss"
|
|
pkl_path = tmp_path / "corrupt.pkl"
|
|
idx_path.write_bytes(b"faiss-bytes")
|
|
pkl_path.write_bytes(b"pkl-bytes")
|
|
|
|
rag_idx = self._make_rag_index()
|
|
rag_idx.index_path = str(idx_path)
|
|
self._patch_get_or_create(svc, rag_idx)
|
|
|
|
with patch.object(Path, "unlink") as mock_unlink:
|
|
svc.load_or_create_faiss_index("coll-1")
|
|
|
|
# Should NOT have unlinked anything in the verify-failure path
|
|
mock_unlink.assert_not_called()
|
|
# Originals quarantined (renamed away from their paths)
|
|
assert not idx_path.exists()
|
|
assert not pkl_path.exists()
|
|
# Both files preserved under .corrupt-<ns> names
|
|
corrupt_faiss = list(tmp_path.glob("corrupt.faiss.corrupt-*"))
|
|
corrupt_pkl = list(tmp_path.glob("corrupt.pkl.corrupt-*"))
|
|
assert len(corrupt_faiss) == 1
|
|
assert len(corrupt_pkl) == 1
|
|
assert corrupt_faiss[0].read_bytes() == b"faiss-bytes"
|
|
# Should return a new FAISS instance, not load_local
|
|
mock_faiss_cls.load_local.assert_not_called()
|
|
mock_faiss_cls.assert_called_once()
|
|
|
|
@patch(f"{_MOD}.FAISS")
|
|
@patch(f"{_MOD}.InMemoryDocstore")
|
|
@patch(f"{_MOD}.IndexFlatIP")
|
|
def test_quarantine_rename_oserror_re_raises(
|
|
self, mock_flat_ip, mock_docstore, mock_faiss_cls, tmp_path
|
|
):
|
|
"""Disk-full / read-only fs during quarantine MUST propagate.
|
|
Silently falling through would let the next save_local truncate
|
|
the corrupt bytes, recreating the very data loss #4197 fixes.
|
|
"""
|
|
svc = _make_service()
|
|
svc.embedding_manager = MagicMock()
|
|
svc.integrity_manager = MagicMock()
|
|
svc.integrity_manager.verify_file.return_value = (False, "bad hash")
|
|
|
|
idx_path = tmp_path / "corrupt.faiss"
|
|
idx_path.write_bytes(b"faiss-bytes")
|
|
(tmp_path / "corrupt.pkl").write_bytes(b"pkl-bytes")
|
|
|
|
rag_idx = self._make_rag_index()
|
|
rag_idx.index_path = str(idx_path)
|
|
self._patch_get_or_create(svc, rag_idx)
|
|
|
|
with patch.object(Path, "rename", side_effect=OSError("ENOSPC")):
|
|
with pytest.raises(OSError, match="ENOSPC"):
|
|
svc.load_or_create_faiss_index("coll-1")
|
|
|
|
# Fresh index must NOT have been created — corrupt bytes are
|
|
# still on disk and creating one would let the next save
|
|
# overwrite them.
|
|
mock_faiss_cls.assert_not_called()
|
|
|
|
@patch(f"{_MOD}.get_user_db_session")
|
|
@patch(f"{_MOD}.FAISS")
|
|
@patch(f"{_MOD}.InMemoryDocstore")
|
|
@patch(f"{_MOD}.IndexFlatIP")
|
|
def test_dimension_mismatch_deletes_and_rebuilds(
|
|
self, mock_flat_ip, mock_docstore, mock_faiss_cls, mock_session_ctx
|
|
):
|
|
svc = _make_service()
|
|
svc.embedding_manager = MagicMock()
|
|
# Current model returns dim 768 but index was stored with 384
|
|
svc.embedding_manager.embeddings.embed_query.return_value = [0.0] * 768
|
|
svc.integrity_manager = MagicMock()
|
|
svc.integrity_manager.verify_file.return_value = (True, None)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session_ctx.return_value.__enter__ = MagicMock(
|
|
return_value=mock_session
|
|
)
|
|
mock_session_ctx.return_value.__exit__ = MagicMock(return_value=None)
|
|
mock_session.query.return_value.filter_by.return_value.first.return_value = MagicMock()
|
|
|
|
rag_idx = self._make_rag_index(dim=384)
|
|
rag_idx.index_path = "/tmp/old_dim.faiss"
|
|
self._patch_get_or_create(svc, rag_idx)
|
|
|
|
with (
|
|
patch.object(Path, "exists", return_value=True),
|
|
patch.object(Path, "unlink") as mock_unlink,
|
|
patch.object(Path, "with_suffix") as mock_with_suffix,
|
|
):
|
|
mock_pkl = MagicMock()
|
|
mock_pkl.exists.return_value = True
|
|
mock_with_suffix.return_value = mock_pkl
|
|
svc.load_or_create_faiss_index("coll-1")
|
|
|
|
# The old file should have been unlinked
|
|
mock_unlink.assert_called()
|
|
# A new FAISS should be created (not loaded)
|
|
mock_faiss_cls.assert_called_once()
|
|
|
|
@patch(f"{_MOD}.safe_load_faiss")
|
|
@patch(f"{_MOD}.FAISS")
|
|
@patch(f"{_MOD}.InMemoryDocstore")
|
|
@patch(f"{_MOD}.IndexFlatIP")
|
|
def test_load_failure_quarantines_and_creates_new_index(
|
|
self,
|
|
mock_flat_ip,
|
|
mock_docstore,
|
|
mock_faiss_cls,
|
|
mock_safe_load,
|
|
tmp_path,
|
|
):
|
|
"""When the loader raises (torn .pkl, malformed pickle, or a
|
|
rejected/tampered pickle), the .faiss and .pkl are quarantined
|
|
before falling through to a fresh index. Previously the old code
|
|
silently discarded the broken-state files without preserving
|
|
evidence.
|
|
"""
|
|
svc = _make_service()
|
|
svc.embedding_manager = MagicMock()
|
|
svc.embedding_manager.embeddings.embed_query.return_value = [0.0] * 384
|
|
svc.integrity_manager = MagicMock()
|
|
svc.integrity_manager.verify_file.return_value = (True, None)
|
|
|
|
idx_path = tmp_path / "broken.faiss"
|
|
pkl_path = tmp_path / "broken.pkl"
|
|
idx_path.write_bytes(b"faiss-bytes")
|
|
pkl_path.write_bytes(b"pkl-bytes")
|
|
|
|
rag_idx = self._make_rag_index(dim=384)
|
|
rag_idx.index_path = str(idx_path)
|
|
self._patch_get_or_create(svc, rag_idx)
|
|
|
|
mock_safe_load.side_effect = RuntimeError("corrupted file")
|
|
|
|
svc.load_or_create_faiss_index("coll-1")
|
|
|
|
# Both files quarantined (renamed away)
|
|
assert not idx_path.exists()
|
|
assert not pkl_path.exists()
|
|
assert len(list(tmp_path.glob("broken.faiss.corrupt-*"))) == 1
|
|
assert len(list(tmp_path.glob("broken.pkl.corrupt-*"))) == 1
|
|
# Should fall through and create new index
|
|
mock_faiss_cls.assert_called_once()
|
|
|
|
@patch(f"{_MOD}.FAISS")
|
|
@patch(f"{_MOD}.InMemoryDocstore")
|
|
@patch(f"{_MOD}.IndexFlatIP")
|
|
def test_embedding_provider_failure_raises_and_preserves_index(
|
|
self, mock_flat_ip, mock_docstore, mock_faiss_cls, tmp_path
|
|
):
|
|
"""When the embedding provider is unreachable (e.g. Ollama down),
|
|
the dimension-check embed_query raises. That says nothing about
|
|
the index files, so the error must propagate WITHOUT quarantining
|
|
the healthy index or replacing it with an empty one.
|
|
"""
|
|
svc = _make_service()
|
|
svc.embedding_manager = MagicMock()
|
|
svc.embedding_manager.embeddings.embed_query.side_effect = (
|
|
ConnectionError("Ollama connection refused")
|
|
)
|
|
svc.integrity_manager = MagicMock()
|
|
svc.integrity_manager.verify_file.return_value = (True, None)
|
|
|
|
idx_path = tmp_path / "healthy.faiss"
|
|
pkl_path = tmp_path / "healthy.pkl"
|
|
idx_path.write_bytes(b"faiss-bytes")
|
|
pkl_path.write_bytes(b"pkl-bytes")
|
|
|
|
rag_idx = self._make_rag_index(dim=384)
|
|
rag_idx.index_path = str(idx_path)
|
|
self._patch_get_or_create(svc, rag_idx)
|
|
|
|
with pytest.raises(ConnectionError, match="Ollama"):
|
|
svc.load_or_create_faiss_index("coll-1")
|
|
|
|
# The healthy index files must be untouched — not quarantined
|
|
assert idx_path.read_bytes() == b"faiss-bytes"
|
|
assert pkl_path.read_bytes() == b"pkl-bytes"
|
|
assert list(tmp_path.glob("*.corrupt-*")) == []
|
|
# And no empty replacement index was created
|
|
mock_faiss_cls.load_local.assert_not_called()
|
|
mock_faiss_cls.assert_not_called()
|
|
|
|
@patch(f"{_MOD}.FAISS")
|
|
@patch(f"{_MOD}.InMemoryDocstore")
|
|
@patch(f"{_MOD}.IndexFlatIP")
|
|
def test_normalize_vectors_passed_to_faiss(
|
|
self, mock_flat_ip, mock_docstore, mock_faiss_cls
|
|
):
|
|
svc = _make_service(normalize_vectors=False)
|
|
rag_idx = self._make_rag_index(index_path="/nonexistent/test.faiss")
|
|
self._patch_get_or_create(svc, rag_idx)
|
|
svc.embedding_manager = MagicMock()
|
|
|
|
svc.load_or_create_faiss_index("coll-1")
|
|
call_kwargs = mock_faiss_cls.call_args
|
|
assert call_kwargs[1]["normalize_L2"] is False
|
|
|
|
|
|
# =========================================================================
|
|
# index_document
|
|
# =========================================================================
|
|
class TestIndexDocument:
|
|
"""Tests for index_document."""
|
|
|
|
@patch(f"{_MOD}.get_user_db_session")
|
|
def test_returns_error_when_document_not_found(self, mock_session_ctx):
|
|
svc = _make_service()
|
|
mock_session = MagicMock()
|
|
mock_session_ctx.return_value.__enter__ = MagicMock(
|
|
return_value=mock_session
|
|
)
|
|
mock_session_ctx.return_value.__exit__ = MagicMock(return_value=None)
|
|
mock_session.query.return_value.filter_by.return_value.first.return_value = None
|
|
|
|
result = svc.index_document("doc-1", "coll-1")
|
|
assert result["status"] == "error"
|
|
assert "not found" in result["error"]
|
|
|
|
@patch(f"{_MOD}.ensure_in_collection")
|
|
@patch(f"{_MOD}.get_user_db_session")
|
|
def test_returns_error_when_no_text_content(
|
|
self, mock_session_ctx, mock_ensure
|
|
):
|
|
svc = _make_service()
|
|
mock_session = MagicMock()
|
|
mock_session_ctx.return_value.__enter__ = MagicMock(
|
|
return_value=mock_session
|
|
)
|
|
mock_session_ctx.return_value.__exit__ = MagicMock(return_value=None)
|
|
|
|
mock_document = MagicMock()
|
|
mock_document.text_content = None
|
|
|
|
mock_session.query.return_value.filter_by.return_value.first.return_value = mock_document
|
|
|
|
mock_ensure.return_value = MagicMock(indexed=False, chunk_count=0)
|
|
|
|
result = svc.index_document("doc-1", "coll-1")
|
|
assert result["status"] == "error"
|
|
assert "no text content" in result["error"]
|
|
|
|
@patch(f"{_MOD}.ensure_in_collection")
|
|
@patch(f"{_MOD}.get_user_db_session")
|
|
def test_skips_already_indexed_document(
|
|
self, mock_session_ctx, mock_ensure
|
|
):
|
|
svc = _make_service()
|
|
mock_session = MagicMock()
|
|
mock_session_ctx.return_value.__enter__ = MagicMock(
|
|
return_value=mock_session
|
|
)
|
|
mock_session_ctx.return_value.__exit__ = MagicMock(return_value=None)
|
|
|
|
mock_document = MagicMock()
|
|
mock_session.query.return_value.filter_by.return_value.first.return_value = mock_document
|
|
|
|
mock_ensure.return_value = MagicMock(indexed=True, chunk_count=42)
|
|
|
|
result = svc.index_document("doc-1", "coll-1", force_reindex=False)
|
|
assert result["status"] == "skipped"
|
|
assert result["chunk_count"] == 42
|
|
|
|
|
|
# =========================================================================
|
|
# index_all_documents (index_collection in the spec)
|
|
# =========================================================================
|
|
class TestIndexAllDocuments:
|
|
"""Tests for index_all_documents."""
|
|
|
|
@patch(f"{_MOD}.get_user_db_session")
|
|
def test_no_documents_returns_info(self, mock_session_ctx):
|
|
svc = _make_service()
|
|
mock_session = MagicMock()
|
|
mock_session_ctx.return_value.__enter__ = MagicMock(
|
|
return_value=mock_session
|
|
)
|
|
mock_session_ctx.return_value.__exit__ = MagicMock(return_value=None)
|
|
mock_session.query.return_value.filter_by.return_value.filter_by.return_value.all.return_value = []
|
|
|
|
result = svc.index_all_documents("coll-1")
|
|
assert result["status"] == "info"
|
|
assert result["successful"] == 0
|
|
|
|
@patch(f"{_MOD}.get_user_db_session")
|
|
def test_progress_callback_invoked(self, mock_session_ctx):
|
|
svc = _make_service()
|
|
mock_session = MagicMock()
|
|
mock_session_ctx.return_value.__enter__ = MagicMock(
|
|
return_value=mock_session
|
|
)
|
|
mock_session_ctx.return_value.__exit__ = MagicMock(return_value=None)
|
|
|
|
mock_dc1 = MagicMock()
|
|
mock_dc1.document_id = "doc-1"
|
|
mock_dc2 = MagicMock()
|
|
mock_dc2.document_id = "doc-2"
|
|
|
|
mock_doc = MagicMock()
|
|
mock_doc.title = "Test Doc"
|
|
|
|
# filter_by(collection_id=...) -> filter_by(indexed=False) -> all()
|
|
mock_session.query.return_value.filter_by.return_value.filter_by.return_value.all.return_value = [
|
|
mock_dc1,
|
|
mock_dc2,
|
|
]
|
|
# query(Document).filter_by(id=...).first() for title lookup
|
|
mock_session.query.return_value.filter_by.return_value.first.return_value = mock_doc
|
|
|
|
# Mock index_document to return success
|
|
svc.index_document = MagicMock(
|
|
return_value={"status": "success", "chunk_count": 10}
|
|
)
|
|
|
|
callback = MagicMock()
|
|
svc.index_all_documents("coll-1", progress_callback=callback)
|
|
|
|
assert callback.call_count == 2
|
|
# Check the callback was called with (idx, total, title, status)
|
|
first_call = callback.call_args_list[0]
|
|
assert first_call[0][0] == 1 # idx
|
|
assert first_call[0][1] == 2 # total
|
|
assert first_call[0][3] == "success" # status
|
|
|
|
@patch(f"{_MOD}.get_user_db_session")
|
|
def test_counts_successful_skipped_failed(self, mock_session_ctx):
|
|
svc = _make_service()
|
|
mock_session = MagicMock()
|
|
mock_session_ctx.return_value.__enter__ = MagicMock(
|
|
return_value=mock_session
|
|
)
|
|
mock_session_ctx.return_value.__exit__ = MagicMock(return_value=None)
|
|
|
|
mock_dc1 = MagicMock()
|
|
mock_dc1.document_id = "doc-1"
|
|
mock_dc2 = MagicMock()
|
|
mock_dc2.document_id = "doc-2"
|
|
mock_dc3 = MagicMock()
|
|
mock_dc3.document_id = "doc-3"
|
|
|
|
mock_doc = MagicMock()
|
|
mock_doc.title = "Title"
|
|
|
|
# With force_reindex=True the code does:
|
|
# query(DocumentCollection).filter_by(collection_id=...).all()
|
|
# (no second filter_by for indexed=False)
|
|
mock_query = MagicMock()
|
|
mock_query.filter_by.return_value.all.return_value = [
|
|
mock_dc1,
|
|
mock_dc2,
|
|
mock_dc3,
|
|
]
|
|
# Document title lookup
|
|
mock_query.filter_by.return_value.first.return_value = mock_doc
|
|
mock_session.query.return_value = mock_query
|
|
|
|
results_sequence = [
|
|
{"status": "success", "chunk_count": 5},
|
|
{
|
|
"status": "skipped",
|
|
"message": "already indexed",
|
|
"chunk_count": 3,
|
|
},
|
|
{"status": "error", "error": "something broke"},
|
|
]
|
|
svc.index_document = MagicMock(side_effect=results_sequence)
|
|
|
|
result = svc.index_all_documents("coll-1", force_reindex=True)
|
|
assert result["successful"] == 1
|
|
assert result["skipped"] == 1
|
|
assert result["failed"] == 1
|
|
assert len(result["errors"]) == 1
|
|
|
|
|
|
# =========================================================================
|
|
# db_password property
|
|
# =========================================================================
|
|
class TestDbPasswordProperty:
|
|
"""Tests for the db_password property propagation."""
|
|
|
|
def test_getter_returns_value(self):
|
|
svc = _make_service()
|
|
svc._db_password = "secret"
|
|
assert svc.db_password == "secret"
|
|
|
|
def test_setter_propagates_to_embedding_manager(self):
|
|
svc = _make_service()
|
|
svc.embedding_manager = MagicMock()
|
|
svc.integrity_manager = MagicMock()
|
|
svc.db_password = "new_pw"
|
|
assert svc.embedding_manager.db_password == "new_pw"
|
|
|
|
def test_setter_propagates_to_integrity_manager(self):
|
|
svc = _make_service()
|
|
svc.embedding_manager = MagicMock()
|
|
svc.integrity_manager = MagicMock()
|
|
svc.db_password = "new_pw"
|
|
assert svc.integrity_manager.password == "new_pw"
|
|
|
|
def test_setter_handles_none_managers(self):
|
|
svc = _make_service()
|
|
svc.embedding_manager = None
|
|
svc.integrity_manager = None
|
|
# Should not raise
|
|
svc.db_password = "pw"
|
|
assert svc._db_password == "pw"
|
|
|
|
|
|
# =========================================================================
|
|
# Context manager / close
|
|
# =========================================================================
|
|
class TestContextManager:
|
|
"""Tests for context manager protocol."""
|
|
|
|
def test_enter_returns_self(self):
|
|
svc = _make_service()
|
|
assert svc.__enter__() is svc
|
|
|
|
def test_exit_calls_close(self):
|
|
svc = _make_service()
|
|
svc.close = MagicMock()
|
|
svc.__exit__(None, None, None)
|
|
svc.close.assert_called_once()
|
|
|
|
def test_exit_returns_false(self):
|
|
svc = _make_service()
|
|
result = svc.__exit__(None, None, None)
|
|
assert result is False
|
|
|
|
|
|
# =========================================================================
|
|
# Quarantine helper + concurrent-write locks (#4197)
|
|
# =========================================================================
|
|
|
|
|
|
def _import_module():
|
|
import local_deep_research.research_library.services.library_rag_service as m
|
|
|
|
return m
|
|
|
|
|
|
class TestCorruptionQuarantine:
|
|
"""Verify the verify-failure and load-failure paths preserve
|
|
corrupted bytes instead of deleting them.
|
|
"""
|
|
|
|
def _make_idx(self, tmp_path, name="idx"):
|
|
idx = tmp_path / f"{name}.faiss"
|
|
pkl = tmp_path / f"{name}.pkl"
|
|
idx.write_bytes(b"faiss-bytes")
|
|
pkl.write_bytes(b"pkl-bytes")
|
|
return idx, pkl
|
|
|
|
def test_renames_both_faiss_and_pkl(self, tmp_path):
|
|
svc = _make_service()
|
|
idx, pkl = self._make_idx(tmp_path)
|
|
svc._quarantine_corrupt_index(idx, "test")
|
|
# Original files are gone from their paths
|
|
assert not idx.exists()
|
|
assert not pkl.exists()
|
|
# Both quarantined as .corrupt-*
|
|
corrupted_faiss = list(tmp_path.glob("idx.faiss.corrupt-*"))
|
|
corrupted_pkl = list(tmp_path.glob("idx.pkl.corrupt-*"))
|
|
assert len(corrupted_faiss) == 1
|
|
assert len(corrupted_pkl) == 1
|
|
assert corrupted_faiss[0].read_bytes() == b"faiss-bytes"
|
|
assert corrupted_pkl[0].read_bytes() == b"pkl-bytes"
|
|
|
|
def test_skips_pkl_when_missing(self, tmp_path):
|
|
svc = _make_service()
|
|
idx = tmp_path / "idx.faiss"
|
|
idx.write_bytes(b"faiss-bytes")
|
|
# No .pkl present
|
|
svc._quarantine_corrupt_index(idx, "test")
|
|
assert not idx.exists()
|
|
corrupted = list(tmp_path.glob("idx.faiss.corrupt-*"))
|
|
assert len(corrupted) == 1
|
|
# No orphan pkl created
|
|
assert not any(tmp_path.glob("idx.pkl.corrupt-*"))
|
|
|
|
def test_rename_oserror_re_raises(self, tmp_path):
|
|
"""Disk-full / permission failure must propagate, not be swallowed."""
|
|
svc = _make_service()
|
|
idx, _pkl = self._make_idx(tmp_path)
|
|
with patch.object(Path, "rename", side_effect=OSError("EROFS")):
|
|
with pytest.raises(OSError, match="EROFS"):
|
|
svc._quarantine_corrupt_index(idx, "test")
|
|
|
|
def test_collision_increments_suffix(self, tmp_path):
|
|
"""Pre-existing quarantine path → loop increments to -1, -2, ..."""
|
|
svc = _make_service()
|
|
idx, pkl = self._make_idx(tmp_path)
|
|
# Force a specific timestamp so we can predict collision targets
|
|
with patch.object(
|
|
_import_module().time,
|
|
"time_ns",
|
|
return_value=12345,
|
|
):
|
|
# Pre-create the base collision target
|
|
(tmp_path / "idx.faiss.corrupt-12345").write_bytes(b"old1")
|
|
svc._quarantine_corrupt_index(idx, "test")
|
|
# Original moved out
|
|
assert not idx.exists()
|
|
# New corrupt-12345-1 created (the -1 increment)
|
|
moved = tmp_path / "idx.faiss.corrupt-12345-1"
|
|
assert moved.exists()
|
|
assert moved.read_bytes() == b"faiss-bytes"
|
|
|
|
def test_collision_cap_raises(self, tmp_path):
|
|
"""If 32 collisions in a row, surface an OSError, don't hang."""
|
|
svc = _make_service()
|
|
mod = _import_module()
|
|
idx, pkl = self._make_idx(tmp_path)
|
|
with patch.object(mod.time, "time_ns", return_value=999):
|
|
# Pre-create base + 32 collisions
|
|
(tmp_path / "idx.faiss.corrupt-999").write_bytes(b"")
|
|
for n in range(1, mod._QUARANTINE_SUFFIX_RETRY_CAP + 1):
|
|
(tmp_path / f"idx.faiss.corrupt-999-{n}").write_bytes(b"")
|
|
with pytest.raises(OSError, match="collisions exceeded"):
|
|
svc._quarantine_corrupt_index(idx, "test")
|
|
|
|
|
|
class TestQuarantineRetention:
|
|
"""Retention sweep keeps the rag_indices/ directory from filling
|
|
with old .corrupt-* files on systems that experience recurring
|
|
corruption.
|
|
"""
|
|
|
|
def _make_idx(self, tmp_path, name="idx"):
|
|
idx = tmp_path / f"{name}.faiss"
|
|
pkl = tmp_path / f"{name}.pkl"
|
|
idx.write_bytes(b"faiss-bytes")
|
|
pkl.write_bytes(b"pkl-bytes")
|
|
return idx, pkl
|
|
|
|
def _make_quarantined(self, tmp_path, name, ns, mtime):
|
|
"""Create a paired (.faiss, .pkl) quarantined-style file at
|
|
the given ns suffix and stamp it with ``mtime`` so the
|
|
retention sweep's age sort is deterministic.
|
|
"""
|
|
faiss_f = tmp_path / f"{name}.faiss.corrupt-{ns}"
|
|
pkl_f = tmp_path / f"{name}.pkl.corrupt-{ns}"
|
|
faiss_f.write_bytes(b"")
|
|
pkl_f.write_bytes(b"")
|
|
import os
|
|
|
|
os.utime(faiss_f, (mtime, mtime))
|
|
os.utime(pkl_f, (mtime, mtime))
|
|
return faiss_f, pkl_f
|
|
|
|
def test_prune_keeps_most_recent_drops_older(self, tmp_path):
|
|
"""After N+2 simulated past corruption events, pruning leaves
|
|
exactly N (most recent by mtime) and removes the oldest 2.
|
|
"""
|
|
svc = _make_service()
|
|
mod = _import_module()
|
|
keep = mod._QUARANTINE_KEEP_RECENT
|
|
|
|
# Pre-seed keep+2 quarantined pairs with increasing mtimes
|
|
# (later ns → newer mtime).
|
|
for i in range(keep + 2):
|
|
self._make_quarantined(
|
|
tmp_path, "idx", ns=1000 + i, mtime=1000.0 + i
|
|
)
|
|
|
|
# Trigger pruning via the public quarantine call (which now
|
|
# invokes the sweep at the end).
|
|
idx, _pkl = self._make_idx(tmp_path)
|
|
svc._quarantine_corrupt_index(idx, "test")
|
|
|
|
faiss_corrupts = sorted(tmp_path.glob("idx.faiss.corrupt-*"))
|
|
pkl_corrupts = sorted(tmp_path.glob("idx.pkl.corrupt-*"))
|
|
# keep pre-existing + 1 fresh = keep total on each side
|
|
assert len(faiss_corrupts) == keep
|
|
assert len(pkl_corrupts) == keep
|
|
|
|
# Oldest two pre-existing pairs (ns 1000, 1001) must be gone.
|
|
for stale_ns in (1000, 1001):
|
|
assert not (tmp_path / f"idx.faiss.corrupt-{stale_ns}").exists()
|
|
assert not (tmp_path / f"idx.pkl.corrupt-{stale_ns}").exists()
|
|
|
|
def test_prune_skipped_when_under_threshold(self, tmp_path):
|
|
"""With fewer than the retention cap, no pruning happens."""
|
|
svc = _make_service()
|
|
mod = _import_module()
|
|
keep = mod._QUARANTINE_KEEP_RECENT
|
|
|
|
# Pre-seed (keep - 2) quarantined pairs. After this call we'll
|
|
# add 1 more via the quarantine, ending at keep - 1 — still
|
|
# under the cap.
|
|
for i in range(keep - 2):
|
|
self._make_quarantined(
|
|
tmp_path, "idx", ns=2000 + i, mtime=2000.0 + i
|
|
)
|
|
|
|
idx, _pkl = self._make_idx(tmp_path)
|
|
svc._quarantine_corrupt_index(idx, "test")
|
|
|
|
# All pre-existing + the new one survive.
|
|
assert len(list(tmp_path.glob("idx.faiss.corrupt-*"))) == keep - 1
|
|
|
|
def test_prune_only_targets_same_base(self, tmp_path):
|
|
"""Pruning of ``idx.faiss.corrupt-*`` must NOT touch
|
|
``other.faiss.corrupt-*`` — different base paths are
|
|
independent.
|
|
"""
|
|
svc = _make_service()
|
|
mod = _import_module()
|
|
keep = mod._QUARANTINE_KEEP_RECENT
|
|
|
|
# Many old corrupts for the other base — these MUST be untouched.
|
|
for i in range(keep + 3):
|
|
self._make_quarantined(
|
|
tmp_path, "other", ns=3000 + i, mtime=3000.0 + i
|
|
)
|
|
# Trigger many corrupts for our base so the sweep runs.
|
|
for i in range(keep + 2):
|
|
self._make_quarantined(
|
|
tmp_path, "idx", ns=4000 + i, mtime=4000.0 + i
|
|
)
|
|
|
|
idx, _pkl = self._make_idx(tmp_path)
|
|
svc._quarantine_corrupt_index(idx, "test")
|
|
|
|
# Other base untouched
|
|
other_corrupts = list(tmp_path.glob("other.faiss.corrupt-*"))
|
|
assert len(other_corrupts) == keep + 3
|
|
# Our base pruned
|
|
idx_corrupts = list(tmp_path.glob("idx.faiss.corrupt-*"))
|
|
assert len(idx_corrupts) == keep
|
|
|
|
def test_prune_failure_does_not_propagate(self, tmp_path):
|
|
"""If unlink raises during the sweep, the quarantine call
|
|
still succeeds — retention is best-effort.
|
|
"""
|
|
svc = _make_service()
|
|
mod = _import_module()
|
|
keep = mod._QUARANTINE_KEEP_RECENT
|
|
|
|
# Seed enough corrupts to trigger pruning.
|
|
for i in range(keep + 2):
|
|
self._make_quarantined(
|
|
tmp_path, "idx", ns=5000 + i, mtime=5000.0 + i
|
|
)
|
|
|
|
before_count = len(list(tmp_path.glob("idx.faiss.corrupt-*")))
|
|
idx, _pkl = self._make_idx(tmp_path)
|
|
|
|
# Allow the quarantine rename to succeed, but make unlink raise
|
|
# for the prune sweep. The quarantine itself uses rename, not
|
|
# unlink, so this only affects the retention path.
|
|
real_unlink = Path.unlink
|
|
|
|
def selective_unlink(self):
|
|
if ".corrupt-" in self.name:
|
|
raise PermissionError(f"denied: {self}")
|
|
return real_unlink(self)
|
|
|
|
with patch.object(Path, "unlink", selective_unlink):
|
|
# Must not raise — sweep is best-effort.
|
|
svc._quarantine_corrupt_index(idx, "test")
|
|
|
|
# Quarantine still produced a new pair even though the sweep
|
|
# was unable to delete anything (count went up, not down).
|
|
after_count = len(list(tmp_path.glob("idx.faiss.corrupt-*")))
|
|
assert after_count == before_count + 1, (
|
|
f"expected one new corrupt file; before={before_count}, "
|
|
f"after={after_count}"
|
|
)
|
|
|
|
def test_retention_sort_uses_filename_ns_not_mtime(self, tmp_path):
|
|
"""On low-resolution filesystems (FAT32, SMB, ext3) mtime can
|
|
round to whole seconds, making same-second quarantines sort
|
|
non-deterministically by ``st_mtime``. The embedded
|
|
``-<ns>`` suffix is the authoritative ordering — confirm
|
|
retention drops the file with the *lowest* ns even when
|
|
its mtime would mark it as the *newest*.
|
|
"""
|
|
svc = _make_service()
|
|
mod = _import_module()
|
|
keep = mod._QUARANTINE_KEEP_RECENT
|
|
import os
|
|
|
|
# Seed exactly `keep` corrupts. The OLDEST ns gets the NEWEST
|
|
# mtime; any retention that sorts by mtime would keep it. A
|
|
# sort by ns must prune it once one more quarantine pushes us
|
|
# over the cap.
|
|
ns_values = list(range(6000, 6000 + keep))
|
|
mtimes = list(reversed([7000.0 + i for i in range(keep)]))
|
|
for ns, mtime in zip(ns_values, mtimes):
|
|
self._make_quarantined(tmp_path, "idx", ns=ns, mtime=mtime)
|
|
|
|
# Add one more quarantine via the real helper. Its ns is
|
|
# `time.time_ns()` which is far larger than 6000+, so by ns
|
|
# it's the freshest. Force its mtime to ancient so any
|
|
# mtime-based sort would prune *it* instead of the seeded
|
|
# lowest-ns file — that's the regression this test guards.
|
|
idx, _pkl = self._make_idx(tmp_path)
|
|
original_rename = Path.rename
|
|
|
|
def rename_then_age(self, target):
|
|
result = original_rename(self, target)
|
|
os.utime(target, (1.0, 1.0))
|
|
return result
|
|
|
|
with patch.object(Path, "rename", rename_then_age):
|
|
svc._quarantine_corrupt_index(idx, "test")
|
|
|
|
# The lowest-ns seeded file (ns=6000) must be the one pruned.
|
|
# The other seeded files (ns=6001..6004) plus the brand-new
|
|
# one (highest ns) survive.
|
|
assert not (tmp_path / f"idx.faiss.corrupt-{ns_values[0]}").exists()
|
|
for ns in ns_values[1:]:
|
|
assert (tmp_path / f"idx.faiss.corrupt-{ns}").exists(), (
|
|
f"ns={ns} should still exist after retention sort"
|
|
)
|
|
|
|
|
|
class TestFaissWriteLock:
|
|
"""Lock infrastructure: identity, parallelism across keys,
|
|
cleanup on user-close.
|
|
"""
|
|
|
|
def _reset_locks(self):
|
|
mod = _import_module()
|
|
with mod._faiss_write_locks_lock:
|
|
mod._faiss_write_locks.clear()
|
|
|
|
def test_same_key_returns_same_lock(self, tmp_path):
|
|
self._reset_locks()
|
|
mod = _import_module()
|
|
p = str(tmp_path / "shared.faiss")
|
|
a = mod._get_faiss_write_lock("u1", p)
|
|
b = mod._get_faiss_write_lock("u1", p)
|
|
assert a is b
|
|
|
|
def test_different_key_returns_different_lock(self, tmp_path):
|
|
self._reset_locks()
|
|
mod = _import_module()
|
|
a = mod._get_faiss_write_lock("u1", str(tmp_path / "a.faiss"))
|
|
b = mod._get_faiss_write_lock("u1", str(tmp_path / "b.faiss"))
|
|
c = mod._get_faiss_write_lock("u2", str(tmp_path / "a.faiss"))
|
|
assert a is not b
|
|
assert a is not c
|
|
|
|
def test_pop_removes_only_target_user(self, tmp_path):
|
|
self._reset_locks()
|
|
mod = _import_module()
|
|
u1_lock = mod._get_faiss_write_lock("u1", str(tmp_path / "a.faiss"))
|
|
mod._get_faiss_write_lock("u1", str(tmp_path / "b.faiss"))
|
|
u2_lock = mod._get_faiss_write_lock("u2", str(tmp_path / "a.faiss"))
|
|
|
|
mod.pop_faiss_locks_for_user("u1")
|
|
|
|
# u1 locks gone (new lookup yields a fresh lock object)
|
|
new_u1 = mod._get_faiss_write_lock("u1", str(tmp_path / "a.faiss"))
|
|
assert new_u1 is not u1_lock
|
|
# u2 lock untouched
|
|
same_u2 = mod._get_faiss_write_lock("u2", str(tmp_path / "a.faiss"))
|
|
assert same_u2 is u2_lock
|
|
|
|
def test_concurrent_holders_serialised_for_same_key(self, tmp_path):
|
|
"""Two threads acquiring ``_get_faiss_write_lock`` for the same
|
|
``(username, index_path)`` MUST observe non-overlapping critical
|
|
sections. Regression for the #4197 race.
|
|
"""
|
|
self._reset_locks()
|
|
mod = _import_module()
|
|
index_path = tmp_path / "idx.faiss"
|
|
|
|
in_flight = 0
|
|
max_in_flight = 0
|
|
observed_overlap = False
|
|
lock_for_counters = threading.Lock()
|
|
first_inside = threading.Event()
|
|
first_can_finish = threading.Event()
|
|
first_holder = [True] # mutable flag to mark which thread enters first
|
|
|
|
def worker():
|
|
with mod._get_faiss_write_lock("u1", str(index_path)):
|
|
nonlocal in_flight, max_in_flight, observed_overlap
|
|
with lock_for_counters:
|
|
in_flight += 1
|
|
if in_flight > max_in_flight:
|
|
max_in_flight = in_flight
|
|
if in_flight > 1:
|
|
observed_overlap = True
|
|
am_first = first_holder[0]
|
|
first_holder[0] = False
|
|
if am_first:
|
|
first_inside.set()
|
|
# Hold the critical section open so the second
|
|
# thread blocks behind us, not races us.
|
|
first_can_finish.wait(timeout=2.0)
|
|
with lock_for_counters:
|
|
in_flight -= 1
|
|
|
|
t1 = threading.Thread(target=worker)
|
|
t2 = threading.Thread(target=worker)
|
|
t1.start()
|
|
assert first_inside.wait(timeout=2.0)
|
|
t2.start()
|
|
first_can_finish.set()
|
|
t1.join(timeout=3.0)
|
|
t2.join(timeout=3.0)
|
|
|
|
assert not observed_overlap, (
|
|
"Two threads held the same FAISS write lock concurrently."
|
|
)
|
|
assert max_in_flight == 1
|
|
|
|
def test_different_keys_run_in_parallel(self, tmp_path):
|
|
"""Two threads acquiring locks for different
|
|
``(username, index_path)`` keys must run simultaneously.
|
|
"""
|
|
self._reset_locks()
|
|
mod = _import_module()
|
|
barrier = threading.Barrier(2, timeout=2.0)
|
|
idx1 = tmp_path / "a.faiss"
|
|
idx2 = tmp_path / "b.faiss"
|
|
|
|
def worker(index_path):
|
|
with mod._get_faiss_write_lock("u1", str(index_path)):
|
|
# If the lock blocked us, barrier.wait would time out
|
|
barrier.wait()
|
|
|
|
t1 = threading.Thread(target=worker, args=(idx1,))
|
|
t2 = threading.Thread(target=worker, args=(idx2,))
|
|
t1.start()
|
|
t2.start()
|
|
t1.join(timeout=3.0)
|
|
t2.join(timeout=3.0)
|
|
# If barrier broke (timeout), one of the threads would be alive
|
|
assert not t1.is_alive()
|
|
assert not t2.is_alive()
|
|
|
|
|
|
class TestNoWalCheckpointInIndexingPath:
|
|
"""Regression guard: the per-document PRAGMA wal_checkpoint(FULL)
|
|
has been removed (#4197 secondary fix from PR #3539). Re-adding it
|
|
would re-introduce 'database is locked' errors under concurrent
|
|
bulk indexing.
|
|
"""
|
|
|
|
def test_module_source_has_no_executable_wal_checkpoint(self):
|
|
import inspect
|
|
|
|
mod = _import_module()
|
|
src = inspect.getsource(mod)
|
|
# Strip out comment / docstring lines so the regression test
|
|
# tolerates explanatory references like 'PRAGMA wal_checkpoint'
|
|
# in comments, but fails if any code line re-adds the call.
|
|
for lineno, line in enumerate(src.splitlines(), 1):
|
|
stripped = line.strip()
|
|
if not stripped:
|
|
continue
|
|
if stripped.startswith("#"):
|
|
continue
|
|
assert "wal_checkpoint(" not in stripped, (
|
|
f"library_rag_service.py:{lineno} re-introduced "
|
|
f"a wal_checkpoint() call: {line!r}"
|
|
)
|
|
|
|
|
|
class TestMergeAndPersistLocked:
|
|
"""Read-modify-write under the FAISS write lock.
|
|
|
|
Regression for the AI-review concern on #4200: two workers
|
|
indexing different documents into the same collection used to
|
|
both load on-disk state X, each add their own chunk in memory,
|
|
then save in sequence — last writer wins, the loser's chunks
|
|
were lost from the FAISS file (chunks survived in the DB so a
|
|
rebuild recovered them, but the index file was wrong until then).
|
|
|
|
The helper reloads from disk under the lock before adding, so
|
|
the second writer sees the first writer's save and merges
|
|
instead of overwriting.
|
|
"""
|
|
|
|
def _reset_locks(self):
|
|
mod = _import_module()
|
|
with mod._faiss_write_locks_lock:
|
|
mod._faiss_write_locks.clear()
|
|
|
|
def test_reload_picks_up_concurrent_writers_save(self, tmp_path):
|
|
"""The merge helper must reload from disk before adding so
|
|
chunks committed by a concurrent writer between the caller's
|
|
in-memory load and this lock acquisition are preserved.
|
|
"""
|
|
self._reset_locks()
|
|
mod = _import_module()
|
|
index_path = tmp_path / "idx.faiss"
|
|
# The on-disk file must exist + verify so the helper actually
|
|
# takes the reload branch (not "no on-disk version, keep
|
|
# stale in-memory").
|
|
index_path.touch()
|
|
|
|
svc = _make_service()
|
|
svc.embedding_manager = MagicMock()
|
|
svc.integrity_manager = MagicMock()
|
|
svc.integrity_manager.verify_file.return_value = (True, None)
|
|
svc.rag_index_record = MagicMock(id=42)
|
|
|
|
# Caller's stale in-memory state knows about no chunks.
|
|
stale_index = MagicMock()
|
|
stale_index.docstore._dict = {}
|
|
svc.faiss_index = stale_index
|
|
|
|
# Disk state (loaded by the reload step) already has another
|
|
# writer's chunk "concurrent-id".
|
|
fresh_index = MagicMock()
|
|
fresh_index.docstore._dict = {"concurrent-id": MagicMock()}
|
|
|
|
new_chunk = LangchainDocument(page_content="our new content")
|
|
|
|
with patch.object(mod, "safe_load_faiss", return_value=fresh_index):
|
|
stats = svc._merge_and_persist_locked(
|
|
index_path,
|
|
[new_chunk],
|
|
["our-id"],
|
|
force_reindex=False,
|
|
)
|
|
|
|
# Helper reloaded → svc.faiss_index is now the fresh state
|
|
assert svc.faiss_index is fresh_index
|
|
# Our chunk was added to the fresh state (not the stale one)
|
|
fresh_index.add_documents.assert_called_once()
|
|
added_ids = fresh_index.add_documents.call_args[1]["ids"]
|
|
assert added_ids == ["our-id"]
|
|
# The stale state was NOT mutated (otherwise we'd be saving it)
|
|
stale_index.add_documents.assert_not_called()
|
|
# save_local was called on the fresh (merged) state
|
|
fresh_index.save_local.assert_called_once()
|
|
assert stats["added"] == 1
|
|
assert stats["added_ids"] == ["our-id"]
|
|
|
|
def test_reload_skips_when_existing_id_now_on_disk(self, tmp_path):
|
|
"""If the caller wanted to add id X but a concurrent writer
|
|
already saved X to disk, dedup against the fresh state must
|
|
drop X (idempotent re-index).
|
|
"""
|
|
self._reset_locks()
|
|
mod = _import_module()
|
|
index_path = tmp_path / "idx.faiss"
|
|
index_path.touch()
|
|
|
|
svc = _make_service()
|
|
svc.embedding_manager = MagicMock()
|
|
svc.integrity_manager = MagicMock()
|
|
svc.integrity_manager.verify_file.return_value = (True, None)
|
|
svc.rag_index_record = MagicMock(id=42)
|
|
svc.faiss_index = MagicMock() # stale, irrelevant
|
|
|
|
# Disk state already has the same id we're about to add.
|
|
fresh_index = MagicMock()
|
|
fresh_index.docstore._dict = {"shared-id": MagicMock()}
|
|
|
|
with patch.object(mod, "safe_load_faiss", return_value=fresh_index):
|
|
stats = svc._merge_and_persist_locked(
|
|
index_path,
|
|
[LangchainDocument(page_content="duplicate")],
|
|
["shared-id"],
|
|
force_reindex=False,
|
|
)
|
|
|
|
# No add (already on disk); save still happens (idempotent +
|
|
# touches mtime so the integrity record updates).
|
|
fresh_index.add_documents.assert_not_called()
|
|
fresh_index.save_local.assert_called_once()
|
|
assert stats["added"] == 0
|
|
assert stats["skipped"] == 1
|
|
|
|
def test_force_reindex_deletes_after_reload(self, tmp_path):
|
|
"""force_reindex deletes the matching IDs from the FRESH
|
|
on-disk state (not the caller's stale in-memory state)
|
|
before re-adding.
|
|
"""
|
|
self._reset_locks()
|
|
mod = _import_module()
|
|
index_path = tmp_path / "idx.faiss"
|
|
index_path.touch()
|
|
|
|
svc = _make_service()
|
|
svc.embedding_manager = MagicMock()
|
|
svc.integrity_manager = MagicMock()
|
|
svc.integrity_manager.verify_file.return_value = (True, None)
|
|
svc.rag_index_record = MagicMock(id=42)
|
|
svc.faiss_index = MagicMock() # stale
|
|
|
|
fresh_index = MagicMock()
|
|
fresh_index.docstore._dict = {"to-update": MagicMock()}
|
|
|
|
with patch.object(mod, "safe_load_faiss", return_value=fresh_index):
|
|
svc._merge_and_persist_locked(
|
|
index_path,
|
|
[LangchainDocument(page_content="updated")],
|
|
["to-update"],
|
|
force_reindex=True,
|
|
)
|
|
|
|
# Old copy removed from FRESH state
|
|
fresh_index.delete.assert_called_once_with(["to-update"])
|
|
# New copy added (no dedup under force_reindex)
|
|
fresh_index.add_documents.assert_called_once()
|
|
added_ids = fresh_index.add_documents.call_args[1]["ids"]
|
|
assert added_ids == ["to-update"]
|
|
|
|
def test_reload_failure_falls_back_to_in_memory(self, tmp_path):
|
|
"""If the loader raises (torn write, partial pickle),
|
|
the merge helper keeps the caller's in-memory state instead
|
|
of losing the write entirely.
|
|
"""
|
|
self._reset_locks()
|
|
mod = _import_module()
|
|
index_path = tmp_path / "idx.faiss"
|
|
index_path.touch()
|
|
|
|
svc = _make_service()
|
|
svc.embedding_manager = MagicMock()
|
|
svc.integrity_manager = MagicMock()
|
|
svc.integrity_manager.verify_file.return_value = (True, None)
|
|
svc.rag_index_record = MagicMock(id=42)
|
|
|
|
in_memory = MagicMock()
|
|
in_memory.docstore._dict = {}
|
|
svc.faiss_index = in_memory
|
|
|
|
with patch.object(
|
|
mod, "safe_load_faiss", side_effect=RuntimeError("torn pickle")
|
|
):
|
|
svc._merge_and_persist_locked(
|
|
index_path,
|
|
[LangchainDocument(page_content="kept")],
|
|
["kept-id"],
|
|
force_reindex=False,
|
|
)
|
|
|
|
# Reload failed → kept the in-memory object, added to it,
|
|
# saved it. The write is preserved.
|
|
assert svc.faiss_index is in_memory
|
|
in_memory.add_documents.assert_called_once()
|
|
in_memory.save_local.assert_called_once()
|
|
|
|
def test_concurrent_writers_both_chunks_survive(self, tmp_path):
|
|
"""End-to-end regression: two workers each call
|
|
``_merge_and_persist_locked`` for the same on-disk index.
|
|
Each starts from a stale in-memory snapshot. After both
|
|
complete, the on-disk state must contain BOTH chunks.
|
|
|
|
Before the read-modify-write fix, this was last-writer-wins:
|
|
worker A's chunk was overwritten by worker B's save because
|
|
B never reloaded A's write.
|
|
"""
|
|
self._reset_locks()
|
|
mod = _import_module()
|
|
index_path = tmp_path / "shared.faiss"
|
|
index_path.touch()
|
|
|
|
# Single shared "disk state" as a dict — proxies for what
|
|
# would be on disk via FAISS.load_local/save_local.
|
|
disk_state: dict = {}
|
|
disk_lock = threading.Lock()
|
|
|
|
def make_index_proxy():
|
|
"""Each call returns a FAISS-like proxy reading/writing
|
|
the shared disk_state. Mimics a fresh load_local()."""
|
|
ix = MagicMock()
|
|
# Snapshot disk_state at load time (proxy for what
|
|
# FAISS.load_local would return).
|
|
snapshot = dict(disk_state)
|
|
ix.docstore = MagicMock()
|
|
ix.docstore._dict = snapshot
|
|
|
|
def add(chunks, ids):
|
|
for cid in ids:
|
|
snapshot[cid] = "doc"
|
|
|
|
def save(_folder, index_name=None):
|
|
# Atomically replace disk_state with this index's
|
|
# current snapshot — mirrors save_local's truncating
|
|
# semantics for the test.
|
|
with disk_lock:
|
|
disk_state.clear()
|
|
disk_state.update(snapshot)
|
|
|
|
ix.add_documents.side_effect = add
|
|
ix.save_local.side_effect = save
|
|
return ix
|
|
|
|
def worker(cid: str, start_event: threading.Event):
|
|
svc = _make_service()
|
|
svc.embedding_manager = MagicMock()
|
|
svc.integrity_manager = MagicMock()
|
|
svc.integrity_manager.verify_file.return_value = (True, None)
|
|
svc.rag_index_record = MagicMock(id=99)
|
|
# Pre-merge in-memory state is stale (empty)
|
|
svc.faiss_index = MagicMock()
|
|
svc.faiss_index.docstore = MagicMock()
|
|
svc.faiss_index.docstore._dict = {}
|
|
start_event.wait() # Race start
|
|
svc._merge_and_persist_locked(
|
|
index_path,
|
|
[LangchainDocument(page_content=f"chunk-{cid}")],
|
|
[cid],
|
|
force_reindex=False,
|
|
)
|
|
|
|
go = threading.Event()
|
|
t1 = threading.Thread(target=worker, args=("A", go))
|
|
t2 = threading.Thread(target=worker, args=("B", go))
|
|
|
|
# Patch safe_load_faiss ONCE, around both threads. Patching
|
|
# inside each worker (the original shape) is a hidden race:
|
|
# ``patch.object`` rewrites a module attribute and is not
|
|
# thread-safe — when the first worker's
|
|
# ``with patch.object(...)`` block exits, it restores whatever
|
|
# it captured as "the original" at entry, which can be the
|
|
# second worker's lambda OR the real ``safe_load_faiss``
|
|
# (depending on which thread entered patch first). If the real
|
|
# ``safe_load_faiss`` becomes active mid-test, the second worker's
|
|
# reload step raises on the empty touched ``shared.faiss``,
|
|
# the production code's ``except`` branch falls back to the
|
|
# caller's stale in-memory MagicMock, and the worker's save is
|
|
# a no-op (the setup MagicMock has no ``save_local`` side
|
|
# effect). Result: that worker's chunk never reaches
|
|
# ``disk_state``, the assertion fails, and the test wrongly
|
|
# blames a regression in the production lock. Applying the
|
|
# patch around both threads removes the race entirely — the
|
|
# patched value is the only value visible while either worker
|
|
# runs.
|
|
with patch.object(
|
|
mod,
|
|
"safe_load_faiss",
|
|
side_effect=lambda *a, **kw: make_index_proxy(),
|
|
):
|
|
t1.start()
|
|
t2.start()
|
|
go.set()
|
|
t1.join(timeout=5.0)
|
|
t2.join(timeout=5.0)
|
|
|
|
# Both writers' chunks present — read-modify-write under the
|
|
# lock did its job. Without the fix, only one of {A, B} would
|
|
# remain.
|
|
assert "A" in disk_state, (
|
|
f"Worker A's chunk lost; disk_state={disk_state}. "
|
|
f"Read-modify-write race regressed."
|
|
)
|
|
assert "B" in disk_state, (
|
|
f"Worker B's chunk lost; disk_state={disk_state}. "
|
|
f"Read-modify-write race regressed."
|
|
)
|