48 lines
1.5 KiB
Python
48 lines
1.5 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
|