78 lines
2.4 KiB
Python
78 lines
2.4 KiB
Python
"""数据库层测试."""
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from src.core.db import VectorDB
|
|
|
|
|
|
class TestVectorDB:
|
|
"""VectorDB 测试."""
|
|
|
|
@pytest.fixture
|
|
def temp_dir(self):
|
|
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as d:
|
|
yield d
|
|
|
|
def test_init_creates_persist_directory(self, temp_dir):
|
|
"""初始化时自动创建持久化目录."""
|
|
db = VectorDB(persist_dir=temp_dir + "/subdir")
|
|
assert Path(temp_dir + "/subdir").exists()
|
|
assert db.client is not None
|
|
|
|
def test_get_or_create_collection(self, temp_dir):
|
|
"""创建或获取 collection."""
|
|
db = VectorDB(persist_dir=temp_dir)
|
|
col = db.get_or_create_collection("test_col")
|
|
assert col.name == "test_col"
|
|
|
|
# 再次获取应返回同一个
|
|
col2 = db.get_or_create_collection("test_col")
|
|
assert col2.name == "test_col"
|
|
|
|
def test_count_returns_zero_for_empty_collection(self, temp_dir):
|
|
"""空 collection 文档数为 0."""
|
|
db = VectorDB(persist_dir=temp_dir)
|
|
col = db.get_or_create_collection("test_col")
|
|
assert col.count() == 0
|
|
|
|
def test_delete_collection(self, temp_dir):
|
|
"""删除 collection."""
|
|
db = VectorDB(persist_dir=temp_dir)
|
|
db.get_or_create_collection("tmp_col")
|
|
db.delete_collection("tmp_col")
|
|
# 再次获取会创建新的
|
|
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
|