107 lines
2.7 KiB
Python
107 lines
2.7 KiB
Python
"""导出功能测试."""
|
|
import csv
|
|
import io
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from src.core.search import Searcher
|
|
|
|
|
|
class FakeExportCollection:
|
|
def count(self):
|
|
return 2
|
|
|
|
def get(self, include=None):
|
|
return {
|
|
"ids": ["doc_0", "doc_1"],
|
|
"documents": ["内容A。\n\n段落B。", "内容C。"],
|
|
"metadatas": [
|
|
{"source_file": "a.md", "section_title": "标题A", "heading_level": 1, "chunk_index": 0},
|
|
{"source_file": "b.md", "section_title": "", "heading_level": 0, "chunk_index": 0},
|
|
],
|
|
}
|
|
|
|
|
|
class FakeExportDB:
|
|
def get_or_create_collection(self, name):
|
|
return FakeExportCollection()
|
|
|
|
def list_collections(self):
|
|
return []
|
|
|
|
|
|
class FakeExportEmbedder:
|
|
@property
|
|
def dimension(self):
|
|
return 4
|
|
|
|
def embed(self, texts):
|
|
return [[0.1, 0.2, 0.3, 0.4] for _ in texts]
|
|
|
|
|
|
@pytest.fixture
|
|
def searcher():
|
|
db = FakeExportDB()
|
|
embedder = FakeExportEmbedder()
|
|
return Searcher(db, embedder, "test_export")
|
|
|
|
|
|
def test_export_json_returns_valid_json(searcher):
|
|
"""export_json 返回合法的 JSON 字符串."""
|
|
output = searcher.export_json()
|
|
data = json.loads(output)
|
|
assert isinstance(data, list)
|
|
assert len(data) == 2
|
|
|
|
|
|
def test_export_json_contains_all_fields(searcher):
|
|
"""导出包含所有必要字段."""
|
|
output = searcher.export_json()
|
|
data = json.loads(output)
|
|
first = data[0]
|
|
assert "id" in first
|
|
assert "content" in first
|
|
assert "source_file" in first
|
|
assert "section_title" in first
|
|
|
|
|
|
def test_export_csv_returns_valid_csv(searcher):
|
|
"""export_csv 返回合法的 CSV 字符串."""
|
|
output = searcher.export_csv()
|
|
reader = csv.DictReader(io.StringIO(output))
|
|
rows = list(reader)
|
|
assert len(rows) == 2
|
|
|
|
|
|
def test_export_csv_has_header(searcher):
|
|
"""CSV 包含表头."""
|
|
output = searcher.export_csv()
|
|
reader = csv.DictReader(io.StringIO(output))
|
|
assert reader.fieldnames is not None
|
|
assert "content" in reader.fieldnames
|
|
assert "source_file" in reader.fieldnames
|
|
|
|
|
|
def test_export_empty_collection():
|
|
"""空 collection 导出空列表."""
|
|
|
|
class EmptyCollection:
|
|
def count(self):
|
|
return 0
|
|
def get(self, include=None):
|
|
return {"ids": [], "documents": [], "metadatas": []}
|
|
|
|
class EmptyDB:
|
|
def get_or_create_collection(self, name):
|
|
return EmptyCollection()
|
|
def list_collections(self):
|
|
return []
|
|
|
|
s = Searcher(EmptyDB(), FakeExportEmbedder(), "empty")
|
|
json_out = s.export_json()
|
|
assert json.loads(json_out) == []
|
|
csv_out = s.export_csv()
|
|
lines = csv_out.strip().split("\n")
|
|
assert len(lines) == 1 # 仅表头
|