test: 补充测试覆盖率 — DB/CLI/Search/Ingest + 修复 chromadb 异常兼容
This commit is contained in:
@@ -61,3 +61,21 @@ class TestCLIJSONOutput:
|
||||
if result.stdout.strip():
|
||||
data = json.loads(result.stdout)
|
||||
assert isinstance(data, dict)
|
||||
|
||||
|
||||
def test_ingest_help():
|
||||
"""ingest --help 正常输出."""
|
||||
result = runner.invoke(app, ["ingest", "--help"])
|
||||
assert result.exit_code == 0
|
||||
|
||||
|
||||
def test_search_help():
|
||||
"""search --help 正常输出."""
|
||||
result = runner.invoke(app, ["search", "--help"])
|
||||
assert result.exit_code == 0
|
||||
|
||||
|
||||
def test_export_help():
|
||||
"""export --help 正常输出."""
|
||||
result = runner.invoke(app, ["export", "--help"])
|
||||
assert result.exit_code == 0
|
||||
|
||||
@@ -45,3 +45,33 @@ class TestVectorDB:
|
||||
# 再次获取会创建新的
|
||||
col = db.get_or_create_collection("tmp_col")
|
||||
assert col.count() == 0
|
||||
|
||||
|
||||
def test_write_guard_context_manager(tmp_path):
|
||||
"""write_guard 上下文管理器正常获取和释放锁."""
|
||||
from src.core.db import VectorDB
|
||||
vdb = VectorDB(persist_dir=str(tmp_path))
|
||||
with vdb.write_guard():
|
||||
pass
|
||||
|
||||
|
||||
def test_close(tmp_path):
|
||||
"""close 正常执行不抛异常."""
|
||||
from src.core.db import VectorDB
|
||||
vdb = VectorDB(persist_dir=str(tmp_path))
|
||||
vdb.close()
|
||||
|
||||
|
||||
def test_delete_collection_nonexistent(tmp_path):
|
||||
"""删除不存在的 collection 不抛异常."""
|
||||
from src.core.db import VectorDB
|
||||
vdb = VectorDB(persist_dir=str(tmp_path))
|
||||
vdb.delete_collection("nonexistent-collection-12345")
|
||||
|
||||
|
||||
def test_delete_by_source_no_match(tmp_path):
|
||||
"""删除不存在的 source 返回 False."""
|
||||
from src.core.db import VectorDB
|
||||
vdb = VectorDB(persist_dir=str(tmp_path))
|
||||
result = vdb.delete_by_source("test_col", "no-such-file.md")
|
||||
assert result is False
|
||||
|
||||
@@ -286,3 +286,38 @@ class TestIncrementalIngest:
|
||||
count2 = ingestor.ingest_file(str(file), incremental=True, force=True)
|
||||
assert count1 > 0
|
||||
assert count2 > 0 # force 模式重新入库
|
||||
|
||||
|
||||
def test_ingest_content_default_splitter(tmp_path):
|
||||
"""未指定 splitter 时用 MarkdownSplitter."""
|
||||
from src.core.config import EmbedConfig
|
||||
from src.core.db import VectorDB
|
||||
from src.core.embedder import create_embedder
|
||||
from src.core.ingest import DocumentIngestor
|
||||
|
||||
db = VectorDB(persist_dir=str(tmp_path / "db"))
|
||||
embedder = create_embedder(EmbedConfig(mode="local"))
|
||||
ingestor = DocumentIngestor(db, embedder, "test_content")
|
||||
count = ingestor.ingest_content("# 测试\n\n一些内容。", "test.md")
|
||||
assert count >= 1
|
||||
|
||||
|
||||
def test_ingest_directory_recursive(tmp_path):
|
||||
"""ingest_directory 递归处理子目录."""
|
||||
from src.core.config import EmbedConfig, ChunkConfig
|
||||
from src.core.db import VectorDB
|
||||
from src.core.embedder import create_embedder
|
||||
from src.core.ingest import DocumentIngestor
|
||||
|
||||
(tmp_path / "sub").mkdir()
|
||||
(tmp_path / "a.md").write_text("# A\n\n内容A。", encoding="utf-8")
|
||||
(tmp_path / "sub" / "b.md").write_text("# B\n\n内容B。", encoding="utf-8")
|
||||
db = VectorDB(persist_dir=str(tmp_path / "db_r"))
|
||||
embedder = create_embedder(EmbedConfig(mode="local"))
|
||||
ingestor = DocumentIngestor(
|
||||
db, embedder, "test_recurse",
|
||||
chunk_config=ChunkConfig(max_size=1000, overlap=100),
|
||||
)
|
||||
results = ingestor.ingest_directory(str(tmp_path))
|
||||
assert len(results) >= 2
|
||||
assert all(v > 0 for v in results.values())
|
||||
|
||||
@@ -122,3 +122,25 @@ class TestSearcher:
|
||||
searcher = Searcher(db, embedder, "empty_coll")
|
||||
sources = searcher.list_sources()
|
||||
assert sources == []
|
||||
|
||||
|
||||
def test_list_sources_empty():
|
||||
"""空 collection 的 list_sources 返回空列表."""
|
||||
from src.core.search import Searcher
|
||||
|
||||
class EmptyColl:
|
||||
def count(self): return 0
|
||||
def get(self, **kwargs):
|
||||
return {"ids": [], "documents": [], "metadatas": []}
|
||||
|
||||
class EmptyDB:
|
||||
def get_or_create_collection(self, name): return EmptyColl()
|
||||
def list_collections(self): return []
|
||||
|
||||
class FakeEmb:
|
||||
@property
|
||||
def dimension(self): return 4
|
||||
def embed(self, texts): return [[0.0] * 4]
|
||||
|
||||
s = Searcher(EmptyDB(), FakeEmb(), "empty")
|
||||
assert s.list_sources() == []
|
||||
|
||||
Reference in New Issue
Block a user