wehub-resource-sync
e4dcfc49aa
Tests / Lint and Format (push) Waiting to run
Tests / Web Node Tests (push) Waiting to run
Tests / Import Check (Python 3.11) (push) Waiting to run
Tests / Import Check (Python 3.12) (push) Waiting to run
Tests / Import Check (Python 3.13) (push) Waiting to run
Tests / Import Check (Python 3.14) (push) Waiting to run
Tests / Python Tests (Python 3.11) (push) Blocked by required conditions
Tests / Python Tests (Python 3.12) (push) Blocked by required conditions
Tests / Python Tests (Python 3.13) (push) Blocked by required conditions
Tests / Python Tests (Python 3.14) (push) Blocked by required conditions
Tests / Test Summary (push) Blocked by required conditions
106 lines
4.1 KiB
Python
106 lines
4.1 KiB
Python
"""Manager handling of linked KBs (``type: linked`` engine-index pointers).
|
|
|
|
A linked KB mounts a pre-built index in place: no on-disk folder under
|
|
``base_dir``, read-only, and — critically — deleting it must never touch the
|
|
user's external files. Mirrors the Obsidian pointer guarantees but for an
|
|
engine-backed index queried via its bound ``rag_provider``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from deeptutor.knowledge.manager import KnowledgeBaseManager
|
|
|
|
|
|
def _external_index(tmp_path: Path) -> Path:
|
|
"""A self-contained external folder holding a ready LlamaIndex version."""
|
|
ext = tmp_path / "external_kb"
|
|
version = ext / "version-1"
|
|
version.mkdir(parents=True)
|
|
(version / "docstore.json").write_text("{}", encoding="utf-8")
|
|
(version / "index_store.json").write_text("{}", encoding="utf-8")
|
|
(version / "meta.json").write_text(
|
|
json.dumps({"version": "version-1", "signature": "abc", "layout": "flat"}),
|
|
encoding="utf-8",
|
|
)
|
|
return ext
|
|
|
|
|
|
def test_register_linked_kb_writes_pointer(tmp_path: Path) -> None:
|
|
ext = _external_index(tmp_path)
|
|
manager = KnowledgeBaseManager(base_dir=str(tmp_path / "kbs"))
|
|
|
|
entry = manager.register_linked_kb("Linked", str(ext), "llamaindex", stats={"doc_count": 5})
|
|
|
|
assert entry["type"] == "linked"
|
|
assert entry["rag_provider"] == "llamaindex"
|
|
assert Path(entry["external_path"]) == ext.resolve()
|
|
assert entry["status"] == "ready"
|
|
assert entry["last_indexed_count"] == 5
|
|
# No KB folder is created under base_dir.
|
|
assert not (manager.base_dir / "Linked").exists()
|
|
|
|
|
|
def test_register_linked_kb_rejects_missing_folder(tmp_path: Path) -> None:
|
|
manager = KnowledgeBaseManager(base_dir=str(tmp_path / "kbs"))
|
|
with pytest.raises(ValueError):
|
|
manager.register_linked_kb("X", str(tmp_path / "nope"), "llamaindex")
|
|
|
|
|
|
def test_register_linked_kb_rejects_name_clash(tmp_path: Path) -> None:
|
|
ext = _external_index(tmp_path)
|
|
manager = KnowledgeBaseManager(base_dir=str(tmp_path / "kbs"))
|
|
manager.register_linked_kb("Dup", str(ext), "llamaindex")
|
|
with pytest.raises(ValueError):
|
|
manager.register_linked_kb("Dup", str(ext), "graphrag")
|
|
|
|
|
|
def test_delete_linked_kb_preserves_external_folder(tmp_path: Path) -> None:
|
|
ext = _external_index(tmp_path)
|
|
manager = KnowledgeBaseManager(base_dir=str(tmp_path / "kbs"))
|
|
manager.register_linked_kb("Linked", str(ext), "llamaindex")
|
|
|
|
assert manager.delete_knowledge_base("Linked", confirm=True) is True
|
|
|
|
# The pointer entry is gone, but the user's external index is untouched.
|
|
assert "Linked" not in manager.list_knowledge_bases()
|
|
assert (ext / "version-1" / "docstore.json").exists()
|
|
|
|
|
|
def test_linked_entry_survives_orphan_prune(tmp_path: Path) -> None:
|
|
ext = _external_index(tmp_path)
|
|
manager = KnowledgeBaseManager(base_dir=str(tmp_path / "kbs"))
|
|
manager.register_linked_kb("Linked", str(ext), "graphrag")
|
|
|
|
# No ``kbs/Linked`` directory exists, yet the pointer must not be pruned.
|
|
assert "Linked" in manager.list_knowledge_bases()
|
|
|
|
|
|
def test_get_metadata_surfaces_external_path_and_provider(tmp_path: Path) -> None:
|
|
ext = _external_index(tmp_path)
|
|
manager = KnowledgeBaseManager(base_dir=str(tmp_path / "kbs"))
|
|
manager.register_linked_kb("Linked", str(ext), "graphrag")
|
|
|
|
meta = manager.get_metadata("Linked")
|
|
assert meta["type"] == "linked"
|
|
assert Path(meta["external_path"]) == ext.resolve()
|
|
assert meta["rag_provider"] == "graphrag"
|
|
|
|
|
|
def test_reconcile_skips_linked_entry(tmp_path: Path) -> None:
|
|
ext = _external_index(tmp_path)
|
|
manager = KnowledgeBaseManager(base_dir=str(tmp_path / "kbs"))
|
|
manager.register_linked_kb("Linked", str(ext), "llamaindex")
|
|
|
|
# A fresh load runs the reconcile path; the linked pointer stays intact and
|
|
# is never flagged for reindex.
|
|
reloaded = KnowledgeBaseManager(base_dir=str(tmp_path / "kbs"))
|
|
entry = reloaded.config["knowledge_bases"]["Linked"]
|
|
assert entry["type"] == "linked"
|
|
assert entry.get("needs_reindex") is not True
|
|
assert "index_versions" not in entry
|