wehub-resource-sync
e4dcfc49aa
Tests / Import Check (Python 3.13) (push) Has been cancelled
Tests / Import Check (Python 3.14) (push) Has been cancelled
Tests / Python Tests (Python 3.11) (push) Has been cancelled
Tests / Python Tests (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.14) (push) Has been cancelled
Tests / Test Summary (push) Has been cancelled
Tests / Lint and Format (push) Has been cancelled
Tests / Web Node Tests (push) Has been cancelled
Tests / Import Check (Python 3.11) (push) Has been cancelled
Tests / Import Check (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.13) (push) Has been cancelled
524 lines
18 KiB
Python
524 lines
18 KiB
Python
"""Unit tests for the LightRAG RAG pipeline + provider routing.
|
|
|
|
RAG-Anything / LightRAG is an optional dependency that is NOT installed in CI,
|
|
so these tests exercise everything that does not require the package (factory
|
|
routing, config bridge, storage, lifecycle gating, parse-layer consumption)
|
|
directly, and stub the thin ``engine`` adapter + the parse service to cover the
|
|
index/search orchestration without the heavy deps.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import types
|
|
|
|
import pytest
|
|
|
|
from deeptutor.services.rag.factory import (
|
|
LIGHTRAG_PROVIDER,
|
|
get_pipeline,
|
|
list_pipelines,
|
|
normalize_provider_name,
|
|
)
|
|
from deeptutor.services.rag.index_versioning import resolve_storage_dir_for_read
|
|
from deeptutor.services.rag.pipelines.lightrag import config as lr_config
|
|
from deeptutor.services.rag.pipelines.lightrag import engine, storage
|
|
from deeptutor.services.rag.pipelines.lightrag.pipeline import LightRagPipeline
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# factory routing + config
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_factory_dispatches_lightrag_lazily(tmp_path) -> None:
|
|
pipe = get_pipeline("lightrag", kb_base_dir=str(tmp_path))
|
|
assert type(pipe).__name__ == "LightRagPipeline"
|
|
# Building the pipeline must NOT import the heavy optional dependency.
|
|
assert "raganything" not in sys.modules
|
|
|
|
|
|
def test_list_pipelines_includes_lightrag(monkeypatch) -> None:
|
|
monkeypatch.setattr(lr_config, "is_lightrag_available", lambda: False)
|
|
entry = next(p for p in list_pipelines() if p["id"] == LIGHTRAG_PROVIDER)
|
|
assert entry["requires_api_key"] is False
|
|
assert entry["configured"] is False
|
|
|
|
|
|
def test_normalize_provider_keeps_lightrag() -> None:
|
|
assert normalize_provider_name("lightrag") == "lightrag"
|
|
assert normalize_provider_name("LightRAG") == "lightrag"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"given,expected",
|
|
[
|
|
("hybrid", "hybrid"),
|
|
("MIX", "mix"),
|
|
("naive", "naive"),
|
|
("local", "local"),
|
|
("global", "global"),
|
|
("", "hybrid"),
|
|
(None, "hybrid"),
|
|
("bogus", "hybrid"),
|
|
],
|
|
)
|
|
def test_normalize_mode(given, expected) -> None:
|
|
assert lr_config.normalize_mode(given) == expected
|
|
|
|
|
|
def test_is_lightrag_available_false_when_dependency_missing(monkeypatch) -> None:
|
|
def fake_find_spec(name):
|
|
return None if name == "raganything" else object()
|
|
|
|
monkeypatch.setattr(lr_config.importlib.util, "find_spec", fake_find_spec)
|
|
assert lr_config.is_lightrag_available() is False
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# storage
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_storage_meta_and_has_output(tmp_path) -> None:
|
|
root = tmp_path / "version-1"
|
|
root.mkdir()
|
|
assert storage.has_output(root) is False
|
|
assert storage.has_output(None) is False
|
|
|
|
(root / "vdb_chunks.json").write_text("{}", encoding="utf-8")
|
|
assert storage.has_output(root) is False
|
|
|
|
(root / "graph_chunk_entity_relation.graphml").write_text("<graph/>", encoding="utf-8")
|
|
assert storage.has_output(root) is False
|
|
|
|
(root / "kv_store_doc_status.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"doc-1": {
|
|
"status": "failed",
|
|
"file_path": "bad.docx",
|
|
"error_msg": "embedding failed",
|
|
"chunks_list": [],
|
|
}
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
assert storage.has_output(root) is False
|
|
assert storage.failure_summary(root) == "bad.docx: embedding failed"
|
|
assert storage.document_error(root, "doc-1") == "embedding failed"
|
|
|
|
(root / "kv_store_doc_status.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"doc-1": {
|
|
"status": "processed",
|
|
"file_path": "good.docx",
|
|
"chunks_list": ["chunk-1"],
|
|
}
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
assert storage.has_output(root) is True
|
|
|
|
storage.write_meta(root)
|
|
meta = json.loads((root / storage.META_FILENAME).read_text())
|
|
assert meta["signature"] == "lightrag"
|
|
assert meta["provider"] == "lightrag"
|
|
|
|
|
|
def test_embedding_func_returns_numpy_array(monkeypatch) -> None:
|
|
class _FakeEmbeddingFunc:
|
|
def __init__(self, *, embedding_dim, max_token_size, func) -> None:
|
|
self.embedding_dim = embedding_dim
|
|
self.max_token_size = max_token_size
|
|
self.func = func
|
|
|
|
fake_lightrag = types.ModuleType("lightrag")
|
|
fake_utils = types.ModuleType("lightrag.utils")
|
|
fake_utils.EmbeddingFunc = _FakeEmbeddingFunc
|
|
monkeypatch.setitem(sys.modules, "lightrag", fake_lightrag)
|
|
monkeypatch.setitem(sys.modules, "lightrag.utils", fake_utils)
|
|
|
|
class _Config:
|
|
dim = 3
|
|
max_tokens = 99
|
|
|
|
class _Client:
|
|
def get_embedding_func(self):
|
|
async def embed(texts):
|
|
return [[1, 2, 3] for _ in texts]
|
|
|
|
return embed
|
|
|
|
monkeypatch.setattr("deeptutor.services.embedding.get_embedding_config", lambda: _Config())
|
|
monkeypatch.setattr("deeptutor.services.embedding.get_embedding_client", lambda: _Client())
|
|
|
|
embedding = lr_config.build_embedding_func()
|
|
vectors = asyncio.run(embedding.func(["a", "b"]))
|
|
assert embedding.embedding_dim == 3
|
|
assert embedding.max_token_size == 99
|
|
assert vectors.shape == (2, 3)
|
|
assert hasattr(vectors, "size")
|
|
|
|
|
|
def test_lightrag_llm_adapter_preserves_messages_and_drops_extra_kwargs(
|
|
monkeypatch,
|
|
) -> None:
|
|
captured: dict[str, object] = {}
|
|
|
|
class _Client:
|
|
def get_model_func(self):
|
|
async def model_func(prompt, **kwargs):
|
|
captured["prompt"] = prompt
|
|
captured.update(kwargs)
|
|
return "ok"
|
|
|
|
return model_func
|
|
|
|
monkeypatch.setattr("deeptutor.services.llm.get_llm_client", lambda: _Client())
|
|
|
|
func = lr_config.build_llm_model_func()
|
|
result = asyncio.run(
|
|
func(
|
|
"",
|
|
system_prompt="sys",
|
|
messages=[{"role": "user", "content": "from messages"}],
|
|
response_format={"type": "json_object"},
|
|
hashing_kv=object(),
|
|
keyword_extraction=True,
|
|
)
|
|
)
|
|
|
|
assert result == "ok"
|
|
assert captured["prompt"] == ""
|
|
assert captured["system_prompt"] == "sys"
|
|
assert captured["history_messages"] == []
|
|
assert captured["messages"] == [{"role": "user", "content": "from messages"}]
|
|
assert "response_format" not in captured
|
|
assert "hashing_kv" not in captured
|
|
assert "keyword_extraction" not in captured
|
|
|
|
|
|
def test_lightrag_vision_adapter_preserves_messages(monkeypatch) -> None:
|
|
captured: dict[str, object] = {}
|
|
|
|
class _Client:
|
|
def get_vision_model_func(self):
|
|
async def model_func(prompt, **kwargs):
|
|
captured["prompt"] = prompt
|
|
captured.update(kwargs)
|
|
return "ok"
|
|
|
|
return model_func
|
|
|
|
monkeypatch.setattr("deeptutor.services.llm.get_llm_client", lambda: _Client())
|
|
|
|
func = lr_config.build_vision_model_func()
|
|
result = asyncio.run(
|
|
func(
|
|
"",
|
|
image_data="abc123",
|
|
messages=[{"role": "user", "content": [{"type": "text", "text": "hi"}]}],
|
|
)
|
|
)
|
|
|
|
assert result == "ok"
|
|
assert captured["prompt"] == ""
|
|
assert captured["image_data"] == "abc123"
|
|
assert captured["messages"] == [{"role": "user", "content": [{"type": "text", "text": "hi"}]}]
|
|
|
|
|
|
def test_build_rag_skips_raganything_parser_install_check(monkeypatch) -> None:
|
|
"""Regression for issue #594.
|
|
|
|
RAG-Anything validates its *default* parser (``mineru``) at LightRAG-init
|
|
time, even though DeepTutor only ever inserts a pre-parsed ``content_list``
|
|
and never uses RAG-Anything's parser. ``build_rag`` must pre-satisfy that
|
|
check so indexing with a different parse engine (e.g. pymupdf4llm) doesn't
|
|
hard-fail when MinerU is absent.
|
|
"""
|
|
captured: dict[str, object] = {}
|
|
|
|
class _FakeConfig:
|
|
def __init__(self, *, working_dir) -> None:
|
|
self.working_dir = working_dir
|
|
self.parser = "mineru" # RAG-Anything's default
|
|
|
|
class _FakeRagAnything:
|
|
def __init__(self, *, config, llm_model_func, vision_model_func, embedding_func) -> None:
|
|
# Mirror the real constructor: the install check starts unsatisfied.
|
|
self._parser_installation_checked = False
|
|
captured["config"] = config
|
|
|
|
fake_module = types.ModuleType("raganything")
|
|
fake_module.RAGAnything = _FakeRagAnything
|
|
fake_module.RAGAnythingConfig = _FakeConfig
|
|
monkeypatch.setitem(sys.modules, "raganything", fake_module)
|
|
monkeypatch.setattr(engine, "build_llm_model_func", lambda: "llm")
|
|
monkeypatch.setattr(engine, "build_vision_model_func", lambda: "vision")
|
|
monkeypatch.setattr(engine, "build_embedding_func", lambda: "embed")
|
|
|
|
rag = engine.build_rag(Path("/tmp/kb-wd")) # noqa: S108
|
|
|
|
assert rag._parser_installation_checked is True
|
|
assert captured["config"].working_dir == "/tmp/kb-wd"
|
|
|
|
|
|
def test_lightrag_query_initializes_raganything_before_aquery(monkeypatch) -> None:
|
|
calls: list[str] = []
|
|
|
|
class _Rag:
|
|
lightrag = None
|
|
|
|
async def _ensure_lightrag_initialized(self):
|
|
calls.append("ensure")
|
|
self.lightrag = object()
|
|
return {"success": True}
|
|
|
|
async def aquery(self, question, mode=None, **kwargs):
|
|
calls.append("aquery")
|
|
assert self.lightrag is not None
|
|
assert question == "hello"
|
|
assert mode == "hybrid"
|
|
assert kwargs == {}
|
|
return "answer"
|
|
|
|
monkeypatch.setattr(engine, "query_kwargs_from_settings", lambda: {})
|
|
|
|
result = asyncio.run(engine.query(_Rag(), "hello", "hybrid"))
|
|
|
|
assert result == "answer"
|
|
assert calls == ["ensure", "aquery"]
|
|
|
|
|
|
def test_lightrag_query_surfaces_raganything_initialization_failure() -> None:
|
|
class _Rag:
|
|
lightrag = None
|
|
|
|
async def _ensure_lightrag_initialized(self):
|
|
return {"success": False, "error": "storage failed"}
|
|
|
|
async def aquery(self, question, mode=None, **kwargs): # pragma: no cover
|
|
raise AssertionError("aquery should not run")
|
|
|
|
with pytest.raises(RuntimeError, match="storage failed"):
|
|
asyncio.run(engine.query(_Rag(), "hello", "hybrid"))
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# pipeline lifecycle (engine + parse service stubbed)
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class _FakeRag:
|
|
def __init__(self, working_dir) -> None:
|
|
self.working_dir = Path(working_dir)
|
|
|
|
|
|
def _force_available(monkeypatch, available: bool = True) -> None:
|
|
monkeypatch.setattr(lr_config, "is_lightrag_available", lambda: available)
|
|
|
|
|
|
def _stub_engine(monkeypatch, answer: str = "ANSWER") -> list[dict]:
|
|
"""Stub the engine so insert writes a readiness marker and query echoes."""
|
|
inserts: list[dict] = []
|
|
monkeypatch.setattr(engine, "build_rag", lambda wd: _FakeRag(wd))
|
|
|
|
async def fake_insert(rag, content_list, *, file_name, doc_id):
|
|
inserts.append({"file": file_name, "doc_id": doc_id, "blocks": content_list})
|
|
(rag.working_dir / "vdb_chunks.json").write_text(
|
|
json.dumps({"vectors": [[1.0]]}), encoding="utf-8"
|
|
)
|
|
(rag.working_dir / "kv_store_doc_status.json").write_text(
|
|
json.dumps(
|
|
{
|
|
doc_id: {
|
|
"status": "processed",
|
|
"file_path": file_name,
|
|
"chunks_list": ["chunk-1"],
|
|
}
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
async def fake_query(rag, question, mode):
|
|
return f"{answer}|{mode}"
|
|
|
|
monkeypatch.setattr(engine, "insert", fake_insert)
|
|
monkeypatch.setattr(engine, "query", fake_query)
|
|
return inserts
|
|
|
|
|
|
def _stub_parse(monkeypatch, *, blocks=None, markdown: str = "# md") -> None:
|
|
from deeptutor.services.parsing.types import ParsedDocument
|
|
|
|
class _Service:
|
|
def parse(self, path, **_):
|
|
return ParsedDocument(
|
|
markdown=markdown,
|
|
blocks=blocks,
|
|
source_hash="h_" + Path(path).stem,
|
|
engine="fake",
|
|
)
|
|
|
|
monkeypatch.setattr("deeptutor.services.parsing.get_parse_service", lambda: _Service())
|
|
|
|
|
|
def test_initialize_requires_lightrag(tmp_path, monkeypatch) -> None:
|
|
_force_available(monkeypatch, False)
|
|
pipe = LightRagPipeline(kb_base_dir=str(tmp_path))
|
|
pdf = tmp_path / "a.pdf"
|
|
pdf.write_bytes(b"%PDF")
|
|
with pytest.raises(lr_config.LightRagNotAvailableError):
|
|
asyncio.run(pipe.initialize("kb", [str(pdf)]))
|
|
|
|
|
|
def test_initialize_orchestrates_index_and_uses_blocks(tmp_path, monkeypatch) -> None:
|
|
_force_available(monkeypatch, True)
|
|
inserts = _stub_engine(monkeypatch)
|
|
_stub_parse(monkeypatch, blocks=[{"type": "text", "text": "hi", "page_idx": 0}])
|
|
pipe = LightRagPipeline(kb_base_dir=str(tmp_path))
|
|
pdf = tmp_path / "exam.pdf"
|
|
pdf.write_bytes(b"%PDF")
|
|
|
|
ok = asyncio.run(pipe.initialize("kb", [str(pdf)]))
|
|
assert ok is True
|
|
assert len(inserts) == 1
|
|
assert inserts[0]["file"] == "exam.pdf"
|
|
# blocks from the parse layer are passed through verbatim (multimodal path).
|
|
assert inserts[0]["blocks"] == [{"type": "text", "text": "hi", "page_idx": 0}]
|
|
# version dir is marked ready.
|
|
root = resolve_storage_dir_for_read(tmp_path / "kb", None)
|
|
assert storage.has_output(root) is True
|
|
|
|
|
|
def test_ingest_falls_back_to_markdown_when_no_blocks(tmp_path, monkeypatch) -> None:
|
|
_force_available(monkeypatch, True)
|
|
inserts = _stub_engine(monkeypatch)
|
|
_stub_parse(monkeypatch, blocks=None, markdown="# only markdown")
|
|
pipe = LightRagPipeline(kb_base_dir=str(tmp_path))
|
|
pdf = tmp_path / "notes.pdf"
|
|
pdf.write_bytes(b"%PDF")
|
|
|
|
asyncio.run(pipe.initialize("kb", [str(pdf)]))
|
|
assert inserts[0]["blocks"] == [{"type": "text", "text": "# only markdown", "page_idx": 0}]
|
|
|
|
|
|
def test_initialize_no_content_returns_false(tmp_path, monkeypatch) -> None:
|
|
_force_available(monkeypatch, True)
|
|
inserts = _stub_engine(monkeypatch)
|
|
_stub_parse(monkeypatch, blocks=None, markdown="") # empty parse
|
|
pipe = LightRagPipeline(kb_base_dir=str(tmp_path))
|
|
pdf = tmp_path / "blank.pdf"
|
|
pdf.write_bytes(b"%PDF")
|
|
|
|
ok = asyncio.run(pipe.initialize("kb", [str(pdf)]))
|
|
assert ok is False
|
|
assert inserts == []
|
|
|
|
|
|
def test_initialize_fails_when_lightrag_records_doc_failure(tmp_path, monkeypatch) -> None:
|
|
_force_available(monkeypatch, True)
|
|
monkeypatch.setattr(engine, "build_rag", lambda wd: _FakeRag(wd))
|
|
|
|
async def fake_insert(rag, content_list, *, file_name, doc_id):
|
|
(rag.working_dir / "kv_store_doc_status.json").write_text(
|
|
json.dumps(
|
|
{
|
|
doc_id: {
|
|
"status": "failed",
|
|
"file_path": file_name,
|
|
"error_msg": "'list' object has no attribute 'size'",
|
|
"chunks_list": [],
|
|
}
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
monkeypatch.setattr(engine, "insert", fake_insert)
|
|
_stub_parse(monkeypatch, blocks=[{"type": "text", "text": "hi", "page_idx": 0}])
|
|
pipe = LightRagPipeline(kb_base_dir=str(tmp_path))
|
|
docx = tmp_path / "bad.docx"
|
|
docx.write_bytes(b"docx")
|
|
|
|
with pytest.raises(RuntimeError, match="list.*size"):
|
|
asyncio.run(pipe.initialize("kb", [str(docx)]))
|
|
|
|
assert resolve_storage_dir_for_read(tmp_path / "kb", None) is None
|
|
|
|
|
|
def test_search_needs_reindex_without_output(tmp_path) -> None:
|
|
res = asyncio.run(LightRagPipeline(kb_base_dir=str(tmp_path)).search("q", "missing"))
|
|
assert res["needs_reindex"] is True
|
|
assert res["provider"] == "lightrag"
|
|
|
|
|
|
def test_search_not_configured_when_unavailable(tmp_path, monkeypatch) -> None:
|
|
_force_available(monkeypatch, True)
|
|
_stub_engine(monkeypatch)
|
|
_stub_parse(monkeypatch, blocks=[{"type": "text", "text": "x", "page_idx": 0}])
|
|
pipe = LightRagPipeline(kb_base_dir=str(tmp_path))
|
|
pdf = tmp_path / "a.pdf"
|
|
pdf.write_bytes(b"%PDF")
|
|
asyncio.run(pipe.initialize("kb", [str(pdf)]))
|
|
|
|
_force_available(monkeypatch, False)
|
|
res = asyncio.run(pipe.search("q", "kb"))
|
|
assert res["error_type"] == "not_configured"
|
|
|
|
|
|
def test_search_happy_path_resolves_mode(tmp_path, monkeypatch) -> None:
|
|
_force_available(monkeypatch, True)
|
|
_stub_engine(monkeypatch, answer="GROUNDED")
|
|
_stub_parse(monkeypatch, blocks=[{"type": "text", "text": "x", "page_idx": 0}])
|
|
pipe = LightRagPipeline(kb_base_dir=str(tmp_path))
|
|
pdf = tmp_path / "a.pdf"
|
|
pdf.write_bytes(b"%PDF")
|
|
asyncio.run(pipe.initialize("kb", [str(pdf)]))
|
|
|
|
# Per-KB search_mode is read from kb_config.json next to the store.
|
|
(tmp_path / "kb_config.json").write_text(
|
|
json.dumps({"knowledge_bases": {"kb": {"search_mode": "local"}}}), encoding="utf-8"
|
|
)
|
|
res = asyncio.run(pipe.search("question?", "kb"))
|
|
assert res["answer"] == "GROUNDED|local"
|
|
assert res["mode"] == "local"
|
|
assert res["provider"] == "lightrag"
|
|
|
|
|
|
def test_explicit_mode_overrides_kb_config(tmp_path, monkeypatch) -> None:
|
|
_force_available(monkeypatch, True)
|
|
_stub_engine(monkeypatch, answer="A")
|
|
_stub_parse(monkeypatch, blocks=[{"type": "text", "text": "x", "page_idx": 0}])
|
|
pipe = LightRagPipeline(kb_base_dir=str(tmp_path))
|
|
pdf = tmp_path / "a.pdf"
|
|
pdf.write_bytes(b"%PDF")
|
|
asyncio.run(pipe.initialize("kb", [str(pdf)]))
|
|
|
|
res = asyncio.run(pipe.search("q", "kb", mode="global"))
|
|
assert res["mode"] == "global"
|
|
|
|
|
|
def test_global_provider_mode_used_when_kb_has_none(tmp_path, monkeypatch) -> None:
|
|
_force_available(monkeypatch, True)
|
|
_stub_engine(monkeypatch, answer="A")
|
|
_stub_parse(monkeypatch, blocks=[{"type": "text", "text": "x", "page_idx": 0}])
|
|
pipe = LightRagPipeline(kb_base_dir=str(tmp_path))
|
|
pdf = tmp_path / "a.pdf"
|
|
pdf.write_bytes(b"%PDF")
|
|
asyncio.run(pipe.initialize("kb", [str(pdf)]))
|
|
|
|
# No per-KB search_mode, but a global default mode set from the engine card.
|
|
(tmp_path / "kb_config.json").write_text(
|
|
json.dumps({"defaults": {"provider_modes": {"lightrag": "naive"}}}), encoding="utf-8"
|
|
)
|
|
res = asyncio.run(pipe.search("q", "kb"))
|
|
assert res["mode"] == "naive"
|