test: add DELETE endpoint tests and ingest integration tests
This commit is contained in:
@@ -110,3 +110,31 @@ class TestSecurity:
|
||||
json={"file_path": "C:\\Windows\\System32\\config\\SAM"},
|
||||
)
|
||||
assert response.status_code in (400, 403)
|
||||
|
||||
|
||||
class TestDeleteEndpoint:
|
||||
"""删除端点."""
|
||||
|
||||
def test_delete_nonexistent(self, client):
|
||||
"""删除不存在的文件返回 404."""
|
||||
response = client.delete("/api/v1/documents/nonexistent.md")
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_delete_ingested(self, client):
|
||||
"""删除已入库文件后搜索不再返回."""
|
||||
# 入库
|
||||
ingest_resp = client.post(
|
||||
"/api/v1/ingest",
|
||||
json={"content": "# Test Delete\nHello.", "file_name": "delete-test.md"},
|
||||
)
|
||||
assert ingest_resp.status_code == 200, f"ingest failed: {ingest_resp.json()}"
|
||||
# 删除
|
||||
response = client.delete("/api/v1/documents/delete-test.md")
|
||||
assert response.status_code == 200, f"delete failed: {response.json()}"
|
||||
# 搜索验证已删除
|
||||
search_resp = client.post(
|
||||
"/api/v1/search", json={"query": "Test Delete", "top_k": 3}
|
||||
)
|
||||
results = search_resp.json()["results"]
|
||||
sources = [r["source_file"] for r in results]
|
||||
assert "delete-test.md" not in sources
|
||||
|
||||
@@ -85,3 +85,37 @@ class TestDocumentIngestor:
|
||||
content = Path(temp_md_dir + "/test.md").read_text(encoding="utf-8")
|
||||
assert "测试" in content
|
||||
assert "这是测试内容" in content
|
||||
|
||||
|
||||
class TestIngestorIntegration:
|
||||
"""入库器集成测试 (使用真实 embedder)."""
|
||||
|
||||
def test_ingest_content_real(self, tmp_path):
|
||||
"""真实入库: 分块→嵌入→入库."""
|
||||
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))
|
||||
embedder = create_embedder(EmbedConfig(mode="local"))
|
||||
ingestor = DocumentIngestor(db, embedder, "test_integration")
|
||||
|
||||
count = ingestor.ingest_content("# Hello\nWorld.", "hello.md")
|
||||
assert count > 0
|
||||
assert ingestor.collection.count() == count
|
||||
|
||||
def test_ingest_deduplicates(self, tmp_path):
|
||||
"""重复入库同一文件会去重."""
|
||||
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))
|
||||
embedder = create_embedder(EmbedConfig(mode="local"))
|
||||
ingestor = DocumentIngestor(db, embedder, "test_dedup")
|
||||
|
||||
c1 = ingestor.ingest_content("# A", "dup.md")
|
||||
c2 = ingestor.ingest_content("# B", "dup.md")
|
||||
assert ingestor.collection.count() == c2
|
||||
|
||||
Reference in New Issue
Block a user